zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for resilient PDS record fetching with DID resolution fallback."""
2
3from unittest.mock import AsyncMock, MagicMock, patch
4
5import pytest
6
7from backend._internal.atproto.records.fm_plyr.track import (
8 get_record_public_resilient,
9)
10
11RECORD_URI = "at://did:plc:abc123/fm.plyr.track/rkey1"
12STALE_PDS = "https://shiitake.us-east.host.bsky.network"
13CORRECT_PDS = "https://eurosky.social"
14RECORD_DATA = {"uri": RECORD_URI, "value": {"title": "test track"}}
15
16
17async def test_resilient_retries_with_resolved_pds() -> None:
18 """when the cached PDS URL fails, resolve the DID and retry."""
19 mock_atproto_data = MagicMock(pds=CORRECT_PDS)
20
21 with (
22 patch(
23 "backend._internal.atproto.records.fm_plyr.track.get_record_public",
24 new_callable=AsyncMock,
25 side_effect=[Exception("404 not found"), RECORD_DATA],
26 ) as mock_get,
27 patch(
28 "backend._internal.atproto.records.fm_plyr.track.AsyncDidResolver"
29 ) as mock_resolver_cls,
30 ):
31 mock_resolver = MagicMock()
32 mock_resolver.resolve_atproto_data = AsyncMock(return_value=mock_atproto_data)
33 mock_resolver_cls.return_value = mock_resolver
34
35 data, resolved_url = await get_record_public_resilient(RECORD_URI, STALE_PDS)
36
37 assert data == RECORD_DATA
38 assert resolved_url == CORRECT_PDS
39 # first call with stale PDS, second with resolved
40 assert mock_get.call_count == 2
41 mock_get.assert_any_call(RECORD_URI, STALE_PDS)
42 mock_get.assert_any_call(RECORD_URI, CORRECT_PDS)
43
44
45async def test_resilient_reraises_when_resolved_url_same() -> None:
46 """when DID resolution returns the same PDS URL, re-raise original error."""
47 mock_atproto_data = MagicMock(pds=STALE_PDS)
48
49 with (
50 patch(
51 "backend._internal.atproto.records.fm_plyr.track.get_record_public",
52 new_callable=AsyncMock,
53 side_effect=Exception("404 not found"),
54 ),
55 patch(
56 "backend._internal.atproto.records.fm_plyr.track.AsyncDidResolver"
57 ) as mock_resolver_cls,
58 ):
59 mock_resolver = MagicMock()
60 mock_resolver.resolve_atproto_data = AsyncMock(return_value=mock_atproto_data)
61 mock_resolver_cls.return_value = mock_resolver
62
63 with pytest.raises(Exception, match="404 not found"):
64 await get_record_public_resilient(RECORD_URI, STALE_PDS)
65
66
67async def test_resilient_reraises_when_no_pds_url_provided() -> None:
68 """when no pds_url is provided, don't attempt DID resolution."""
69 with (
70 patch(
71 "backend._internal.atproto.records.fm_plyr.track.get_record_public",
72 new_callable=AsyncMock,
73 side_effect=Exception("fetch failed"),
74 ),
75 pytest.raises(Exception, match="fetch failed"),
76 ):
77 await get_record_public_resilient(RECORD_URI, pds_url=None)
78
79
80async def test_resilient_returns_none_when_first_try_succeeds() -> None:
81 """when the initial fetch succeeds, resolved_pds_url is None."""
82 with patch(
83 "backend._internal.atproto.records.fm_plyr.track.get_record_public",
84 new_callable=AsyncMock,
85 return_value=RECORD_DATA,
86 ):
87 data, resolved_url = await get_record_public_resilient(RECORD_URI, STALE_PDS)
88
89 assert data == RECORD_DATA
90 assert resolved_url is None
91
92
93async def test_resilient_reraises_when_did_resolution_fails() -> None:
94 """when DID resolution itself fails, re-raise the original PDS error."""
95 with (
96 patch(
97 "backend._internal.atproto.records.fm_plyr.track.get_record_public",
98 new_callable=AsyncMock,
99 side_effect=Exception("pds unreachable"),
100 ),
101 patch(
102 "backend._internal.atproto.records.fm_plyr.track.AsyncDidResolver"
103 ) as mock_resolver_cls,
104 ):
105 mock_resolver = MagicMock()
106 mock_resolver.resolve_atproto_data = AsyncMock(
107 side_effect=Exception("DID resolution failed")
108 )
109 mock_resolver_cls.return_value = mock_resolver
110
111 with pytest.raises(Exception, match="pds unreachable"):
112 await get_record_public_resilient(RECORD_URI, STALE_PDS)