feat: 保存并登记当前 Client (#11)

client/test/test_http_admin_gateway.py

@@ -0,0 +1,168 @@
+"""Admin 登记 HTTP 契约测试，不访问真实网络。"""
+
+import io
+import json
+import socket
+import unittest
+from http.client import RemoteDisconnected
+from urllib.error import HTTPError, URLError
+
+from src.admin_gateway import (
+    AdminGatewayError,
+    AndroidDeviceInfo,
+    ClaimCapabilities,
+    ClientInfo,
+)
+from src.http_admin_gateway import HttpAdminGateway
+
+
+class FakeResponse:
+    def __init__(self, status: int, payload):
+        self.status = status
+        self._body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+
+    def getcode(self):
+        return self.status
+
+    def read(self):
+        return self._body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+
+class RecordingOpener:
+    def __init__(self, response):
+        self.response = response
+        self.request = None
+        self.timeout = None
+
+    def __call__(self, request, timeout):
+        self.request = request
+        self.timeout = timeout
+        if isinstance(self.response, Exception):
+            raise self.response
+        return self.response
+
+
+class HttpAdminGatewayTest(unittest.TestCase):
+    def _capabilities(self, with_device=True):
+        device = (
+            AndroidDeviceInfo("192.168.0.173:5555") if with_device else None
+        )
+        return ClaimCapabilities(device=device)
+
+    def test_register_sends_confirmed_contract_and_parses_response(self):
+        opener = RecordingOpener(
+            FakeResponse(
+                200,
+                {
+                    "registered": True,
+                    "client_id": "CLIENT-001",
+                    "registered_at": "2026-08-06T10:00:00Z",
+                },
+            )
+        )
+        gateway = HttpAdminGateway(
+            "http://127.0.0.1:8080/", "secret-token", 2.5, opener
+        )
+
+        receipt = gateway.register_client(
+            ClientInfo("CLIENT-001", "办公室电脑"), self._capabilities()
+        )
+
+        self.assertTrue(receipt.registered)
+        self.assertEqual(opener.request.get_method(), "PUT")
+        self.assertEqual(
+            opener.request.full_url,
+            "http://127.0.0.1:8080/api/v1/client/registration",
+        )
+        headers = {
+            key.lower(): value for key, value in opener.request.header_items()
+        }
+        self.assertEqual(headers["x-client-id"], "CLIENT-001")
+        self.assertTrue(headers["x-request-id"])
+        self.assertEqual(headers["authorization"], "Bearer secret-token")
+        body = json.loads(opener.request.data.decode("utf-8"))
+        self.assertEqual(body["client"]["name"], "办公室电脑")
+        self.assertEqual(body["supported_types"], ["collect", "purchase"])
+        self.assertEqual(body["device"]["platform"], "android")
+        self.assertEqual(body["capabilities"]["purchase_mode"], "dry_run")
+        self.assertEqual(opener.timeout, 2.5)
+
+    def test_optional_device_is_omitted(self):
+        opener = RecordingOpener(
+            FakeResponse(
+                200,
+                {
+                    "registered": True,
+                    "client_id": "CLIENT-001",
+                    "registered_at": "2026-08-06T10:00:00Z",
+                },
+            )
+        )
+        gateway = HttpAdminGateway(opener=opener)
+
+        gateway.register_client(
+            ClientInfo("CLIENT-001"), self._capabilities(False)
+        )
+
+        body = json.loads(opener.request.data.decode("utf-8"))
+        self.assertNotIn("device", body)
+        self.assertNotIn("authorization", {
+            key.lower(): value for key, value in opener.request.header_items()
+        })
+
+    def test_admin_error_preserves_code_retry_and_request_id(self):
+        error_body = json.dumps(
+            {
+                "error": {
+                    "code": "INVALID_CLIENT_PROFILE",
+                    "message": "资料无效",
+                    "retryable": False,
+                    "request_id": "server-request-id",
+                    "details": {"field": "supported_types"},
+                }
+            }
+        ).encode("utf-8")
+        error = HTTPError(
+            "http://admin/api/v1/client/registration",
+            422,
+            "Unprocessable Entity",
+            {},
+            io.BytesIO(error_body),
+        )
+        gateway = HttpAdminGateway(opener=RecordingOpener(error))
+
+        with self.assertRaises(AdminGatewayError) as context:
+            gateway.register_client(
+                ClientInfo("CLIENT-001"), self._capabilities(False)
+            )
+
+        self.assertEqual(context.exception.code, "INVALID_CLIENT_PROFILE")
+        self.assertFalse(context.exception.retryable)
+        self.assertEqual(context.exception.request_id, "server-request-id")
+
+    def test_timeout_and_connection_failure_are_retryable(self):
+        for exception, code in (
+            (URLError(socket.timeout()), "ADMIN_TIMEOUT"),
+            (URLError("connection refused"), "ADMIN_UNAVAILABLE"),
+            (RemoteDisconnected("closed"), "ADMIN_UNAVAILABLE"),
+        ):
+            with self.subTest(code=code):
+                gateway = HttpAdminGateway(
+                    opener=RecordingOpener(exception)
+                )
+                with self.assertRaises(AdminGatewayError) as context:
+                    gateway.register_client(
+                        ClientInfo("CLIENT-001"), self._capabilities(False)
+                    )
+                self.assertEqual(context.exception.code, code)
+                self.assertTrue(context.exception.retryable)
+
+
+if __name__ == "__main__":
+    unittest.main()