| from typing import Annotated, Any |
|
|
| import pytest |
| from fastapi import Depends |
| from fastapi.testclient import TestClient |
| from pytest_benchmark.fixture import BenchmarkFixture |
|
|
|
|
| @pytest.fixture(scope="package") |
| def client(): |
| from hibiapi.app import app |
|
|
| with TestClient(app, base_url="http://testserver/") as client: |
| yield client |
|
|
|
|
| def test_openapi(client: TestClient, in_stress: bool = False): |
| response = client.get("/openapi.json") |
| assert response.status_code == 200 |
| assert response.json() |
|
|
| if in_stress: |
| return True |
|
|
|
|
| def test_doc_page(client: TestClient, in_stress: bool = False): |
| response = client.get("/docs") |
| assert response.status_code == 200 |
| assert response.text |
|
|
| response = client.get("/docs/test") |
| assert response.status_code == 200 |
| assert response.text |
|
|
| if in_stress: |
| return True |
|
|
|
|
| def test_openapi_stress(client: TestClient, benchmark: BenchmarkFixture): |
| assert benchmark.pedantic( |
| test_openapi, |
| args=(client, True), |
| rounds=200, |
| warmup_rounds=10, |
| iterations=3, |
| ) |
|
|
|
|
| def test_doc_page_stress(client: TestClient, benchmark: BenchmarkFixture): |
| assert benchmark.pedantic( |
| test_doc_page, args=(client, True), rounds=200, iterations=3 |
| ) |
|
|
|
|
| def test_notfound(client: TestClient): |
| from hibiapi.utils.exceptions import ExceptionReturn |
|
|
| response = client.get("/notexistpath") |
| assert response.status_code == 404 |
| assert ExceptionReturn.parse_obj(response.json()) |
|
|
|
|
| @pytest.mark.xfail(reason="not implemented yet") |
| def test_net_request(): |
| from hibiapi.utils.net import BaseNetClient |
| from hibiapi.utils.routing import BaseEndpoint, SlashRouter |
|
|
| test_headers = {"x-test-header": "random-string"} |
| test_data = {"test": "test"} |
|
|
| class TestEndpoint(BaseEndpoint): |
| base = "https://httpbin.org" |
|
|
| async def request(self, path: str, params: dict[str, Any]): |
| url = self._join(self.base, path, params) |
| response = await self.client.post(url, data=params) |
| response.raise_for_status() |
| return response.json() |
|
|
| async def form(self, *, data: dict[str, Any]): |
| return await self.request("/post", data) |
|
|
| async def teapot(self): |
| return await self.request("/status/{codes}", {"codes": 418}) |
|
|
| class TestNetClient(BaseNetClient): |
| pass |
|
|
| async def net_client(): |
| async with TestNetClient(headers=test_headers) as client: |
| yield TestEndpoint(client) |
|
|
| router = SlashRouter() |
|
|
| @router.post("form") |
| async def form( |
| *, |
| endpoint: Annotated[TestEndpoint, Depends(net_client)], |
| data: dict[str, Any], |
| ): |
| return await endpoint.form(data=data) |
|
|
| @router.post("teapot") |
| async def teapot(endpoint: Annotated[TestEndpoint, Depends(net_client)]): |
| return await endpoint.teapot() |
|
|
| from hibiapi.app.routes import router as api_router |
|
|
| api_router.include_router(router, prefix="/test") |
|
|
| from hibiapi.app import app |
| from hibiapi.utils.exceptions import ExceptionReturn |
|
|
| with TestClient(app, base_url="http://testserver/api/test/") as client: |
| response = client.post("form", json=test_data) |
| assert response.status_code == 200 |
| response_data = response.json() |
| assert response_data["form"] == test_data |
| request_headers = {k.lower(): v for k, v in response_data["headers"].items()} |
| assert test_headers.items() <= request_headers.items() |
|
|
| response = client.post("teapot", json=test_data) |
| exception_return = ExceptionReturn.parse_obj(response.json()) |
| assert exception_return.code == response.status_code |
|
|
