From 706fed9c08e8fcb5f611a130b36080287c513fc6 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sat, 26 Sep 2026 13:03:47 +0000 Subject: [PATCH] style: auto-format with black + isort + ruff + prettier [skip ci-format-check] --- tests/unit/test_gpu_encoder.py | 263 ++++++++++++++++++++++----------- tests/unit/test_gpu_relay.py | 28 +++- 2 files changed, 196 insertions(+), 95 deletions(-) diff --git a/tests/unit/test_gpu_encoder.py b/tests/unit/test_gpu_encoder.py index 8c11fa609..17d9f4451 100644 --- a/tests/unit/test_gpu_encoder.py +++ b/tests/unit/test_gpu_encoder.py @@ -1,4 +1,5 @@ """GpuEncoderClient 单元测试：mock HTTP，覆盖 health/sync/fallback/singleton 等完整路径。""" + from __future__ import annotations import json @@ -62,8 +63,12 @@ def _fake_response(status: int = 200, body: dict | bytes | None = None, headers= # ── Health check ──────────────────────────────────────────────────── class TestHealthCheck: def test_healthy_nvenc_available(self, client): - body = {"status": "healthy", "worker": "w1", - "gpu": {"name": "Quadro P4000"}, "nvenc": {"h264_nvenc": True, "hevc_nvenc": True}} + body = { + "status": "healthy", + "worker": "w1", + "gpu": {"name": "Quadro P4000"}, + "nvenc": {"h264_nvenc": True, "hevc_nvenc": True}, + } with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=body)): h = client.check_health() assert h.healthy and h.nvenc_h264 and h.ready @@ -98,55 +103,73 @@ class TestHealthCheck: # ── _post_sync ────────────────────────────────────────────────────── class TestPostSync: def test_completed_job_returns_dict(self, client): - result_body = {"job_id":"j1","status":"completed","ffmpeg_rc":0,"uploaded":True,"duration":5.1,"size":123456} + result_body = { + "job_id": "j1", + "status": "completed", + "ffmpeg_rc": 0, + "uploaded": True, + "duration": 5.1, + "size": 123456, + } with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=result_body)) as m: res = client._post_sync( - {"inputs":{"in.mp4":"http://x"},"ffmpeg_args":["-i","in.mp4"], - "output_url":"http://relay/k?token=s","timeout":30}, - mezzanine_path=Path("/tmp/fake.mp4")) + { + "inputs": {"in.mp4": "http://x"}, + "ffmpeg_args": ["-i", "in.mp4"], + "output_url": "http://relay/k?token=s", + "timeout": 30, + }, + mezzanine_path=Path("/tmp/fake.mp4"), + ) assert res["status"] == "completed" and res["ffmpeg_rc"] == 0 req = m.call_args[0][0] assert req.full_url == "http://gpu.example.com:8900/api/render/sync" def test_ffmpeg_failure_raises(self, client): - body = {"status":"failed","ffmpeg_rc":1,"message":"Invalid data"} + body = {"status": "failed", "ffmpeg_rc": 1, "message": "Invalid data"} with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=body)): with pytest.raises(GpuEncodeError, match="rc=1"): - client._post_sync({"inputs":{},"ffmpeg_args":[],"output_url":"","timeout":10}, - mezzanine_path=Path("/tmp/x")) + client._post_sync( + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10}, mezzanine_path=Path("/tmp/x") + ) def test_http_4xx_raises(self, client): - err = urllib.error.HTTPError(url="http://gpu/render/sync", code=422, msg="Unprocessable", - hdrs={}, fp=BytesIO(b"bad request")) + err = urllib.error.HTTPError( + url="http://gpu/render/sync", code=422, msg="Unprocessable", hdrs={}, fp=BytesIO(b"bad request") + ) with mock.patch("urllib.request.urlopen", side_effect=err): with pytest.raises(GpuEncodeError, match="HTTP 422"): - client._post_sync({"inputs":{},"ffmpeg_args":[],"output_url":"","timeout":10}, - mezzanine_path=Path("/tmp/x")) + client._post_sync( + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10}, mezzanine_path=Path("/tmp/x") + ) def test_connection_error_raises(self, client): with mock.patch("urllib.request.urlopen", side_effect=urllib.error.URLError("conn refused")): with pytest.raises(GpuEncodeError, match="connection error"): - client._post_sync({"inputs":{},"ffmpeg_args":[],"output_url":"","timeout":10}, - mezzanine_path=Path("/tmp/x")) + client._post_sync( + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10}, mezzanine_path=Path("/tmp/x") + ) def test_timeout_error_raises(self, client): with mock.patch("urllib.request.urlopen", side_effect=socket.timeout("timed out")): with pytest.raises(GpuEncodeError, match="connection error"): - client._post_sync({"inputs":{},"ffmpeg_args":[],"output_url":"","timeout":10}, - mezzanine_path=Path("/tmp/x")) + client._post_sync( + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10}, mezzanine_path=Path("/tmp/x") + ) def test_bad_json_raises(self, client): with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=b"not-json")): with pytest.raises(GpuEncodeError, match="bad JSON"): - client._post_sync({"inputs":{},"ffmpeg_args":[],"output_url":"","timeout":10}, - mezzanine_path=Path("/tmp/x")) + client._post_sync( + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10}, mezzanine_path=Path("/tmp/x") + ) def test_uploaded_false_logs_warning_but_succeeds(self, client, caplog): - body = {"status":"completed","ffmpeg_rc":0,"uploaded":False,"job_id":"j"} - with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=body)), \ - caplog.at_level("WARNING"): - res = client._post_sync({"inputs":{},"ffmpeg_args":[],"output_url":"","timeout":10}, - mezzanine_path=Path("/tmp/x")) + body = {"status": "completed", "ffmpeg_rc": 0, "uploaded": False, "job_id": "j"} + with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=body)), caplog.at_level("WARNING"): + res = client._post_sync( + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10}, mezzanine_path=Path("/tmp/x") + ) assert res["status"] == "completed" assert "uploaded=false" in caplog.text @@ -239,18 +262,28 @@ class TestDownloadToFile: # ── encode_mezzanine_to_output ───────────────────────────────────── class TestEncodeMezzanine: def test_happy_path_with_audio(self, client, tmp_path): - mezz = tmp_path / "mezz.mp4"; mezz.write_bytes(b"M" * 100) + mezz = tmp_path / "mezz.mp4" + mezz.write_bytes(b"M" * 100) out = tmp_path / "out" / "final.mp4" with ( mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/signed", "osskey1")), - mock.patch.object(client, "_post_sync", return_value={ - "job_id":"j1","status":"completed","ffmpeg_rc":0,"uploaded":True,"size":5000,"duration":1.2, - }) as m_post, + mock.patch.object( + client, + "_post_sync", + return_value={ + "job_id": "j1", + "status": "completed", + "ffmpeg_rc": 0, + "uploaded": True, + "size": 5000, + "duration": 1.2, + }, + ) as m_post, mock.patch.object(client, "_download_to_file", return_value=5000) as m_dl, mock.patch.object(client, "_relay_delete") as m_del, mock.patch.object(client, "_delete_oss") as m_ossdel, ): - result = client.encode_mezzanine_to_output(mezz, out, audio_args=["-c:a","aac"]) + result = client.encode_mezzanine_to_output(mezz, out, audio_args=["-c:a", "aac"]) assert result["output_size"] == 5000 and str(out) == result["output_path"] body = m_post.call_args[0][0] assert "-c:a" in body["ffmpeg_args"] and "aac" in body["ffmpeg_args"] @@ -261,11 +294,16 @@ class TestEncodeMezzanine: m_ossdel.assert_called_once_with("osskey1") def test_happy_path_no_audio_uses_an_and_cq(self, client, tmp_path): - mezz = tmp_path / "m.mp4"; mezz.write_bytes(b"M") + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"M") out = tmp_path / "o.mp4" with ( - mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u","k")), - mock.patch.object(client, "_post_sync", return_value={"status":"completed","ffmpeg_rc":0,"uploaded":True,"job_id":"j"}) as m_post, + mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u", "k")), + mock.patch.object( + client, + "_post_sync", + return_value={"status": "completed", "ffmpeg_rc": 0, "uploaded": True, "job_id": "j"}, + ) as m_post, mock.patch.object(client, "_download_to_file", return_value=100), mock.patch.object(client, "_relay_delete"), mock.patch.object(client, "_delete_oss"), @@ -276,18 +314,26 @@ class TestEncodeMezzanine: assert str(client.crf) in body["ffmpeg_args"] def test_bitrate_set_uses_bv_instead_of_cq(self, tmp_path): - c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api", - relay_internal_base_url="http://api-int:8000", relay_secret="s", bitrate="2M") - mezz = tmp_path / "m.mp4"; mezz.write_bytes(b"x") + c = GpuEncoderClient( + endpoint="http://gpu", + relay_base_url="http://api", + relay_internal_base_url="http://api-int:8000", + relay_secret="s", + bitrate="2M", + ) + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") out = tmp_path / "o.mp4" with ( - mock.patch.object(c, "_upload_mezzanine", return_value=("http://oss/u","k")), - mock.patch.object(c, "_post_sync", return_value={"status":"completed","ffmpeg_rc":0,"uploaded":True,"job_id":"j"}) as m_post, + mock.patch.object(c, "_upload_mezzanine", return_value=("http://oss/u", "k")), + mock.patch.object( + c, "_post_sync", return_value={"status": "completed", "ffmpeg_rc": 0, "uploaded": True, "job_id": "j"} + ) as m_post, mock.patch.object(c, "_download_to_file", return_value=10), mock.patch.object(c, "_relay_delete"), mock.patch.object(c, "_delete_oss"), ): - c.encode_mezzanine_to_output(mezz, out, extra_video_args=["-vf","scale=1280:-2"]) + c.encode_mezzanine_to_output(mezz, out, extra_video_args=["-vf", "scale=1280:-2"]) body = m_post.call_args[0][0] assert "-b:v" in body["ffmpeg_args"] and "2M" in body["ffmpeg_args"] assert "-cq" not in body["ffmpeg_args"] @@ -295,44 +341,48 @@ class TestEncodeMezzanine: def test_mezzanine_not_found_raises(self, client, tmp_path): with pytest.raises(GpuEncodeError, match="mezzanine file not found"): - client.encode_mezzanine_to_output(tmp_path/"nope.mp4", tmp_path/"o.mp4") + client.encode_mezzanine_to_output(tmp_path / "nope.mp4", tmp_path / "o.mp4") def test_relay_base_not_configured_raises(self, tmp_path): c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="", relay_secret="s") - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") with pytest.raises(GpuEncodeError, match="relay_base_url"): - c.encode_mezzanine_to_output(mezz, tmp_path/"o.mp4") + c.encode_mezzanine_to_output(mezz, tmp_path / "o.mp4") def test_unexpected_exception_is_wrapped(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") with ( - mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u","k")), + mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u", "k")), mock.patch.object(client, "_post_sync", side_effect=RuntimeError("boom")), mock.patch.object(client, "_delete_oss"), ): with pytest.raises(GpuEncodeError, match="unexpected: boom"): - client.encode_mezzanine_to_output(mezz, tmp_path/"o.mp4") + client.encode_mezzanine_to_output(mezz, tmp_path / "o.mp4") def test_gpu_encode_error_re_raised_directly(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") with ( - mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u","k")), + mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u", "k")), mock.patch.object(client, "_post_sync", side_effect=GpuEncodeError("direct fail")), mock.patch.object(client, "_delete_oss"), ): with pytest.raises(GpuEncodeError, match="direct fail"): - client.encode_mezzanine_to_output(mezz, tmp_path/"o.mp4") + client.encode_mezzanine_to_output(mezz, tmp_path / "o.mp4") def test_oss_cleanup_runs_on_failure(self, client, tmp_path, caplog): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") with ( - mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u","ossk")), + mock.patch.object(client, "_upload_mezzanine", return_value=("http://oss/u", "ossk")), mock.patch.object(client, "_post_sync", side_effect=GpuEncodeError("enc fail")), mock.patch.object(client, "_delete_oss", side_effect=Exception("oss down")) as m_ossdel, caplog.at_level("WARNING"), ): with pytest.raises(GpuEncodeError): - client.encode_mezzanine_to_output(mezz, tmp_path/"o.mp4") + client.encode_mezzanine_to_output(mezz, tmp_path / "o.mp4") m_ossdel.assert_called_once_with("ossk") @@ -352,15 +402,18 @@ class TestRelayDelete: # ── OSS helpers ──────────────────────────────────────────────────── class TestOssHelpers: def test_upload_storage_import_error(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") # 删除 sys.modules 中 packages.shared.storage 使导入失败 saved = sys.modules.pop("packages.shared.storage", None) try: - real_import = __builtins__.__import__ if hasattr(__builtins__,"__import__") else __import__ + real_import = __builtins__.__import__ if hasattr(__builtins__, "__import__") else __import__ + def fake_import(name, *a, **kw): if name == "packages.shared.storage" or name.startswith("packages.shared.storage."): raise ImportError("no storage") return real_import(name, *a, **kw) + with mock.patch("builtins.__import__", side_effect=fake_import): with pytest.raises(GpuEncodeError, match="storage service unavailable"): client._upload_mezzanine(mezz) @@ -369,33 +422,45 @@ class TestOssHelpers: sys.modules["packages.shared.storage"] = saved def test_upload_storage_none(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.return_value = None + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.return_value = None with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}): with pytest.raises(GpuEncodeError, match="OSS storage not configured"): client._upload_mezzanine(mezz) def test_upload_bucket_none(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") - svc = mock.MagicMock(); svc.bucket = None - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.return_value = svc + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") + svc = mock.MagicMock() + svc.bucket = None + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.return_value = svc with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}): with pytest.raises(GpuEncodeError, match="OSS storage not configured"): client._upload_mezzanine(mezz) def test_upload_failure_raises(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") - svc = mock.MagicMock(); svc.bucket = object(); svc.upload_file.side_effect = RuntimeError("oss err") - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.return_value = svc + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") + svc = mock.MagicMock() + svc.bucket = object() + svc.upload_file.side_effect = RuntimeError("oss err") + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.return_value = svc with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}): with pytest.raises(GpuEncodeError, match="failed to upload mezzanine"): client._upload_mezzanine(mezz) def test_upload_success(self, client, tmp_path): - mezz = tmp_path/"m.mp4"; mezz.write_bytes(b"x") - svc = mock.MagicMock(); svc.bucket = object() + mezz = tmp_path / "m.mp4" + mezz.write_bytes(b"x") + svc = mock.MagicMock() + svc.bucket = object() svc.get_download_url.return_value = "https://oss/signed?sig=abc" - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.return_value = svc + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.return_value = svc with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}): url, key = client._upload_mezzanine(mezz) assert url.startswith("https://oss/signed") @@ -403,21 +468,26 @@ class TestOssHelpers: svc.upload_file.assert_called_once() def test_delete_oss_exception_swallowed(self, client, caplog): - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.side_effect = RuntimeError("svc down") + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.side_effect = RuntimeError("svc down") with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}), caplog.at_level("DEBUG"): client._delete_oss("somekey") assert "OSS delete" in caplog.text def test_delete_oss_bucket_none_noop(self, client): - svc = mock.MagicMock(); svc.bucket = None - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.return_value = svc + svc = mock.MagicMock() + svc.bucket = None + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.return_value = svc with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}): client._delete_oss("k") svc.delete_file.assert_not_called() def test_delete_oss_success(self, client): - svc = mock.MagicMock(); svc.bucket = object() - fake_mod = mock.MagicMock(); fake_mod.get_storage_service.return_value = svc + svc = mock.MagicMock() + svc.bucket = object() + fake_mod = mock.MagicMock() + fake_mod.get_storage_service.return_value = svc with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}): client._delete_oss("k") svc.delete_file.assert_called_once_with("k") @@ -436,36 +506,51 @@ class TestSingletonFactory: sys.modules["packages.config"] = saved def test_build_client_not_enabled_returns_none(self): - s = mock.MagicMock(); s.enable_gpu_encode = False - fake_mod = mock.MagicMock(); fake_mod.get_shared_settings.return_value = s + s = mock.MagicMock() + s.enable_gpu_encode = False + fake_mod = mock.MagicMock() + fake_mod.get_shared_settings.return_value = s with mock.patch.dict("sys.modules", {"packages.config": fake_mod}): assert _build_client_from_settings() is None def test_build_client_missing_endpoint(self): - s = mock.MagicMock(); s.enable_gpu_encode = True - s.gpu_encode_endpoint = ""; s.gpu_encode_relay_base_url = "http://api" + s = mock.MagicMock() + s.enable_gpu_encode = True + s.gpu_encode_endpoint = "" + s.gpu_encode_relay_base_url = "http://api" s.gpu_encode_relay_internal_base_url = "" - fake_mod = mock.MagicMock(); fake_mod.get_shared_settings.return_value = s + fake_mod = mock.MagicMock() + fake_mod.get_shared_settings.return_value = s with mock.patch.dict("sys.modules", {"packages.config": fake_mod}): assert _build_client_from_settings() is None def test_build_client_missing_relay(self): - s = mock.MagicMock(); s.enable_gpu_encode = True - s.gpu_encode_endpoint = "http://gpu"; s.gpu_encode_relay_base_url = "" + s = mock.MagicMock() + s.enable_gpu_encode = True + s.gpu_encode_endpoint = "http://gpu" + s.gpu_encode_relay_base_url = "" s.gpu_encode_relay_internal_base_url = "" - fake_mod = mock.MagicMock(); fake_mod.get_shared_settings.return_value = s + fake_mod = mock.MagicMock() + fake_mod.get_shared_settings.return_value = s with mock.patch.dict("sys.modules", {"packages.config": fake_mod}): assert _build_client_from_settings() is None def test_build_client_success(self): - s = mock.MagicMock(); s.enable_gpu_encode = True - s.gpu_encode_endpoint = "http://gpu"; s.gpu_encode_relay_base_url = "http://api/" + s = mock.MagicMock() + s.enable_gpu_encode = True + s.gpu_encode_endpoint = "http://gpu" + s.gpu_encode_relay_base_url = "http://api/" s.gpu_encode_relay_internal_base_url = "http://api-int:8000/" - s.gpu_encode_sync_timeout = 120; s.gpu_encode_health_timeout = 1.0 - s.gpu_encode_vcodec = "h264_nvenc"; s.gpu_encode_preset = "p7" - s.gpu_encode_crf = 20; s.gpu_encode_bitrate = ""; s.gpu_encode_relay_secret = "s" + s.gpu_encode_sync_timeout = 120 + s.gpu_encode_health_timeout = 1.0 + s.gpu_encode_vcodec = "h264_nvenc" + s.gpu_encode_preset = "p7" + s.gpu_encode_crf = 20 + s.gpu_encode_bitrate = "" + s.gpu_encode_relay_secret = "s" s.gpu_encode_oss_tmp_prefix = "tmp/x/" - fake_mod = mock.MagicMock(); fake_mod.get_shared_settings.return_value = s + fake_mod = mock.MagicMock() + fake_mod.get_shared_settings.return_value = s with mock.patch.dict("sys.modules", {"packages.config": fake_mod}): c = _build_client_from_settings() assert c is not None and c.endpoint == "http://gpu" @@ -474,8 +559,10 @@ class TestSingletonFactory: assert c.preset == "p7" and c.crf == 20 def test_get_gpu_encoder_init_failure_returns_none(self, caplog): - with mock.patch("packages.shared.gpu_encoder._build_client_from_settings", - side_effect=RuntimeError("boom")), caplog.at_level("WARNING"): + with ( + mock.patch("packages.shared.gpu_encoder._build_client_from_settings", side_effect=RuntimeError("boom")), + caplog.at_level("WARNING"), + ): assert get_gpu_encoder() is None assert "failed to init client" in caplog.text @@ -493,11 +580,11 @@ class TestSingletonFactory: # ── Constructor edge cases ──────────────────────────────────────── class TestConstructor: def test_oss_tmp_prefix_empty_uses_default(self): - c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api", - relay_secret="s", oss_tmp_prefix="") + c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api", relay_secret="s", oss_tmp_prefix="") assert c.oss_tmp_prefix == "tmp/gpu-mezzanine/" def test_oss_tmp_prefix_strips_and_adds_slash(self): - c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api", - relay_secret="s", oss_tmp_prefix="tmp/foo") + c = GpuEncoderClient( + endpoint="http://gpu", relay_base_url="http://api", relay_secret="s", oss_tmp_prefix="tmp/foo" + ) assert c.oss_tmp_prefix == "tmp/foo/" diff --git a/tests/unit/test_gpu_relay.py b/tests/unit/test_gpu_relay.py index 36c630370..b5077b73a 100644 --- a/tests/unit/test_gpu_relay.py +++ b/tests/unit/test_gpu_relay.py @@ -1,4 +1,5 @@ """gpu_relay API 路由单元测试：覆盖 helper 函数 + PUT/GET/HEAD/DELETE handler。""" + from __future__ import annotations import os @@ -101,8 +102,9 @@ class TestBuildRelayUrls: assert url == "http://api.example.com/api/v1/internal/gpu-relay/k1?token=s" def test_get_url_same_as_put(self): - assert gpu_relay.build_relay_get_url("http://api", "k", "s") == \ - gpu_relay.build_relay_put_url("http://api", "k", "s") + assert gpu_relay.build_relay_get_url("http://api", "k", "s") == gpu_relay.build_relay_put_url( + "http://api", "k", "s" + ) def test_generate_key_is_hex(self): k = gpu_relay.generate_key() @@ -116,11 +118,14 @@ class TestPutObject: async def test_put_writes_file(self, tmp_path, monkeypatch): monkeypatch.setenv("GPU_ENCODE_RELAY_SECRET", "s") monkeypatch.setenv("GENERATED_FILES_DIR", str(tmp_path)) + # async request.stream 模拟 async def _stream(): yield b"chunk1" yield b"chunk2" - req = mock.MagicMock(); req.stream = _stream + + req = mock.MagicMock() + req.stream = _stream resp = await gpu_relay.put_object(key="abc123", request=req, token="s") assert resp["ok"] is True assert resp["size"] == len(b"chunk1") + len(b"chunk2") @@ -148,10 +153,13 @@ class TestPutObject: async def test_put_write_error_cleans_tmp(self, tmp_path, monkeypatch): monkeypatch.setenv("GPU_ENCODE_RELAY_SECRET", "s") monkeypatch.setenv("GENERATED_FILES_DIR", str(tmp_path)) + async def _bad_stream(): yield b"x" raise OSError("disk full") - req = mock.MagicMock(); req.stream = _bad_stream + + req = mock.MagicMock() + req.stream = _bad_stream with pytest.raises(HTTPException) as ei: await gpu_relay.put_object(key="abc", request=req, token="s") assert ei.value.status_code == 500 @@ -192,7 +200,9 @@ class TestHeadObject: async def test_head_returns_content_length(self, tmp_path, monkeypatch): monkeypatch.setenv("GPU_ENCODE_RELAY_SECRET", "s") monkeypatch.setenv("GENERATED_FILES_DIR", str(tmp_path)) - p = tmp_path / "gpu_relay" / "k"; p.parent.mkdir(parents=True, exist_ok=True); p.write_bytes(b"12345") + p = tmp_path / "gpu_relay" / "k" + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(b"12345") resp = await gpu_relay.head_object(key="k", token="s") assert resp.status_code == 200 assert resp.headers["Content-Length"] == "5" @@ -204,7 +214,9 @@ class TestDeleteObject: async def test_delete_existing(self, tmp_path, monkeypatch): monkeypatch.setenv("GPU_ENCODE_RELAY_SECRET", "s") monkeypatch.setenv("GENERATED_FILES_DIR", str(tmp_path)) - p = tmp_path / "gpu_relay" / "k"; p.parent.mkdir(parents=True, exist_ok=True); p.write_bytes(b"x") + p = tmp_path / "gpu_relay" / "k" + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(b"x") resp = await gpu_relay.delete_object(key="k", token="s") assert resp["ok"] is True assert not p.exists() @@ -219,7 +231,9 @@ class TestDeleteObject: async def test_delete_unlink_error_500(self, tmp_path, monkeypatch): monkeypatch.setenv("GPU_ENCODE_RELAY_SECRET", "s") monkeypatch.setenv("GENERATED_FILES_DIR", str(tmp_path)) - p = tmp_path / "gpu_relay" / "k"; p.parent.mkdir(parents=True, exist_ok=True); p.write_bytes(b"x") + p = tmp_path / "gpu_relay" / "k" + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(b"x") with mock.patch.object(Path, "unlink", side_effect=OSError("perm denied")): with pytest.raises(HTTPException) as ei: await gpu_relay.delete_object(key="k", token="s")