"""只读契约冻结：Python→Java HTTP 回调/代理侧超时与重试不受 Java 配置影响（task-175）。

背景：模块 10 的 Java 外部客户端配置治理只动 Java 内部客户端（Coze/品牌/紫鸟/图片下载），
spec §3 明确「不改 Python Worker 的 requests 调用 / 不改 Python 回调请求超时和重试约定」。
本仓库内 Python 对 Java 的 HTTP 面是 blueprints.admin_api 的 _get_backend_java_session /
_proxy_backend_java：requests.Session + HTTPAdapter(max_retries=0)，默认 timeout=10s，
全部为 Python 侧字面量/参数默认值，不读取任何 aiimage.http-client.* Java 配置。
本文件把这些约定固化为快照断言，防止将来误把 Java 配置引进来改变 Python 行为。

只读任务：不改任何生产 Python 代码，只新增本测试。
"""
import inspect
import sys
import unittest
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from blueprints import admin_api


class _FakeSession:
    """记录请求的伪 requests.Session：捕获 kwargs，返回 Java ApiResponse 成功体。"""

    def __init__(self, raise_on_request=False):
        self.calls = []
        self.raise_on_request = raise_on_request

    def request(self, method, url, **kwargs):
        self.calls.append((method, url, kwargs))
        if self.raise_on_request:
            import requests
            raise requests.RequestException("backend-java 服务不可用")

        class _Resp:
            status_code = 200

            def json(self):
                return {"success": True, "data": []}

        return _Resp()


class PythonJavaHttpContractTest(unittest.TestCase):

    def _proxy_default_http_timeout(self):
        return inspect.signature(admin_api._proxy_backend_java).parameters["timeout"].default

    def test_default_timeout_is_python_side_ten_seconds(self):
        # 契约：未显式传 timeout 时默认 (连接/读) 10s，且是签名里的字面量，非来自任何配置
        self.assertEqual(10, self._proxy_default_http_timeout())
        self.assertTrue(isinstance(self._proxy_default_http_timeout(), int))

    def test_override_timeout_forwarded_verbatim(self):
        fake = _FakeSession()
        with patch.object(admin_api, "_get_backend_java_session", return_value=fake):
            result, error_response, status = admin_api._proxy_backend_java(
                "GET", "/api/foo", params={"a": "1"}, timeout=(10, 1800))
        self.assertIsNone(error_response)
        method, url, kwargs = fake.calls[0]
        self.assertEqual((10, 1800), kwargs["timeout"], "显式超时应原样转发给 requests")

    def test_session_never_auto_retries(self):
        # 契约：Python→Java session 不自动重试（max_retries=0），失败即报错由上层处理
        session = admin_api._get_backend_java_session()
        http_adapter = session.get_adapter("http://")
        self.assertEqual(0, http_adapter.max_retries.total)

    def test_session_mounts_http_and_https(self):
        session = admin_api._get_backend_java_session()
        self.assertIsNotNone(session.get_adapter("http://"))
        self.assertIsNotNone(session.get_adapter("https://"))

    def test_timeout_is_not_read_from_java_http_client_config(self):
        # 契约：Python 侧无 aiimage.http-client.* 读取点；超时仅来自参数/字面量
        source = inspect.getsource(admin_api._proxy_backend_java)
        self.assertNotIn("http-client", source)
        self.assertNotIn("AIIMAGE_HTTP_CLIENT", source)
        self.assertNotIn("connect-timeout", source)

    def test_timeout_or_connection_failure_maps_to_502(self):
        # 契约：Python→Java 超时/连接失败统一转 502，不静默吞掉也不自动重试
        fake = _FakeSession(raise_on_request=True)
        with patch.object(admin_api, "_get_backend_java_session", return_value=fake):
            with admin_api_app_context():
                result, error_response, status = admin_api._proxy_backend_java("GET", "/api/foo")
        self.assertIsNone(result)
        self.assertEqual(502, status)

    def test_java_config_namespace_absent_in_python_sources(self):
        # 快照：仓库非测试 Python 代码不存在 aiimage.http-client 配置引用，Java 治理不会外溢
        repo_py_root = Path(admin_api.__file__).resolve().parents[2]
        hits = []
        for py in repo_py_root.rglob("*.py"):
            if "__pycache__" in str(py) or "/tests/" in str(py).replace("\\", "/"):
                continue
            text = py.read_text(encoding="utf-8", errors="ignore")
            if "aiimage.http-client" in text or "aiimage_http_client" in text:
                hits.append(str(py))
        self.assertEqual([], hits, f"Python 侧不应引用 Java 统一命名空间: {hits}")

    def test_contract_frozen_documented_values_match_code(self):
        # 自检快照：文档化的 Python 回调契约（timeout=10 / max_retries=0）与代码一致
        self.assertEqual(10, self._proxy_default_http_timeout())
        session = admin_api._get_backend_java_session()
        self.assertEqual(0, session.get_adapter("http://").max_retries.total)
        self.assertEqual(0, session.get_adapter("https://").max_retries.total)


def admin_api_app_context():
    from flask import Flask
    app = Flask(__name__)
    return app.app_context()


if __name__ == "__main__":
    unittest.main()