wren
135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReleaseTestConfig:
|
|
base_url: str
|
|
admin_username: str
|
|
admin_password: str
|
|
timeout_seconds: float
|
|
tenant_a_code: str
|
|
tenant_a_name: str
|
|
tenant_b_code: str
|
|
tenant_b_name: str
|
|
module_name: str
|
|
module_path: str
|
|
|
|
|
|
class ReleaseApiClient:
|
|
def __init__(self, config: ReleaseTestConfig, token: str | None = None) -> None:
|
|
self.config = config
|
|
self.token = token
|
|
self._client = httpx.Client(base_url=config.base_url, timeout=config.timeout_seconds)
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
def with_token(self, token: str) -> "ReleaseApiClient":
|
|
return ReleaseApiClient(self.config, token=token)
|
|
|
|
def request(self, method: str, path: str, *, expected_status: int | None = 200, **kwargs: Any) -> httpx.Response:
|
|
headers = dict(kwargs.pop("headers", {}) or {})
|
|
if self.token:
|
|
headers["Authorization"] = f"Bearer {self.token}"
|
|
response = self._client.request(method, path, headers=headers, **kwargs)
|
|
if expected_status is not None:
|
|
assert response.status_code == expected_status, self._format_error(response)
|
|
return response
|
|
|
|
def get(self, path: str, **kwargs: Any) -> httpx.Response:
|
|
return self.request("GET", path, **kwargs)
|
|
|
|
def post(self, path: str, **kwargs: Any) -> httpx.Response:
|
|
return self.request("POST", path, **kwargs)
|
|
|
|
def put(self, path: str, **kwargs: Any) -> httpx.Response:
|
|
return self.request("PUT", path, **kwargs)
|
|
|
|
def patch(self, path: str, **kwargs: Any) -> httpx.Response:
|
|
return self.request("PATCH", path, **kwargs)
|
|
|
|
def delete(self, path: str, **kwargs: Any) -> httpx.Response:
|
|
return self.request("DELETE", path, **kwargs)
|
|
|
|
def login_password(self, username: str, password: str) -> dict[str, Any]:
|
|
response = self.post(
|
|
"/api/auth/login",
|
|
json={"username": username, "password": password},
|
|
expected_status=200,
|
|
)
|
|
payload = response.json()
|
|
assert payload.get("success") is True, payload
|
|
data = payload.get("data") or {}
|
|
assert data.get("access_token"), payload
|
|
return data
|
|
|
|
def login_oauth(
|
|
self,
|
|
*,
|
|
sub: str,
|
|
username: str,
|
|
nickname: str,
|
|
ou_id: str,
|
|
ou_name: str,
|
|
area: str | None = None,
|
|
) -> dict[str, Any]:
|
|
response = self.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"userInfo": {
|
|
"sub": sub,
|
|
"username": username,
|
|
"nickname": nickname,
|
|
"email": f"{username}@pytest.local",
|
|
"phone_number": "13800000000",
|
|
"ou_id": ou_id,
|
|
"ou_name": ou_name,
|
|
"is_leader": False,
|
|
},
|
|
"area": area,
|
|
"expiresIn": 3600,
|
|
},
|
|
expected_status=200,
|
|
)
|
|
payload = response.json()
|
|
assert payload.get("success") is True, payload
|
|
data = payload.get("data") or {}
|
|
assert data.get("access_token"), payload
|
|
return data
|
|
|
|
@staticmethod
|
|
def json_data(response: httpx.Response) -> Any:
|
|
payload = response.json()
|
|
if isinstance(payload, dict) and "data" in payload:
|
|
return payload["data"]
|
|
return payload
|
|
|
|
@staticmethod
|
|
def _format_error(response: httpx.Response) -> str:
|
|
try:
|
|
body = response.json()
|
|
except Exception:
|
|
body = response.text
|
|
return f"{response.request.method} {response.request.url} -> {response.status_code}: {body}"
|
|
|
|
|
|
def flatten_route_paths(routes: list[dict[str, Any]]) -> set[str]:
|
|
paths: set[str] = set()
|
|
|
|
def collect(items: list[dict[str, Any]]) -> None:
|
|
for item in items:
|
|
route_path = str(item.get("route_path") or "")
|
|
if route_path:
|
|
paths.add(route_path)
|
|
children = item.get("children") or []
|
|
if isinstance(children, list):
|
|
collect(children)
|
|
|
|
collect(routes)
|
|
return paths
|