Spaces:
Running
Running
File size: 3,713 Bytes
0a1b571 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 |
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
|