"""GpuEncoderClient 单元测试：mock HTTP，覆盖 health/sync/fallback/singleton 等完整路径。"""

from __future__ import annotations

import json
import socket
import sys
import urllib.error
import urllib.request
from http.client import HTTPResponse
from io import BytesIO
from pathlib import Path
from unittest import mock

import pytest

from packages.shared.gpu_encoder import (
    GpuEncodeError,
    GpuEncoderClient,
    GpuHealth,
    _build_client_from_settings,
    get_gpu_encoder,
    is_gpu_encode_enabled,
    reset_gpu_encoder_for_tests,
)


@pytest.fixture(autouse=True)
def _reset_singleton():
    reset_gpu_encoder_for_tests()
    yield
    reset_gpu_encoder_for_tests()


@pytest.fixture
def client():
    return GpuEncoderClient(
        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",
    )


def _fake_response(status: int = 200, body: dict | bytes | None = None, headers=None):
    if isinstance(body, dict):
        data = json.dumps(body).encode("utf-8")
    elif body is None:
        data = b""
    else:
        data = body
    bio = BytesIO(data)
    resp = mock.MagicMock(spec=HTTPResponse)
    resp.status = status
    resp.read.side_effect = lambda n=-1: bio.read(n)
    resp.__enter__ = mock.MagicMock(return_value=resp)
    resp.__exit__ = mock.MagicMock(return_value=False)
    return resp


# ── 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},
        }
        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
        assert h.gpu_name == "Quadro P4000"

    def test_connection_error_returns_unhealthy(self, client):
        with mock.patch("urllib.request.urlopen", side_effect=urllib.error.URLError("timeout")):
            h = client.check_health()
        assert not h.healthy
        assert "health probe failed" in h.error

    def test_bad_json_returns_unhealthy(self, client):
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=b"not json")):
            h = client.check_health()
        assert not h.healthy

    def test_nvenc_unavailable(self, client):
        body = {"status": "healthy", "gpu": {"name": "t"}, "nvenc": {"h264_nvenc": False}}
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=body)):
            h = client.check_health()
        assert h.healthy and not h.ready

    def test_malformed_response_inner_exception(self, client):
        """data 是合法 JSON 但 gpu 字段类型错（字符串）触发内部 except."""
        body = {"status": "healthy", "gpu": "not-a-dict", "nvenc": {}}
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=body)):
            h = client.check_health()
        assert not h.healthy
        assert "malformed" in h.error


# ── _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,
        }
        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,
                },
            )
        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"}
        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})

    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")
        )
        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})

    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})

    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})

    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})

    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})
        assert res["status"] == "completed"
        assert "uploaded=false" in caplog.text


# ── Relay URL builders ──────────────────────────────────────────────
class TestRelayUrl:
    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_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_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 get == put

    def test_encode_uses_different_put_and_get_urls(self, client):
        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:
    def test_explicit_secret_used(self, client):
        assert client._get_relay_secret() == "test-secret"

    def test_env_secret_used_when_not_explicit(self, monkeypatch):
        monkeypatch.setenv("GPU_ENCODE_RELAY_SECRET", "from-env")
        monkeypatch.setenv("APP_ENV", "staging")
        c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api")
        assert c._get_relay_secret() == "from-env"

    def test_prod_without_secret_raises(self, monkeypatch):
        monkeypatch.delenv("GPU_ENCODE_RELAY_SECRET", raising=False)
        monkeypatch.setenv("APP_ENV", "production")
        c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api")
        with pytest.raises(GpuEncodeError, match="GPU_ENCODE_RELAY_SECRET"):
            c._get_relay_secret()

    def test_dev_without_secret_raises(self, monkeypatch):
        """未设置 secret 且非 production 也 raise（worker 必须显式配置）。"""
        monkeypatch.delenv("GPU_ENCODE_RELAY_SECRET", raising=False)
        monkeypatch.setenv("APP_ENV", "development")
        c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api")
        with pytest.raises(GpuEncodeError, match="GPU_ENCODE_RELAY_SECRET not set"):
            c._get_relay_secret()


# ── _download_to_file ──────────────────────────────────────────────
class TestDownloadToFile:
    def test_writes_file(self, client, tmp_path):
        data = b"hello" * 1000
        out = tmp_path / "out.mp4"
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=data)):
            size = client._download_to_file("http://relay/k?token=s", out)
        assert size == len(data) and out.read_bytes() == data

    def test_empty_file_raises(self, client, tmp_path):
        out = tmp_path / "out.mp4"
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(body=b"")):
            with pytest.raises(GpuEncodeError, match="empty file"):
                client._download_to_file("http://relay/k", out)
        assert not out.exists()

    def test_non_200_status_raises(self, client, tmp_path):
        out = tmp_path / "o.mp4"
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(status=404, body=b"")):
            with pytest.raises(GpuEncodeError, match="HTTP 404"):
                client._download_to_file("http://relay/k", out)

    def test_url_error_cleans_up_tmp(self, client, tmp_path):
        out = tmp_path / "o.mp4"
        tmp_file = out.with_suffix(out.suffix + ".gpu_tmp")
        tmp_file.write_bytes(b"partial")
        assert tmp_file.exists()
        with mock.patch("urllib.request.urlopen", side_effect=urllib.error.URLError("net down")):
            with pytest.raises(GpuEncodeError, match="failed to download"):
                client._download_to_file("http://relay/k", out)
        assert not tmp_file.exists()


# ── 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)
        out = tmp_path / "out" / "final.mp4"
        with (
            mock.patch.object(client, "_upload_mezzanine_to_oss", 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, "_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"])
        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"]
        assert "-an" not in body["ffmpeg_args"]
        assert body["output_url"].startswith("http://api.example.com/")
        assert "api-internal:8000" in m_dl.call_args[0][0]
        m_del.assert_called_once()
        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")
        out = tmp_path / "o.mp4"
        with (
            mock.patch.object(client, "_upload_mezzanine_to_oss", 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"),
        ):
            client.encode_mezzanine_to_output(mezz, out)
            body = m_post.call_args[0][0]
            assert "-an" in body["ffmpeg_args"] and "-cq" in body["ffmpeg_args"]
            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")
        out = tmp_path / "o.mp4"
        with (
            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,
            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"])
            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"]
            assert "-vf" in body["ffmpeg_args"]

    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")

    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")
        with pytest.raises(GpuEncodeError, match="relay_base_url"):
            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")
        with (
            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"),
        ):
            with pytest.raises(GpuEncodeError, match="unexpected: boom"):
                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")
        with (
            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"),
        ):
            with pytest.raises(GpuEncodeError, match="direct fail"):
                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")
        with (
            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"),
        ):
            with pytest.raises(GpuEncodeError):
                client.encode_mezzanine_to_output(mezz, tmp_path / "o.mp4")
            m_ossdel.assert_called_once_with("ossk")


# ── _relay_delete ──────────────────────────────────────────────────
class TestRelayDelete:
    def test_exception_is_swallowed(self, client, caplog):
        with mock.patch("urllib.request.urlopen", side_effect=RuntimeError("boom")), caplog.at_level("DEBUG"):
            client._relay_delete("http://relay/k?token=s")
        assert "cleanup delete failed" in caplog.text

    def test_success_issues_delete(self, client):
        with mock.patch("urllib.request.urlopen", return_value=_fake_response(status=204, body=b"")) as m:
            client._relay_delete("http://relay/k?token=s")
        assert m.call_args[0][0].get_method() == "DELETE"


# ── OSS helpers ────────────────────────────────────────────────────
class TestOssHelpers:
    def test_upload_storage_import_error(self, client, tmp_path):
        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__

            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_to_oss(mezz)
        finally:
            if saved is not None:
                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
        with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}):
            with pytest.raises(GpuEncodeError, match="OSS storage not configured"):
                client._upload_mezzanine_to_oss(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
        with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}):
            with pytest.raises(GpuEncodeError, match="OSS storage not configured"):
                client._upload_mezzanine_to_oss(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
        with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}):
            with pytest.raises(GpuEncodeError, match="failed to upload mezzanine"):
                client._upload_mezzanine_to_oss(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()
        svc.get_download_url.return_value = "https://oss/signed?sig=abc"
        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_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()

    def test_delete_oss_exception_swallowed(self, client, caplog):
        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
        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
        with mock.patch.dict("sys.modules", {"packages.shared.storage": fake_mod}):
            client._delete_oss("k")
        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):
        saved = sys.modules.get("packages.config")
        sys.modules["packages.config"] = None
        try:
            with mock.patch("builtins.__import__", side_effect=RuntimeError("no cfg")):
                assert _build_client_from_settings() is None
        finally:
            if saved is not None:
                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
        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.gpu_encode_relay_internal_base_url = ""
        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.gpu_encode_relay_internal_base_url = ""
        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.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_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}):
            c = _build_client_from_settings()
        assert c is not None and c.endpoint == "http://gpu"
        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 (
            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

    def test_get_gpu_encoder_returns_singleton_and_enabled(self):
        c = GpuEncoderClient(endpoint="http://gpu", relay_base_url="http://api", relay_secret="s")
        with mock.patch("packages.shared.gpu_encoder._build_client_from_settings", return_value=c):
            assert get_gpu_encoder() is c and get_gpu_encoder() is c
        assert is_gpu_encode_enabled() is True

    def test_is_gpu_encode_enabled_when_none(self):
        with mock.patch("packages.shared.gpu_encoder._build_client_from_settings", return_value=None):
            assert is_gpu_encode_enabled() is False


# ── 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="")
        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"
        )
        assert c.oss_tmp_prefix == "tmp/foo/"