| from __future__ import annotations |
|
|
| import os |
| import sys |
| import tempfile |
| from pathlib import Path |
|
|
| from fastapi.testclient import TestClient |
|
|
| _repo_root = Path(__file__).resolve().parents[2] |
| if str(_repo_root) not in sys.path: |
| sys.path.insert(0, str(_repo_root)) |
|
|
| from Spider_XHS.service.app import create_app |
| from Spider_XHS.service.errors import ERROR_CODE_SUCCESS |
|
|
|
|
| def _assert(cond: bool, msg: str) -> None: |
| if not cond: |
| raise AssertionError(msg) |
|
|
|
|
| def _assert_wrapper(body: dict) -> None: |
| _assert(isinstance(body.get("code"), int), f"missing code: {body}") |
| _assert(isinstance(body.get("msg"), str), f"missing msg: {body}") |
| _assert("data" in body, f"missing data: {body}") |
| _assert(body.get("code") == ERROR_CODE_SUCCESS, f"unexpected code={body.get('code')}") |
|
|
|
|
| def _get(client: TestClient, path: str) -> dict: |
| resp = client.get(path) |
| _assert(resp.status_code == 200, f"expected 200, got={resp.status_code}, path={path}, body={resp.text}") |
| body = resp.json() |
| _assert(isinstance(body, dict), f"response not dict: {body}") |
| _assert_wrapper(body) |
| data = body.get("data") |
| _assert(isinstance(data, dict), f"data not dict: {data}") |
| return data |
|
|
|
|
| def verify_resources_api_v1() -> None: |
| old_env = dict(os.environ) |
| try: |
| os.environ["ENABLE_LEGACY_ROUTES"] = "0" |
| with tempfile.TemporaryDirectory() as td: |
| os.environ["STORAGE_ROOT"] = td |
| app = create_app() |
| with TestClient(app) as client: |
| accounts = _get(client, "/api/v1/resources/accounts") |
| _assert(isinstance(accounts.get("now"), str) and accounts.get("now"), "accounts.now missing") |
| _assert(isinstance(accounts.get("seed"), int), "accounts.seed missing") |
| acc_list = accounts.get("accounts") |
| _assert(isinstance(acc_list, list) and len(acc_list) >= 1, "accounts.accounts missing") |
| acc0 = acc_list[0] |
| _assert(isinstance(acc0, dict), "accounts.accounts[0] not dict") |
| for k in ( |
| "id", |
| "tags", |
| "risk_score", |
| "enabled", |
| "available", |
| "cooldown_remaining_s", |
| "cooldown_until", |
| "last_error_kind", |
| "last_error_at", |
| ): |
| _assert(k in acc0, f"account missing {k}") |
|
|
| sessions = _get(client, "/api/v1/resources/sessions") |
| _assert(isinstance(sessions.get("now"), str) and sessions.get("now"), "sessions.now missing") |
| sess_list = sessions.get("sessions") |
| _assert(isinstance(sess_list, list) and len(sess_list) >= 1, "sessions.sessions missing") |
| sess0 = sess_list[0] |
| _assert(isinstance(sess0, dict), "sessions.sessions[0] not dict") |
| for k in ( |
| "id", |
| "account_id", |
| "has_cookie", |
| "has_storage_state", |
| "cookie_ok", |
| "storage_state_ok", |
| "cookie_reason", |
| "storage_state_reason", |
| "checked_at", |
| ): |
| _assert(k in sess0, f"session missing {k}") |
|
|
| proxies = _get(client, "/api/v1/resources/proxies") |
| for k in ( |
| "available_count", |
| "avg_score", |
| "ejected_total", |
| "failures_total_by_reason", |
| "recent_failures_by_reason", |
| "last_fail_reasons_by_reason", |
| ): |
| _assert(k in proxies, f"proxies missing {k}") |
| _assert(isinstance(proxies.get("available_count"), int), "proxies.available_count not int") |
| _assert(isinstance(proxies.get("avg_score"), (int, float)), "proxies.avg_score not number") |
| _assert(isinstance(proxies.get("ejected_total"), int), "proxies.ejected_total not int") |
| _assert(isinstance(proxies.get("failures_total_by_reason"), dict), "proxies.failures_total_by_reason not dict") |
| _assert(isinstance(proxies.get("recent_failures_by_reason"), dict), "proxies.recent_failures_by_reason not dict") |
| _assert( |
| isinstance(proxies.get("last_fail_reasons_by_reason"), dict), |
| "proxies.last_fail_reasons_by_reason not dict", |
| ) |
| finally: |
| os.environ.clear() |
| os.environ.update(old_env) |
|
|
|
|
| def main() -> None: |
| verify_resources_api_v1() |
| print("ok") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
|
|
