diff --git a/tests/unit/test_gpu_encoder.py b/tests/unit/test_gpu_encoder.py index 17d9f4451..b4d01dc9e 100644 --- a/tests/unit/test_gpu_encoder.py +++ b/tests/unit/test_gpu_encoder.py @@ -38,6 +38,7 @@ def client(): endpoint="http://gpu.example.com:8900", relay_base_url="http://api.example.com", relay_internal_base_url="http://api-internal:8000", + mezzanine_transport="oss", # 旧测试只 mock OSS 上传，走 OSS 路径 sync_timeout=60, health_timeout=2, relay_secret="test-secret", @@ -119,7 +120,6 @@ class TestPostSync: "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] @@ -130,7 +130,7 @@ class TestPostSync: 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") + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10} ) def test_http_4xx_raises(self, client): @@ -140,64 +140,87 @@ class TestPostSync: 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") + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10} ) 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") + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10} ) 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") + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10} ) 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") + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10} ) 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") - ) + {"inputs": {}, "ffmpeg_args": [], "output_url": "", "timeout": 10} + ) assert res["status"] == "completed" assert "uploaded=false" in caplog.text # ── Relay URL builders ────────────────────────────────────────────── class TestRelayUrl: - def test_put_url_uses_external_base(self, client): - url = client._relay_put_url("abc123", "secret!") + def test_result_put_url_uses_external_base(self, client): + """P4000 回传编码结果的 PUT URL 应走外部 base（Tailscale 可达）。""" + url = client._result_put_url("abc123", "secret!") assert "abc123" in url assert "token=secret%21" in url assert url.startswith("http://api.example.com/api/v1/internal/gpu-relay/") + assert "/mezzanine/" not in url - def test_internal_url_uses_internal_base(self, client): - url = client._relay_internal_url("abc123", "s") + def test_result_get_url_uses_internal_base(self, client): + """Worker 下载结果使用 internal base（Docker DNS 直连）。""" + url = client._result_get_url("abc123", "s") assert url.startswith("http://api-internal:8000/api/v1/internal/gpu-relay/abc123") + assert "/mezzanine/" not in url - def test_internal_url_falls_back_to_external_when_not_set(self): - c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api.example.com", relay_secret="s") - put = c._relay_put_url("k", "s") - internal = c._relay_internal_url("k", "s") + def test_result_get_url_falls_back_to_external_when_not_set(self): + c = GpuEncoderClient( + endpoint="http://gpu", relay_base_url="http://api.example.com", + relay_secret="s", mezzanine_transport="oss", + ) + put = c._result_put_url("k", "s") + get = c._result_get_url("k", "s") assert put.startswith("http://api.example.com/") - assert internal == put + assert get == put def test_encode_uses_different_put_and_get_urls(self, client): - put_url = client._relay_put_url("k", "test-secret") - get_url = client._relay_internal_url("k", "test-secret") + put_url = client._result_put_url("k", "test-secret") + get_url = client._result_get_url("k", "test-secret") assert "api.example.com" in put_url and "api-internal:8000" in get_url and put_url != get_url + def test_mezz_put_url_uses_internal_base(self, client): + """Worker 上传 mezzanine 的 PUT URL 应走 internal base（Docker 内网快）。""" + url = client._mezz_put_url("m1", "test-secret") + assert url.startswith("http://api-internal:8000/api/v1/internal/gpu-relay/mezzanine/m1") + + def test_mezz_get_url_for_p4000_uses_external_base(self, client): + """P4000 下载 mezzanine 的 GET URL 必须是 P4000 可达的（Tailscale 外部 base）。""" + url = client._mezz_get_url_for_p4000("m1", "test-secret") + assert url.startswith("http://api.example.com/api/v1/internal/gpu-relay/mezzanine/m1") + + def test_mezz_put_get_use_different_bases(self, client): + """worker 上传 mezzanine 用 internal，P4000 下载 mezzanine 用 external。""" + put = client._mezz_put_url("m", "test-secret") + get = client._mezz_get_url_for_p4000("m", "test-secret") + assert "api-internal:8000" in put and "api.example.com" in get and put != get + # ── _get_relay_secret ────────────────────────────────────────────── class TestGetRelaySecret: @@ -266,7 +289,7 @@ class TestEncodeMezzanine: 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, "_upload_mezzanine_to_oss", return_value=("http://oss/signed", "osskey1")), mock.patch.object( client, "_post_sync", @@ -298,7 +321,7 @@ class TestEncodeMezzanine: 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, "_upload_mezzanine_to_oss", return_value=("http://oss/u", "k")), mock.patch.object( client, "_post_sync", @@ -325,7 +348,7 @@ class TestEncodeMezzanine: 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, "_upload_mezzanine_to_oss", 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, @@ -354,7 +377,7 @@ class TestEncodeMezzanine: 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_to_oss", return_value=("http://oss/u", "k")), mock.patch.object(client, "_post_sync", side_effect=RuntimeError("boom")), mock.patch.object(client, "_delete_oss"), ): @@ -365,7 +388,7 @@ class TestEncodeMezzanine: 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_to_oss", return_value=("http://oss/u", "k")), mock.patch.object(client, "_post_sync", side_effect=GpuEncodeError("direct fail")), mock.patch.object(client, "_delete_oss"), ): @@ -376,7 +399,7 @@ class TestEncodeMezzanine: 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_to_oss", 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"), @@ -416,7 +439,7 @@ class TestOssHelpers: with mock.patch("builtins.__import__", side_effect=fake_import): with pytest.raises(GpuEncodeError, match="storage service unavailable"): - client._upload_mezzanine(mezz) + client._upload_mezzanine_to_oss(mezz) finally: if saved is not None: sys.modules["packages.shared.storage"] = saved @@ -428,7 +451,7 @@ class TestOssHelpers: 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) + client._upload_mezzanine_to_oss(mezz) def test_upload_bucket_none(self, client, tmp_path): mezz = tmp_path / "m.mp4" @@ -439,7 +462,7 @@ class TestOssHelpers: 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) + client._upload_mezzanine_to_oss(mezz) def test_upload_failure_raises(self, client, tmp_path): mezz = tmp_path / "m.mp4" @@ -451,7 +474,7 @@ class TestOssHelpers: 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) + client._upload_mezzanine_to_oss(mezz) def test_upload_success(self, client, tmp_path): mezz = tmp_path / "m.mp4" @@ -462,7 +485,7 @@ class TestOssHelpers: 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) + url, key = client._upload_mezzanine_to_oss(mezz) assert url.startswith("https://oss/signed") assert key.startswith("tmp/gpu-mezzanine/") and key.endswith(".mp4") svc.upload_file.assert_called_once() @@ -493,6 +516,105 @@ class TestOssHelpers: svc.delete_file.assert_called_once_with("k") + +# ── encode_mezzanine_to_output: relay transport (default path) ────── +class TestEncodeMezzanineRelay: + def test_relay_happy_path_uploads_to_relay(self, tmp_path): + """默认 relay 模式：worker PUT mezzanine 到 internal relay；P4000 GET 用 external base。""" + c = GpuEncoderClient( + endpoint="http://gpu.example.com:8900", + relay_base_url="http://api.example.com", + relay_internal_base_url="http://api-internal:8000", + mezzanine_transport="relay", + sync_timeout=60, + relay_secret="test-secret", + ) + mezz = tmp_path / "mezz.mp4" + mezz.write_bytes(b"X" * 200) + out = tmp_path / "out.mp4" + with ( + mock.patch.object(c, "_upload_file_put") as m_put, + mock.patch.object( + c, "_post_sync", + return_value={"job_id": "j", "status": "completed", "ffmpeg_rc": 0, + "uploaded": True, "size": 100, "duration": 1.0}, + ) as m_post, + mock.patch.object(c, "_download_to_file", return_value=100), + mock.patch.object(c, "_relay_delete") as m_del, + mock.patch.object(c, "_delete_oss") as m_ossdel, + ): + result = c.encode_mezzanine_to_output(mezz, out) + assert result["transport"] == "relay" + # _upload_file_put called once (mezz uploaded to relay) + m_put.assert_called_once() + put_url = m_put.call_args[0][0] + assert "/api/v1/internal/gpu-relay/mezzanine/" in put_url + assert "api-internal:8000" in put_url # worker uses internal + # P4000 body.inputs["in.mp4"] must use external base (P4000 reachable) + body = m_post.call_args[0][0] + assert body["inputs"]["in.mp4"].startswith("http://api.example.com/api/v1/internal/gpu-relay/mezzanine/") + # No OSS involved + m_ossdel.assert_not_called() + # cleanup: one result delete + one mezz delete + assert m_del.call_count == 2 + + def test_relay_put_failure_falls_back_to_oss(self, tmp_path): + """relay PUT 失败时应自动 fallback 到 OSS，且 transport 标记为 oss。""" + c = GpuEncoderClient( + endpoint="http://gpu.example.com:8900", + relay_base_url="http://api.example.com", + relay_internal_base_url="http://api-internal:8000", + mezzanine_transport="relay", + sync_timeout=60, + relay_secret="test-secret", + ) + mezz = tmp_path / "mezz.mp4" + mezz.write_bytes(b"X") + out = tmp_path / "out.mp4" + with ( + mock.patch.object(c, "_upload_file_put", side_effect=GpuEncodeError("relay 500")), + mock.patch.object(c, "_upload_mezzanine_to_oss", return_value=("https://oss/signed", "ossk")), + mock.patch.object( + c, "_post_sync", + return_value={"job_id": "j", "status": "completed", "ffmpeg_rc": 0, "uploaded": True}, + ) as m_post, + mock.patch.object(c, "_download_to_file", return_value=50), + mock.patch.object(c, "_relay_delete"), + mock.patch.object(c, "_delete_oss") as m_ossdel, + ): + result = c.encode_mezzanine_to_output(mezz, out) + assert result["transport"] == "oss" + body = m_post.call_args[0][0] + assert body["inputs"]["in.mp4"] == "https://oss/signed" + m_ossdel.assert_called_once_with("ossk") + + def test_upload_file_put_streaming_sends_content_length(self, client, tmp_path): + """_upload_file_put 应用 Content-Length 头发送文件。""" + f = tmp_path / "x.mp4" + f.write_bytes(b"ABCDEFGH") # 8 bytes + captured = {} + + def fake_urlopen(req, timeout=None): + captured["method"] = req.get_method() + captured["cl"] = req.get_header("Content-length") + captured["ct"] = req.get_header("Content-type") + captured["data"] = req.data.read() + return _fake_response(status=200, body=b"") + + with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen): + client._upload_file_put("http://relay/m?token=s", f, "video/mp4") + assert captured["method"] == "PUT" + assert captured["cl"] == "8" + assert captured["ct"] == "video/mp4" + assert captured["data"] == b"ABCDEFGH" + + def test_upload_file_put_non_2xx_raises(self, client, tmp_path): + f = tmp_path / "x.mp4" + f.write_bytes(b"x") + with mock.patch("urllib.request.urlopen", return_value=_fake_response(status=500, body=b"err")): + with pytest.raises(GpuEncodeError, match="relay PUT failed"): + client._upload_file_put("http://relay/m", f, "video/mp4") + # ── Singleton / factory ──────────────────────────────────────────── class TestSingletonFactory: def test_build_client_import_error_returns_none(self): @@ -549,6 +671,7 @@ class TestSingletonFactory: s.gpu_encode_bitrate = "" s.gpu_encode_relay_secret = "s" s.gpu_encode_oss_tmp_prefix = "tmp/x/" + s.gpu_encode_mezzanine_transport = "relay" fake_mod = mock.MagicMock() fake_mod.get_shared_settings.return_value = s with mock.patch.dict("sys.modules", {"packages.config": fake_mod}): @@ -557,6 +680,7 @@ class TestSingletonFactory: assert c.relay_base_url == "http://api" assert c.relay_internal_base_url == "http://api-int:8000" assert c.preset == "p7" and c.crf == 20 + assert c.mezzanine_transport == "relay" def test_get_gpu_encoder_init_failure_returns_none(self, caplog): with (