lami/lami-base-v1/backend/tests_python/client.py

"""Express PC API 测试客户端（统一响应契约）"""

from __future__ import annotations

from typing import Any

import requests


class ApiClient:
    def __init__(self, base_url: str):
        self.base = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})

    def url(self, path: str) -> str:
        if not path.startswith("/"):
            path = f"/{path}"
        return f"{self.base}{path}"

    def get(self, path: str, **kwargs: Any) -> requests.Response:
        return self.session.get(self.url(path), timeout=30, **kwargs)

    def post(self, path: str, json: Any = None, **kwargs: Any) -> requests.Response:
        return self.session.post(self.url(path), json=json, timeout=30, **kwargs)

    def put(self, path: str, json: Any = None, **kwargs: Any) -> requests.Response:
        return self.session.put(self.url(path), json=json, timeout=30, **kwargs)

    def delete(self, path: str, **kwargs: Any) -> requests.Response:
        return self.session.delete(self.url(path), timeout=30, **kwargs)

    @staticmethod
    def parse_json(resp: requests.Response) -> dict[str, Any]:
        return resp.json()

    def assert_success(self, resp: requests.Response, status: int = 200) -> Any:
        assert resp.status_code == status, resp.text
        body = self.parse_json(resp)
        assert body.get("success") is True, body
        assert body.get("error") is None, body
        return body.get("data")

    def assert_error(self, resp: requests.Response, status: int, code: str | None = None) -> dict:
        assert resp.status_code == status, resp.text
        body = self.parse_json(resp)
        assert body.get("success") is False, body
        err = body.get("error") or {}
        if code:
            assert err.get("code") == code, body
        return err