first commit

app/tests/test_api_client.py

@@ -0,0 +1,342 @@
+"""Contract tests for the UI-independent API client and token store."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+
+from doctor_workstation.core.errors import (
+    ApiBusinessError,
+    ApiError,
+    ApiProtocolError,
+    ApiTimeoutError,
+    AuthenticationExpiredError,
+    OpenPageRequiredError,
+    WorkWechatBindingRequiredError,
+)
+from doctor_workstation.services.api_client import ApiClient
+from doctor_workstation.services.token_store import TokenStore
+
+
+def test_get_normalises_adminapi_and_sends_contract_headers() -> None:
+    """The site base and already-prefixed base resolve to the same API URL."""
+
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json={"code": 1, "data": {"ok": True}})
+
+    with ApiClient(
+        "https://example.test/root/",
+        token="secret-token",
+        transport=httpx.MockTransport(handler),
+    ) as client:
+        assert client.get("/doctor.appointment/lists", {"page_no": 2}) == {"ok": True}
+
+    request = requests[0]
+    assert str(request.url) == (
+        "https://example.test/root/adminapi/doctor.appointment/lists?page_no=2"
+    )
+    assert request.headers["token"] == "secret-token"
+    assert request.headers["version"] == "1.9.4"
+    assert ApiClient.normalise_base_url("https://example.test/adminapi") == (
+        "https://example.test/adminapi/"
+    )
+
+
+def test_post_uses_json_and_never_retries_timeout() -> None:
+    """Writes use JSON and a timeout never causes an automatic duplicate POST."""
+
+    attempts = 0
+    bodies: list[dict[str, object]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        bodies.append(json.loads(request.content))
+        raise httpx.ReadTimeout("slow write", request=request)
+
+    client = ApiClient(
+        "https://example.test",
+        max_retries=5,
+        transport=httpx.MockTransport(handler),
+    )
+    with pytest.raises(ApiTimeoutError) as caught:
+        client.post("doctor.appointment/complete", {"id": 42})
+    client.close()
+
+    assert attempts == 1
+    assert bodies == [{"id": 42}]
+    assert caught.value.data["attempts"] == 1
+
+
+def test_multipart_post_lets_httpx_set_boundary_and_sends_form_fields(
+    tmp_path: Path,
+) -> None:
+    """Uploads use real multipart encoding without the JSON content type."""
+
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(
+            200,
+            json={"code": 1, "data": {"uri": "/uploads/demo.jpg"}},
+        )
+
+    source = tmp_path / "demo.jpg"
+    source.write_bytes(b"jpeg-demo-bytes")
+    with (
+        ApiClient(
+            "https://example.test",
+            transport=httpx.MockTransport(handler),
+        ) as client,
+        source.open("rb") as stream,
+    ):
+        result = client.post_multipart(
+            "upload/image",
+            files={"file": (source.name, stream, "image/jpeg")},
+            data={"cid": "0"},
+        )
+
+    assert result == {"uri": "/uploads/demo.jpg"}
+    request = requests[0]
+    content_type = request.headers["content-type"]
+    assert content_type.startswith("multipart/form-data; boundary=")
+    assert "application/json" not in content_type
+    assert b'name="file"; filename="demo.jpg"' in request.content
+    assert b'name="cid"' in request.content
+    assert b"jpeg-demo-bytes" in request.content
+
+
+def test_get_retries_only_timeouts_then_returns_data() -> None:
+    """A GET may recover from a bounded number of timeout failures."""
+
+    attempts = 0
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        if attempts < 3:
+            raise httpx.ReadTimeout("temporary", request=request)
+        return httpx.Response(200, json={"code": "1", "data": ["ready"]})
+
+    with ApiClient(
+        "https://example.test/adminapi/",
+        max_retries=2,
+        transport=httpx.MockTransport(handler),
+    ) as client:
+        assert client.get("health") == ["ready"]
+    assert attempts == 3
+
+
+def test_get_bytes_downloads_relative_public_image_without_api_token() -> None:
+    """Generated QR images bypass the JSON envelope and never leak the API token."""
+
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, headers={"content-type": "image/png"}, content=b"png-data")
+
+    with ApiClient(
+        "https://example.test/root",
+        token="private-token",
+        transport=httpx.MockTransport(handler),
+    ) as client:
+        assert client.get_bytes("/uploads/qrcode.png") == b"png-data"
+
+    request = requests[0]
+    assert str(request.url) == "https://example.test/uploads/qrcode.png"
+    assert "token" not in request.headers
+
+
+@pytest.mark.parametrize(
+    ("code", "exception_type"),
+    [
+        (0, ApiBusinessError),
+        (-1, AuthenticationExpiredError),
+        (10, WorkWechatBindingRequiredError),
+    ],
+)
+def test_envelope_error_codes_are_structured(code: int, exception_type: type[Exception]) -> None:
+    """Known control-flow codes become typed exceptions with response data."""
+
+    transport = httpx.MockTransport(
+        lambda request: httpx.Response(
+            200,
+            headers={"x-request-id": "req-123"},
+            json={"code": code, "msg": "action needed", "data": {"reason": "demo"}},
+        )
+    )
+    with (
+        ApiClient("https://example.test", transport=transport) as client,
+        pytest.raises(exception_type) as caught,
+    ):
+        client.get("auth.admin/mySelf")
+    error = caught.value
+    assert isinstance(error, ApiError)
+    assert error.code == code
+    assert error.data == {"reason": "demo"}
+    assert error.request_id == "req-123"
+
+
+def test_open_page_signal_does_not_open_a_browser() -> None:
+    """Code 2 is surfaced to the UI as data, not executed by the service layer."""
+
+    transport = httpx.MockTransport(
+        lambda request: httpx.Response(
+            200,
+            json={"code": 2, "data": {"url": "https://example.test/continue"}},
+        )
+    )
+    with (
+        ApiClient("https://example.test", transport=transport) as client,
+        pytest.raises(OpenPageRequiredError) as caught,
+    ):
+        client.get("continue")
+    assert caught.value.url == "https://example.test/continue"
+
+
+def test_invalid_envelope_raises_protocol_error() -> None:
+    """Successful HTTP is not mistaken for API success without a valid envelope."""
+
+    transport = httpx.MockTransport(
+        lambda request: httpx.Response(200, json={"data": "missing code"})
+    )
+    with (
+        ApiClient("https://example.test", transport=transport) as client,
+        pytest.raises(ApiProtocolError),
+    ):
+        client.get("broken")
+
+
+def test_token_store_file_fallback_never_persists_plaintext_password(tmp_path: Path) -> None:
+    """The fallback may contain a Windows DPAPI blob, but never plaintext."""
+
+    path = tmp_path / "credentials.json"
+    store = TokenStore(path, keyring_backend=None)
+    store.save_token("token-value", account="doctor")
+    password_saved = store.save_password(
+        "must-not-reach-disk",
+        account="doctor",
+        scope="https://example.test/adminapi",
+    )
+
+    assert store.load_token() == "token-value"
+    assert store.load_account() == "doctor"
+    restored = store.load_password(account="doctor", scope="https://example.test/adminapi")
+    if password_saved:
+        assert restored == "must-not-reach-disk"
+    else:
+        assert restored is None
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    assert payload["token"] == "token-value"
+    assert payload["account"] == "doctor"
+    assert "password" not in path.read_text(encoding="utf-8").lower()
+    assert "must-not-reach-disk" not in path.read_text(encoding="utf-8")
+    store.clear_token()
+    assert store.load_token() is None
+    assert store.load_account() == "doctor"
+    after_logout = store.load_password(
+        account="doctor",
+        scope="https://example.test/adminapi",
+    )
+    if password_saved:
+        assert after_logout == "must-not-reach-disk"
+    else:
+        assert after_logout is None
+    store.clear_account()
+    assert store.load_password(account="doctor", scope="https://example.test/adminapi") is None
+
+
+class _MemoryKeyring:
+    """Minimal deterministic keyring double."""
+
+    def __init__(self) -> None:
+        self.values: dict[tuple[str, str], str] = {}
+
+    def get_password(self, service: str, username: str) -> str | None:
+        """Return an in-memory secret."""
+
+        return self.values.get((service, username))
+
+    def set_password(self, service: str, username: str, password: str) -> None:
+        """Store an in-memory secret."""
+
+        self.values[(service, username)] = password
+
+    def delete_password(self, service: str, username: str) -> None:
+        """Delete an in-memory secret."""
+
+        self.values.pop((service, username), None)
+
+
+def test_token_store_prefers_available_keyring(tmp_path: Path) -> None:
+    """A working keyring keeps the token out of the fallback JSON file."""
+
+    backend = _MemoryKeyring()
+    path = tmp_path / "credentials.json"
+    store = TokenStore(path, keyring_backend=backend)
+    store.save_token("keyring-token", account="doctor")
+
+    assert store.uses_keyring
+    assert store.load_token() == "keyring-token"
+    assert json.loads(path.read_text(encoding="utf-8")) == {"account": "doctor"}
+
+
+def test_token_store_keeps_login_password_in_scoped_keyring_only(tmp_path: Path) -> None:
+    backend = _MemoryKeyring()
+    path = tmp_path / "credentials.json"
+    store = TokenStore(path, keyring_backend=backend)
+    scope = "https://example.test/adminapi/"
+
+    assert store.save_password("secret-value", account="doctor", scope=scope)
+    assert (
+        store.load_password(account="doctor", scope="https://example.test/adminapi")
+        == "secret-value"
+    )
+    assert store.load_password(account="doctor", scope="https://other.test/adminapi") is None
+    assert "secret-value" not in path.read_text(encoding="utf-8")
+    assert json.loads(path.read_text(encoding="utf-8")) == {
+        "account": "doctor",
+        "scope": "https://example.test/adminapi",
+    }
+
+    next_scope = "https://next.test/adminapi"
+    assert store.save_password("next-secret", account="doctor", scope=next_scope)
+    assert store.load_password(account="doctor", scope=scope) is None
+    assert store.load_password(account="doctor", scope=next_scope) == "next-secret"
+
+    store.clear_password(account="doctor", scope=next_scope)
+    assert store.load_password(account="doctor", scope=next_scope) is None
+
+
+def test_token_store_scopes_automatic_restore_and_forgets_account(tmp_path: Path) -> None:
+    """Automatic restore never returns a token issued for another API base."""
+
+    path = tmp_path / "credentials.json"
+    store = TokenStore(path, keyring_backend=None)
+    api_scope = "https://example.test/adminapi/"
+    store.save_token(
+        "scoped-token",
+        account="doctor",
+        scope=api_scope,
+    )
+
+    assert store.load_token(scope="https://example.test/adminapi") == "scoped-token"
+    assert store.load_token(scope="https://other.test/adminapi/") is None
+    assert store.load_token() == "scoped-token"
+    assert json.loads(path.read_text(encoding="utf-8")) == {
+        "token": "scoped-token",
+        "account": "doctor",
+        "scope": "https://example.test/adminapi",
+    }
+
+    store.save_token("next-token", account="", scope=api_scope)
+    assert store.load_account() is None
+    assert "account" not in json.loads(path.read_text(encoding="utf-8"))