parent
bdba903db1
commit
5a433dfe41
2 changed files with 69 additions and 23 deletions
|
|
@ -1,7 +1,8 @@
|
|||
import pytest
|
||||
|
||||
from typing import AsyncGenerator
|
||||
from itertools import combinations
|
||||
|
||||
import pytest
|
||||
from fastapi.routing import APIRoute
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
|
|
@ -12,7 +13,6 @@ from src.contact.models import Contact
|
|||
from src.iam.models import Group, Permission
|
||||
from src.auth.service import get_current_user, get_dev_user
|
||||
from src.auth.dependencies import empty_su_list, get_super_admin_list, testing_su_list
|
||||
|
||||
from src.main import app # inited FastAPI app
|
||||
from src.database import engine, Base, get_db
|
||||
|
||||
|
|
@ -156,25 +156,22 @@ def generate_query_and_status(params) -> list[tuple[str, int]]:
|
|||
return query_and_status
|
||||
|
||||
|
||||
# # Produces a text file with method and path for every endpoint in the API
|
||||
# from fastapi.routing import APIRoute
|
||||
#
|
||||
# def get_testable_routes():
|
||||
# routes = []
|
||||
#
|
||||
# for route in app.routes:
|
||||
# if not isinstance(route, APIRoute):
|
||||
# continue
|
||||
#
|
||||
# for method in route.methods:
|
||||
# if method in {"HEAD", "OPTIONS"}:
|
||||
# continue
|
||||
#
|
||||
# routes.append((route.path, method))
|
||||
#
|
||||
# return routes
|
||||
#
|
||||
#
|
||||
def get_testable_routes():
|
||||
routes = []
|
||||
|
||||
for route in app.routes:
|
||||
if not isinstance(route, APIRoute):
|
||||
continue
|
||||
|
||||
for method in route.methods:
|
||||
if method in {"HEAD", "OPTIONS"}:
|
||||
continue
|
||||
|
||||
routes.append((method, route.path, route.status_code, route.response_model))
|
||||
|
||||
return routes
|
||||
|
||||
|
||||
# with open("endpoints.txt", "w") as f:
|
||||
# for ep in get_testable_routes():
|
||||
# f.write(f"{ep[1]} {ep[0]}\n")
|
||||
# f.write(f"[{ep[0]}]{ep[1]}({ep[2]}) -> {ep[2]}: {ep[3]}\n")
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@
|
|||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from fastapi.routing import APIRoute
|
||||
|
||||
from .conftest import generate_query_and_status
|
||||
|
||||
|
|
@ -143,3 +144,51 @@ async def test_get_self_orgs_success(default_client: AsyncClient):
|
|||
assert isinstance(org["security_contact"], dict)
|
||||
assert org["security_contact"]["email"] == "security@test.org"
|
||||
assert org["security_contact"]["id"] == 3
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_self_orgs_dynamic(default_client: AsyncClient):
|
||||
method = "GET"
|
||||
path = "/user/self/orgs"
|
||||
expected_data = {
|
||||
"organisations": [
|
||||
{
|
||||
"organisation_id": 1,
|
||||
"name": "Test Org",
|
||||
"status": "approved",
|
||||
"root_user_email": "admin@test.com",
|
||||
"owner_contact": {"email": "owner@test.org", "id": 2},
|
||||
"security_contact": {"email": "security@test.org", "id": 3},
|
||||
"billing_contact": {"email": "billing@test.org", "id": 1},
|
||||
"intake_questionnaire": {
|
||||
"question_one": None,
|
||||
"question_three": None,
|
||||
"question_two": "answer two",
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
resp = await default_client.get(path)
|
||||
|
||||
route = next(
|
||||
route
|
||||
for route in default_client._transport.app.routes
|
||||
if isinstance(route, APIRoute)
|
||||
and path in route.path
|
||||
and method in route.methods
|
||||
)
|
||||
|
||||
assert resp.status_code == route.status_code
|
||||
if route.status_code == 204:
|
||||
return
|
||||
|
||||
expected_response_schema = route.response_model
|
||||
data = resp.json()
|
||||
|
||||
response_model = expected_response_schema(**data)
|
||||
assert isinstance(response_model, expected_response_schema)
|
||||
|
||||
expected_response_model = expected_response_schema(**expected_data)
|
||||
|
||||
assert response_model == expected_response_model
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue