test(tests/unit/test_gpu_encoder.py): adapt to veryfast preset + gpu mezzanine relay refactor

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 (