zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for transcoder client."""
2
3import tempfile
4from pathlib import Path
5from unittest.mock import AsyncMock, Mock, patch
6
7import httpx
8import pytest
9
10from backend._internal.clients.transcoder import (
11 TranscoderClient,
12 get_transcoder_client,
13)
14
15
16@pytest.fixture
17def transcoder_client() -> TranscoderClient:
18 """test transcoder client with mock settings."""
19 return TranscoderClient(
20 service_url="https://test-transcoder.example.com",
21 auth_token="test-token",
22 timeout_seconds=30,
23 target_format="mp3",
24 )
25
26
27@pytest.fixture
28def temp_audio_file() -> Path:
29 """create a temporary audio file for testing."""
30 with tempfile.NamedTemporaryFile(mode="wb", suffix=".aiff", delete=False) as f:
31 # write some fake audio data
32 f.write(b"FORM" + b"\x00" * 100)
33 return Path(f.name)
34
35
36async def test_transcoder_client_success(
37 transcoder_client: TranscoderClient, temp_audio_file: Path
38) -> None:
39 """test TranscoderClient.transcode_file() with successful response."""
40 mock_response = Mock()
41 mock_response.content = b"transcoded mp3 data"
42 mock_response.headers = {"content-type": "audio/mpeg"}
43 mock_response.raise_for_status.return_value = None
44
45 with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
46 mock_post.return_value = mock_response
47
48 result = await transcoder_client.transcode_file(temp_audio_file, "aiff")
49
50 assert result.success is True
51 assert result.data == b"transcoded mp3 data"
52 assert result.content_type == "audio/mpeg"
53 assert result.error is None
54 mock_post.assert_called_once()
55
56 # verify the call included auth header
57 call_kwargs = mock_post.call_args.kwargs
58 assert call_kwargs["headers"] == {"X-Transcoder-Key": "test-token"}
59 assert call_kwargs["params"] == {"target": "mp3"}
60
61 # cleanup
62 temp_audio_file.unlink(missing_ok=True)
63
64
65async def test_transcoder_client_timeout(
66 transcoder_client: TranscoderClient, temp_audio_file: Path
67) -> None:
68 """test TranscoderClient.transcode_file() timeout handling."""
69 with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
70 mock_post.side_effect = httpx.TimeoutException("timeout")
71
72 result = await transcoder_client.transcode_file(temp_audio_file, "aiff")
73
74 assert result.success is False
75 assert result.data is None
76 assert result.error is not None
77 assert "timed out" in result.error
78
79 # cleanup
80 temp_audio_file.unlink(missing_ok=True)
81
82
83async def test_transcoder_client_http_error(
84 transcoder_client: TranscoderClient, temp_audio_file: Path
85) -> None:
86 """test TranscoderClient.transcode_file() HTTP error handling."""
87 mock_response = Mock()
88 mock_response.status_code = 500
89 mock_response.text = "Internal Server Error"
90
91 with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
92 mock_post.side_effect = httpx.HTTPStatusError(
93 "Server Error",
94 request=Mock(),
95 response=mock_response,
96 )
97
98 result = await transcoder_client.transcode_file(temp_audio_file, "aiff")
99
100 assert result.success is False
101 assert result.data is None
102 assert result.error is not None
103 assert "500" in result.error
104 assert "Internal Server Error" in result.error
105
106 # cleanup
107 temp_audio_file.unlink(missing_ok=True)
108
109
110async def test_transcoder_client_unexpected_error(
111 transcoder_client: TranscoderClient, temp_audio_file: Path
112) -> None:
113 """test TranscoderClient.transcode_file() unexpected error handling."""
114 with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
115 mock_post.side_effect = RuntimeError("unexpected error")
116
117 result = await transcoder_client.transcode_file(temp_audio_file, "aiff")
118
119 assert result.success is False
120 assert result.data is None
121 assert result.error is not None
122 assert "unexpected" in result.error.lower()
123
124 # cleanup
125 temp_audio_file.unlink(missing_ok=True)
126
127
128async def test_transcoder_client_custom_target_format(
129 transcoder_client: TranscoderClient, temp_audio_file: Path
130) -> None:
131 """test TranscoderClient.transcode_file() with custom target format."""
132 mock_response = Mock()
133 mock_response.content = b"transcoded aac data"
134 mock_response.headers = {"content-type": "audio/aac"}
135 mock_response.raise_for_status.return_value = None
136
137 with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
138 mock_post.return_value = mock_response
139
140 result = await transcoder_client.transcode_file(
141 temp_audio_file, "flac", target_format="aac"
142 )
143
144 assert result.success is True
145 call_kwargs = mock_post.call_args.kwargs
146 assert call_kwargs["params"] == {"target": "aac"}
147
148 # cleanup
149 temp_audio_file.unlink(missing_ok=True)
150
151
152def test_transcoder_client_from_settings() -> None:
153 """test TranscoderClient.from_settings() creates client correctly."""
154 with patch("backend._internal.clients.transcoder.settings") as mock_settings:
155 mock_settings.transcoder.service_url = "https://transcoder.example.com"
156 mock_settings.transcoder.auth_token = "secret-token"
157 mock_settings.transcoder.timeout_seconds = 120
158 mock_settings.transcoder.target_format = "mp3"
159
160 client = TranscoderClient.from_settings()
161
162 assert client.service_url == "https://transcoder.example.com"
163 assert client.auth_token == "secret-token"
164 assert client.target_format == "mp3"
165
166
167def test_get_transcoder_client_singleton() -> None:
168 """test get_transcoder_client() returns singleton."""
169 import backend._internal.clients.transcoder as module
170
171 # reset singleton
172 module._client = None
173
174 with patch("backend._internal.clients.transcoder.settings") as mock_settings:
175 mock_settings.transcoder.service_url = "https://transcoder.example.com"
176 mock_settings.transcoder.auth_token = "token"
177 mock_settings.transcoder.timeout_seconds = 60
178 mock_settings.transcoder.target_format = "mp3"
179
180 client1 = get_transcoder_client()
181 client2 = get_transcoder_client()
182
183 assert client1 is client2
184
185 # cleanup
186 module._client = None