294 lines
9.6 KiB
Python
294 lines
9.6 KiB
Python
"""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_password(tmp_path: Path) -> None:
|
|
"""The fallback contains only an access token and an optional account name."""
|
|
|
|
path = tmp_path / "credentials.json"
|
|
store = TokenStore(path, keyring_backend=None)
|
|
store.save_token("token-value", account="doctor")
|
|
|
|
assert store.load_token() == "token-value"
|
|
assert store.load_account() == "doctor"
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
assert payload == {"token": "token-value", "account": "doctor"}
|
|
assert "password" not in path.read_text(encoding="utf-8").lower()
|
|
store.clear_token()
|
|
assert store.load_token() is None
|
|
assert store.load_account() == "doctor"
|
|
|
|
|
|
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_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"))
|