From 9102f3e669ffd15811ce37c669a51e5bad2d2f29 Mon Sep 17 00:00:00 2001 From: Renovate Date: Mon, 22 Jun 2026 15:54:45 +0000 Subject: [PATCH 01/11] chore(deps): update dependency starlette to v1.3.1 [security] --- uv.lock | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/uv.lock b/uv.lock index 147ce12..130b316 100644 --- a/uv.lock +++ b/uv.lock @@ -931,15 +931,15 @@ wheels = [ [[package]] name = "starlette" -version = "1.1.0" +version = "1.2.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/95/66/4d20cdf39a8d6a51e663b7038e3b828ff211d3891a43a713fe7e4643f3a8/starlette-1.1.0.tar.gz", hash = "sha256:e83c7fe0ddecd8719c5b840080325aec0260acec86e9832899e377b91d65e90f", size = 2660060, upload-time = "2026-05-23T16:55:41.376Z" } +sdist = { url = "https://files.pythonhosted.org/packages/25/44/ec35f1b6e83094b997da438a02c8c9b0ade2b1e84cfc48bd4656780760a6/starlette-1.2.1.tar.gz", hash = "sha256:9b9b5ebb992e67d6093741e63c2f59e4f6fff986f81163c087867bd7b924b3f6", size = 2701854, upload-time = "2026-05-31T01:07:51.847Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/93/79/920b8e0a8b20f793e8d64855095cb8febabf6175b8550b6f7a547d813891/starlette-1.1.0-py3-none-any.whl", hash = "sha256:7f0dfd38e428aad5cb6f9f667f0ca1d2d8ca3f3385dccac8305f79ec98458382", size = 72899, upload-time = "2026-05-23T16:55:39.201Z" }, + { url = "https://files.pythonhosted.org/packages/1c/54/196d0c1db10af76baa4f64894448505d60d3cdf70ef92cbb35f46a4e4c71/starlette-1.2.1-py3-none-any.whl", hash = "sha256:4de0082d08c8f6764a85a54cf1120d6939507a19905c7768acad2a9f875d2b89", size = 73350, upload-time = "2026-05-31T01:07:50.09Z" }, ] [[package]] From 1e438244aa37c472868f914858605f142f1f362b Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 09:47:26 +0100 Subject: [PATCH 02/11] tests: route_data fixture Gets the actual auth level, expected status code, and expected response model from the route definitions. This can be used to make happy-path tests a standard shape, and with better coverage. Auth tests can be directly incorporated into this test shape. --- test/conftest.py | 52 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/test/conftest.py b/test/conftest.py index ffe4065..a97e092 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -256,6 +256,58 @@ def generate_body_and_status(params: dict[str, str]) -> list[tuple[dict, int]]: return body_and_status +@pytest.fixture +async def route_data(): + routes: dict[str, dict] = {} + + contexts = list(iter_route_contexts(app.routes)) + + for route in contexts: + if not route.methods: + continue + if not isinstance(route.route, APIRoute): + continue + + dep_func_names = set() + + unchecked = [route.route.dependant] + while unchecked: + dependant = unchecked.pop(0) + ck = dependant.cache_key[0] + if hasattr(ck, "__name__"): + dep_func_names.add(ck.__name__) + unchecked += [ + dep for dep in dependant.dependencies if isinstance(dep, Dependant) + ] + + auth_level = None + if "get_current_user" in dep_func_names: + auth_level = "User" + if ( + "org_body_root_claims" in dep_func_names + or "org_query_root_claims" in dep_func_names + ): + auth_level = "Root User" + if "user_model_super_admin" in dep_func_names: + auth_level = "Super Admin" + if "valid_service_key" in dep_func_names: + auth_level = "API Key" + + for method in route.methods: + if method in {"HEAD", "OPTIONS"}: + continue + + route_key = f"{method.upper()}{route.route.path}" + + routes[route_key] = { + "status_code": route.route.status_code, + "response_model": route.route.response_model, + "auth_level": auth_level, + } + + return routes + + def get_testable_routes(): routes = [] From ed01e2515f0e09bc08fc06c19d3a2362bd850e11 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 10:10:29 +0100 Subject: [PATCH 03/11] tests: dynamic test using new fixture --- test/test_user.py | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/test/test_user.py b/test/test_user.py index a1f693c..84b7bf1 100644 --- a/test/test_user.py +++ b/test/test_user.py @@ -1,11 +1,12 @@ """ [GET]/user/self/claims is not tested because it requires OIDC authentication. -[DELETE/user/ is not tested because the testing client cannot attach a body to a delete request. +[DELETE]/user/ is not tested because the testing client cannot attach a body to a delete request. """ +from typing import Any + import pytest from httpx import AsyncClient -from fastapi.routing import APIRoute, iter_route_contexts from .conftest import generate_query_and_status @@ -153,9 +154,13 @@ async def test_get_self_orgs_success(default_client: AsyncClient): @pytest.mark.anyio -async def test_get_self_orgs_dynamic(default_client: AsyncClient): +async def test_get_self_orgs_dynamic( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): method = "GET" path = "/user/self/orgs" + auth_level = "User" + query = "" expected_data = { "organisations": [ { @@ -178,24 +183,19 @@ async def test_get_self_orgs_dynamic(default_client: AsyncClient): ] } - resp = await default_client.get(path) + # req_func = getattr(default_client, method); resp = await req_func(url=path, json=body) + resp = await default_client.get(f"{path}{query}") - contexts = list(iter_route_contexts(default_client._transport.app.routes)) # ty:ignore[unresolved-attribute] + route = route_data.get(f"{method.upper()}{path}") + assert route is not None - route = next( - route.route - for route in contexts - if isinstance(route.route, APIRoute) - and path in route.route.path - and isinstance(route.methods, set) - and method in route.methods - ) + assert route["auth_level"] == auth_level - assert resp.status_code == route.status_code - if route.status_code == 204: + assert resp.status_code == route["status_code"] + if route["status_code"] == 204: return - expected_response_schema = route.response_model + expected_response_schema = route["response_model"] data = resp.json() response_model = expected_response_schema(**data) From 41df580a1aab2b6a87f283e312b96265b8aa450e Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 10:26:28 +0100 Subject: [PATCH 04/11] tests: dynamic test standardised The dynamic test structure should now be able applicable to all endpoints and the bulk of its logic has been split into a new function. --- test/conftest.py | 38 +++++++++++++++++++++++++++++++++++++- test/test_user.py | 30 ++++++------------------------ 2 files changed, 43 insertions(+), 25 deletions(-) diff --git a/test/conftest.py b/test/conftest.py index a97e092..aa204c4 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,7 +1,7 @@ from fastapi.dependencies.models import Dependant import pytest -from typing import AsyncGenerator +from typing import AsyncGenerator, Any from itertools import combinations from fastapi.routing import APIRoute, iter_route_contexts from httpx import AsyncClient, ASGITransport @@ -308,6 +308,42 @@ async def route_data(): return routes +async def standard_test( + client: AsyncClient, + method: str, + path: str, + auth_level: str, + query: str, + body: dict, + expected_data: dict, + route_data: dict[str, dict[str, Any]], +): + route = route_data.get(f"{method.upper()}{path}") + assert route is not None + + req_func = getattr(client, method.lower()) + if method in ["GET", "DELETE"]: + resp = await req_func(url=f"{path}{query}") + else: + resp = await req_func(url=f"{path}{query}", json=body) + + assert route["auth_level"] == auth_level + + 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 + + def get_testable_routes(): routes = [] diff --git a/test/test_user.py b/test/test_user.py index 84b7bf1..1d66f16 100644 --- a/test/test_user.py +++ b/test/test_user.py @@ -8,8 +8,7 @@ from typing import Any import pytest from httpx import AsyncClient -from .conftest import generate_query_and_status - +from .conftest import generate_query_and_status, standard_test pytestmark = [ pytest.mark.user_module, @@ -154,13 +153,14 @@ async def test_get_self_orgs_success(default_client: AsyncClient): @pytest.mark.anyio -async def test_get_self_orgs_dynamic( +async def test_get_self_orgs_standard( default_client: AsyncClient, route_data: dict[str, dict[str, Any]] ): method = "GET" path = "/user/self/orgs" auth_level = "User" query = "" + body = {} expected_data = { "organisations": [ { @@ -183,24 +183,6 @@ async def test_get_self_orgs_dynamic( ] } - # req_func = getattr(default_client, method); resp = await req_func(url=path, json=body) - resp = await default_client.get(f"{path}{query}") - - route = route_data.get(f"{method.upper()}{path}") - assert route is not None - - assert route["auth_level"] == auth_level - - 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 + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) From 28c482d8a9c474377d00487444ab4c82ae1847e0 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 10:45:17 +0100 Subject: [PATCH 05/11] tests: user tests use standard model All happy-path tests for the user module have been replaced with the standard test model. --- test/test_user.py | 167 ++++++++++++++++++++++------------------------ 1 file changed, 81 insertions(+), 86 deletions(-) diff --git a/test/test_user.py b/test/test_user.py index 1d66f16..178b1fe 100644 --- a/test/test_user.py +++ b/test/test_user.py @@ -15,36 +15,6 @@ pytestmark = [ ] -@pytest.mark.anyio -async def test_get_self_db_success(default_client: AsyncClient): - resp = await default_client.get("/user/self/db") - data = resp.json() - - assert resp.status_code == 200 - assert data["first_name"] == "Admin" - assert data["last_name"] == "Test" - assert data["email"] == "admin@test.com" - assert "organisations" in data - assert isinstance(data["organisations"], list) - assert "groups" in data - assert isinstance(data["groups"], dict) - - -@pytest.mark.anyio -async def test_get_user_success(default_client: AsyncClient): - resp = await default_client.get("/user?user_id=1") - data = resp.json() - - assert resp.status_code == 200 - assert data["first_name"] == "Admin" - assert data["last_name"] == "Test" - assert data["email"] == "admin@test.com" - assert "organisations" in data - assert isinstance(data["organisations"], list) - assert "groups" in data - assert isinstance(data["groups"], dict) - - @pytest.mark.anyio @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["user_id"])) async def test_get_user_status_checks( @@ -55,30 +25,6 @@ async def test_get_user_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_delete_user_success(default_client: AsyncClient): - resp = await default_client.delete("/user?user_id=1") - - assert resp.status_code == 204 - - -@pytest.mark.anyio -async def test_post_user_invitation_success(default_client: AsyncClient): - body = {"user_email": "admin@test.com", "organisation_id": 1} - resp = await default_client.post("/user/invitation", json=body) - - assert resp.status_code == 200 - data = resp.json() - assert "organisation" in data - assert isinstance(data["organisation"], dict) - assert data["organisation"]["id"] == 1 - assert data["organisation"]["name"] == "Org One" - - assert "invited_email" in data - assert isinstance(data["invited_email"], str) - assert data["invited_email"] == "admin@test.com" - - @pytest.mark.parametrize( "body, expected_status", [ @@ -120,38 +66,6 @@ async def test_post_user_invitation_accept_status_checks( assert resp.json()["detail"] == "Invalid JWS" -@pytest.mark.anyio -async def test_get_self_orgs_success(default_client: AsyncClient): - resp = await default_client.get("/user/self/orgs") - assert resp.status_code == 200 - - data = resp.json() - - assert "organisations" in data - assert isinstance(data["organisations"], list) - assert len(data["organisations"]) > 0 - - org = data["organisations"][0] - assert org["organisation_id"] == 1 - assert org["name"] == "Org One" - assert org["status"] == "approved" - assert org["root_user_email"] == "admin@test.com" - assert "intake_questionnaire" in org - assert isinstance(org["intake_questionnaire"], dict) - - assert isinstance(org["billing_contact"], dict) - assert org["billing_contact"]["email"] == "billing@orgone.com" - assert org["billing_contact"]["id"] == 1 - - assert isinstance(org["owner_contact"], dict) - assert org["owner_contact"]["email"] == "owner@orgone.com" - assert org["owner_contact"]["id"] == 2 - - assert isinstance(org["security_contact"], dict) - assert org["security_contact"]["email"] == "security@orgone.com" - assert org["security_contact"]["id"] == 3 - - @pytest.mark.anyio async def test_get_self_orgs_standard( default_client: AsyncClient, route_data: dict[str, dict[str, Any]] @@ -186,3 +100,84 @@ async def test_get_self_orgs_standard( await standard_test( default_client, method, path, auth_level, query, body, expected_data, route_data ) + + +@pytest.mark.anyio +async def test_get_self_db_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/user/self/db" + auth_level = "User" + query = "" + body = {} + expected_data = { + "id": 1, + "first_name": "Admin", + "last_name": "Test", + "email": "admin@test.com", + "organisations": [{"name": "Org One", "id": 1}], + "groups": {"Org One": [{"name": "Org One Group", "id": 1}]}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_get_user_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/user" + auth_level = "Super Admin" + query = "?user_id=1" + body = {} + expected_data = { + "id": 1, + "first_name": "Admin", + "last_name": "Test", + "email": "admin@test.com", + "organisations": [{"name": "Org One", "id": 1}], + "groups": {"Org One": [{"name": "Org One Group", "id": 1}]}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_user_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/user" + auth_level = "Super Admin" + query = "?user_id=1" + body = {} + expected_data = {} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_post_user_invitation_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "POST" + path = "/user/invitation" + auth_level = "Root User" + query = "" + body = {"user_email": "admin@test.com", "organisation_id": 1} + expected_data = { + "invited_email": "admin@test.com", + "organisation": {"name": "Org One", "id": 1}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) From ce6a5749513a18d79f24b05a0bab14c6194e7fd8 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 13:20:06 +0100 Subject: [PATCH 06/11] tests: user accept org invite test --- test/test_user.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/test/test_user.py b/test/test_user.py index 178b1fe..9b3ef46 100644 --- a/test/test_user.py +++ b/test/test_user.py @@ -3,11 +3,13 @@ [DELETE]/user/ is not tested because the testing client cannot attach a body to a delete request. """ +from datetime import timedelta, datetime, timezone from typing import Any import pytest from httpx import AsyncClient +from src.utils import generate_jwt from .conftest import generate_query_and_status, standard_test pytestmark = [ @@ -181,3 +183,33 @@ async def test_post_user_invitation_standard( await standard_test( default_client, method, path, auth_level, query, body, expected_data, route_data ) + + +@pytest.mark.anyio +async def test_post_user_invitation_accept_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + expiry_delta = timedelta(hours=24) + expiry = datetime.now(timezone.utc) + expiry_delta + claims = { + "email": "admin@test.com", + "org_id": 2, + "exp": expiry, + "type": "org_invite", + } + + token = await generate_jwt(claims) + + method = "POST" + path = "/user/invitation/accept" + auth_level = "User" + query = "" + body = {"jwt": token} + expected_data = { + "organisation": {"name": "Org Two", "id": 2}, + "user": {"email": "admin@test.com", "id": 1}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) From 71adc3fc6a4dd8a0476aa89542292402acf188d0 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 13:27:04 +0100 Subject: [PATCH 07/11] tests: org tests standardised --- test/test_organisation.py | 354 ++++++++++++++++++++++---------------- 1 file changed, 210 insertions(+), 144 deletions(-) diff --git a/test/test_organisation.py b/test/test_organisation.py index 49dd0a3..dfa8c50 100644 --- a/test/test_organisation.py +++ b/test/test_organisation.py @@ -2,10 +2,12 @@ [DELETE] /org/ is not tested because the testing client cannot attach a body to a delete request. """ +from typing import Any + import pytest from httpx import AsyncClient -from .conftest import generate_query_and_status +from .conftest import generate_query_and_status, standard_test pytestmark = [ @@ -13,28 +15,6 @@ pytestmark = [ ] -@pytest.mark.anyio -async def test_get_org_success(default_client: AsyncClient): - resp = await default_client.get("/org?org_id=1") - data = resp.json() - - assert resp.status_code == 200 - - org = data["organisations"][0] - - assert isinstance(org, dict) - assert org["organisation_id"] == 1 - assert org["name"] == "Org One" - assert org["status"] == "approved" - assert org["root_user_email"] == "admin@test.com" - assert "intake_questionnaire" in org - assert isinstance(org["intake_questionnaire"], dict) - - assert org["billing_contact"]["email"] == "billing@orgone.com" - assert org["owner_contact"]["email"] == "owner@orgone.com" - assert org["security_contact"]["email"] == "security@orgone.com" - - @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.anyio async def test_get_org_status_checks( @@ -45,16 +25,6 @@ async def test_get_org_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_post_org_success(default_client: AsyncClient): - resp = await default_client.post("/org", json={"name": "New Test Org"}) - data = resp.json() - - assert resp.status_code == 201 - assert data["name"] == "New Test Org" - assert data["status"] == "partial" - - @pytest.mark.parametrize( "body, expected_status", [ @@ -72,36 +42,6 @@ async def test_post_org_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient): - resp = await default_client.patch( - "/org/questionnaire", - json={ - "organisation_id": 3, - "intake_questionnaire": { - "question_one": "new answer one", - "question_two": None, - "question_three": None, - }, - "partial": True, - }, - ) - data = resp.json() - - assert resp.status_code == 200 - assert data["name"] == "Org Three" - assert data["status"] == "partial" - assert "intake_questionnaire" in data - assert isinstance(data["intake_questionnaire"], dict) - metadata = data["intake_questionnaire"]["metadata"] - assert metadata["version"] == 0 - assert metadata["submission_date"] is None - questions = data["intake_questionnaire"]["questions"] - assert questions["question_one"] == "new answer one" - assert questions["question_two"] == "answer two" - assert questions["question_three"] is None - - @pytest.mark.parametrize( "body, expected_status", [ @@ -205,27 +145,6 @@ async def test_patch_org_status_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_get_org_users_success(default_client: AsyncClient): - resp = await default_client.get("/org/users?org_id=1") - data = resp.json() - - assert resp.status_code == 200 - - assert "users" in data - assert isinstance(data["users"], list) - assert len(data["users"]) == 2 - - user = data["users"][0] - assert isinstance(user, dict) - assert user["email"] == "admin@test.com" - assert user["id"] == 1 - - assert "organisation" in data - assert data["organisation"]["name"] == "Org One" - assert data["organisation"]["id"] == 1 - - @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.anyio async def test_get_org_users_status_checks( @@ -236,24 +155,6 @@ async def test_get_org_users_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_post_org_user_success(default_client: AsyncClient): - resp = await default_client.post("/org/user", json={"organisation_id": 1, "user_id": 3}) - - assert resp.status_code == 200 - - data = resp.json() - - assert "organisation" in data - assert isinstance(data["organisation"], dict) - assert data["organisation"]["id"] == 1 - assert data["organisation"]["name"] == "Org One" - - assert "users" in data - assert isinstance(data["users"], list) - assert len([user for user in data["users"] if user["email"] == "root@orgtwo.com"]) == 1 - - @pytest.mark.parametrize( "body, expected_status", [ @@ -274,19 +175,6 @@ async def test_post_org_user_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_patch_org_root_user_success(default_client: AsyncClient): - resp = await default_client.patch( - "/org/root_user", json={"organisation_id": 1, "user_id": 2} - ) - assert resp.status_code == 200 - - data = resp.json() - - assert data["name"] == "Org One" - assert data["root_user_email"] == "user@orgone.com" - - @pytest.mark.parametrize( "body, expected_status", [ @@ -319,26 +207,6 @@ async def test_patch_org_root_user_non_member(default_client: AsyncClient): assert data["detail"] == "This user does not belong to your organisation." -@pytest.mark.anyio -async def test_get_org_groups_success(default_client: AsyncClient): - resp = await default_client.get("/org/groups?org_id=1") - assert resp.status_code == 200 - - data = resp.json() - - assert "organisation" in data - assert isinstance(data["organisation"], dict) - assert data["organisation"]["id"] == 1 - assert data["organisation"]["name"] == "Org One" - - assert "groups" in data - assert isinstance(data["groups"], list) - group = data["groups"][0] - assert isinstance(group, dict) - assert group["id"] == 1 - assert group["name"] == "Org One Group" - - @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.anyio async def test_get_org_groups_status_checks( @@ -494,21 +362,219 @@ async def test_patch_org_contact_status_checks( @pytest.mark.anyio -async def test_delete_org_success(default_client: AsyncClient): - resp = await default_client.delete("/org?org_id=1") +async def test_get_org_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/org" + auth_level = "Root User" + query = "?org_id=1" + body = {} + expected_data = { + "organisations": [ + { + "organisation_id": 1, + "name": "Org One", + "status": "approved", + "root_user_email": "admin@test.com", + "intake_questionnaire": { + "metadata": {"version": 0, "submission_date": None}, + "questions": { + "question_one": None, + "question_two": "answer two", + "question_three": None, + }, + }, + "billing_contact": {"id": 1, "email": "billing@orgone.com"}, + "owner_contact": {"id": 2, "email": "owner@orgone.com"}, + "security_contact": {"id": 3, "email": "security@orgone.com"}, + } + ] + } - assert resp.status_code == 204 + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) @pytest.mark.anyio -async def test_delete_org_users_success(default_client: AsyncClient): - resp = await default_client.delete("/org/user?org_id=1&user_id=2") +async def test_post_org_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "POST" + path = "/org" + auth_level = "User" + query = "" + body = {"name": "New Test Org"} + expected_data = {"id": 4, "name": "New Test Org", "status": "partial"} - assert resp.status_code == 204 + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) @pytest.mark.anyio -async def test_delete_preapproval_org_success(default_client: AsyncClient): - resp = await default_client.delete("/org/self?org_id=3") +async def test_patch_org_questionnaire_partial_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "PATCH" + path = "/org/questionnaire" + auth_level = "Root User" + query = "" + body = { + "organisation_id": 3, + "intake_questionnaire": { + "question_one": "new answer one", + "question_two": None, + "question_three": None, + }, + "partial": True, + } + expected_data = { + "id": 3, + "name": "Org Three", + "status": "partial", + "intake_questionnaire": { + "metadata": {"version": 0, "submission_date": None}, + "questions": { + "question_one": "new answer one", + "question_two": "answer two", + "question_three": None, + }, + }, + } - assert resp.status_code == 204 + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_get_org_users_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/org/users" + auth_level = "Root User" + query = "?org_id=1" + body = {} + expected_data = { + "users": [ + {"email": "admin@test.com", "id": 1}, + {"email": "user@orgone.com", "id": 2}, + ], + "organisation": {"id": 1, "name": "Org One"}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_post_org_user_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "POST" + path = "/org/user" + auth_level = "Super Admin" + query = "" + body = {"organisation_id": 1, "user_id": 3} + expected_data = { + "users": [ + {"email": "admin@test.com", "id": 1}, + {"email": "user@orgone.com", "id": 2}, + {"email": "root@orgtwo.com", "id": 3}, + ], + "organisation": {"id": 1, "name": "Org One"}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_patch_org_root_user_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "PATCH" + path = "/org/root_user" + auth_level = "Super Admin" + query = "" + body = {"organisation_id": 1, "user_id": 2} + expected_data = {"name": "Org One", "root_user_email": "user@orgone.com"} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_get_org_groups_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/org/groups" + auth_level = "Root User" + query = "?org_id=1" + body = {} + expected_data = { + "groups": [ + {"name": "Org One Group", "id": 1}, + {"name": "Org One Group Two", "id": 3}, + ], + "organisation": {"id": 1, "name": "Org One"}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_org_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/org" + auth_level = "Super Admin" + query = "?org_id=1" + body = {} + expected_data = {} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_org_users_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/org/user" + auth_level = "Root User" + query = "?org_id=1&user_id=2" + body = {} + expected_data = {} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_preapproval_org_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/org/self" + auth_level = "Root User" + query = "?org_id=3" + body = {} + expected_data = {} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) From 1c394e9e2390371511603a59c8f84b2de7ccb420 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 13:50:50 +0100 Subject: [PATCH 08/11] tests: service tests standardised --- test/test_service.py | 51 ++++++++++++++++++++++++++++++-------------- 1 file changed, 35 insertions(+), 16 deletions(-) diff --git a/test/test_service.py b/test/test_service.py index 9b9b98b..8a7bf30 100644 --- a/test/test_service.py +++ b/test/test_service.py @@ -2,28 +2,18 @@ 409 on [POST]/service/ not tested because SQLite throws a different error than Postgres """ +from typing import Any + import pytest from httpx import AsyncClient -from .conftest import generate_query_and_status, generate_body_and_status +from .conftest import generate_query_and_status, generate_body_and_status, standard_test pytestmark = [ pytest.mark.service_module, ] -@pytest.mark.anyio -async def test_get_services_success(default_client: AsyncClient): - resp = await default_client.get("/service?org_id=1") - data = resp.json() - - assert resp.status_code == 200 - assert "services" in data - assert isinstance(data["services"], list) - assert data["services"][0]["id"] == 1 - assert data["services"][0]["name"] == "Test Service" - - @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.anyio async def test_get_services_status_checks( @@ -91,7 +81,36 @@ async def test_patch_services_status_checks( @pytest.mark.anyio -async def test_delete_service_success(default_client: AsyncClient): - resp = await default_client.delete("/service?service_id=1") +async def test_get_services_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/service" + auth_level = "Root User" + query = "?org_id=1" + body = {} + expected_data = { + "services": [ + {"id": 1, "name": "Test Service"}, + ] + } - assert resp.status_code == 204 + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_service_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/service" + auth_level = "Super Admin" + query = "?service_id=1" + body = {} + expected_data = {} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) From 34bd96e14a9233594991a6c6ceb22397c0f0d3c6 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 15:41:19 +0100 Subject: [PATCH 09/11] tests: iam tests standardised --- test/test_iam.py | 487 ++++++++++++++++++++++++++++------------------- 1 file changed, 286 insertions(+), 201 deletions(-) diff --git a/test/test_iam.py b/test/test_iam.py index c890925..314403c 100644 --- a/test/test_iam.py +++ b/test/test_iam.py @@ -1,9 +1,13 @@ """ """ +from datetime import timedelta, datetime, timezone +from typing import Any + import pytest from httpx import AsyncClient -from .conftest import generate_query_and_status, generate_body_and_status +from src.utils import generate_jwt +from .conftest import generate_query_and_status, generate_body_and_status, standard_test pytestmark = [ pytest.mark.iam_module, @@ -144,23 +148,6 @@ async def test_act_on_resource_logic( assert data["allowed"] == expected_response -@pytest.mark.anyio -async def test_get_group_permissions_success(default_client: AsyncClient): - resp = await default_client.get("/iam/group/permissions?org_id=1&group_id=1") - assert resp.status_code == 200 - - data = resp.json() - - assert "permissions" in data - assert isinstance(data["permissions"], list) - - permission = data["permissions"][0] - assert permission["id"] == 1 - assert permission["service_name"] == "Test Service" - assert permission["resource"] == "test_resource" - assert permission["action"] == "read" - - @pytest.mark.parametrize( "query, expected_status", generate_query_and_status(["group_id", "org_id"]) ) @@ -188,31 +175,6 @@ async def test_get_group_permissions_mismatch(default_client: AsyncClient, query assert resp.json()["detail"] == "Group does not belong to this organization" -@pytest.mark.anyio -async def test_get_group_users_success(default_client: AsyncClient): - resp = await default_client.get("/iam/group/users?org_id=1&group_id=1") - assert resp.status_code == 200 - - data = resp.json() - - assert "users" in data - assert isinstance(data["users"], list) - - user = data["users"][0] - assert user["id"] == 1 - assert user["email"] == "admin@test.com" - - assert "group" in data - assert isinstance(data["group"], dict) - assert data["group"]["id"] == 1 - assert data["group"]["name"] == "Org One Group" - - assert "organisation" in data - assert isinstance(data["organisation"], dict) - assert data["organisation"]["id"] == 1 - assert data["organisation"]["name"] == "Org One" - - @pytest.mark.parametrize( "query, expected_status", generate_query_and_status(["group_id", "org_id"]) ) @@ -240,21 +202,6 @@ async def test_get_group_users_mismatch(default_client: AsyncClient, query: str) assert resp.json()["detail"] == "Group does not belong to this organization" -@pytest.mark.anyio -async def test_post_group_success(default_client: AsyncClient): - resp = await default_client.post( - "/iam/group", json={"name": "New Group", "organisation_id": 1} - ) - assert resp.status_code == 201 - - data = resp.json() - - assert "group" in data - assert isinstance(data["group"], dict) - assert data["group"]["name"] == "New Group" - assert data["group"]["id"] == 4 - - @pytest.mark.parametrize( "body, expected_status", generate_body_and_status({"organisation_id": "int", "name": "str"}), @@ -286,31 +233,6 @@ async def test_post_group_non_conflict(default_client: AsyncClient): assert resp.status_code == 201 -@pytest.mark.anyio -async def test_put_group_perm_success(default_client: AsyncClient): - resp = await default_client.put( - "/iam/group/permission", - json={"permission_id": 1, "group_id": 3, "organisation_id": 1}, - ) - assert resp.status_code == 200 - - data = resp.json() - - assert "group" in data - assert isinstance(data["group"], dict) - assert data["group"]["name"] == "Org One Group Two" - assert data["group"]["id"] == 3 - - assert "permissions" in data - assert isinstance(data["permissions"], list) - - permission = data["permissions"][0] - assert permission["id"] == 1 - assert permission["service_name"] == "Test Service" - assert permission["resource"] == "test_resource" - assert permission["action"] == "read" - - @pytest.mark.parametrize( "body, expected_status", generate_body_and_status( @@ -351,30 +273,6 @@ async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict): assert resp.json()["detail"] == "Group does not belong to this organization" -@pytest.mark.anyio -async def test_put_group_user_success(default_client: AsyncClient): - resp = await default_client.put( - "/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1} - ) - assert resp.status_code == 200 - - data = resp.json() - - assert "group" in data - assert isinstance(data["group"], dict) - assert data["group"]["name"] == "Org One Group" - assert data["group"]["id"] == 1 - - assert "users" in data - assert isinstance(data["users"], list) - - user = data["users"][1] - assert user["id"] == 2 - assert user["first_name"] == "User" - assert user["last_name"] == "Test" - assert user["email"] == "user@orgone.com" - - @pytest.mark.parametrize( "body, expected_status", generate_body_and_status( @@ -390,22 +288,6 @@ async def test_put_group_user_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_get_permissions_success(default_client: AsyncClient): - resp = await default_client.get("/iam/permissions?org_id=1") - data = resp.json() - - assert resp.status_code == 200 - assert "permissions" in data - assert isinstance(data["permissions"], list) - - permission = data["permissions"][0] - assert permission["id"] == 1 - assert permission["service_name"] == "Test Service" - assert permission["resource"] == "test_resource" - assert permission["action"] == "read" - - @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.anyio async def test_get_permissions_status_checks( @@ -416,25 +298,6 @@ async def test_get_permissions_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_post_perm_success(default_client: AsyncClient): - resp = await default_client.post( - "/iam/permission", - json={"service_id": 1, "resource": "test_resource", "action": "create"}, - ) - assert resp.status_code == 201 - - data = resp.json() - - assert "permission" in data - assert isinstance(data["permission"], dict) - - assert data["permission"]["id"] == 4 - assert data["permission"]["service_name"] == "Test Service" - assert data["permission"]["resource"] == "test_resource" - assert data["permission"]["action"] == "create" - - @pytest.mark.parametrize( "body, expected_status", [ @@ -621,22 +484,6 @@ async def test_post_perm_search_status_checks( assert resp.status_code == expected_status -@pytest.mark.anyio -async def test_delete_group_permissions_success(default_client: AsyncClient): - resp = await default_client.delete( - "/iam/group/permission?org_id=1&group_id=1&perm_id=1" - ) - data = resp.json() - - assert resp.status_code == 200 - assert "permissions" in data - assert isinstance(data["permissions"], list) - assert len(data["permissions"]) == 0 - assert "group" in data - assert data["group"]["id"] == 1 - assert data["group"]["name"] == "Org One Group" - - @pytest.mark.parametrize( "query, expected_status", generate_query_and_status(["group_id", "org_id", "perm_id"]), @@ -657,49 +504,6 @@ async def test_delete_group_permissions_not_in_group(default_client: AsyncClient assert resp.status_code == 422 -@pytest.mark.anyio -async def test_delete_permissions_success(default_client: AsyncClient): - resp = await default_client.delete("/iam/permission?perm_id=1") - - assert resp.status_code == 204 - - -@pytest.mark.anyio -async def test_delete_group_users_success(default_client: AsyncClient): - resp = await default_client.delete("/iam/group/user?org_id=1&group_id=1&user_id=1") - data = resp.json() - - assert resp.status_code == 200 - assert "users" in data - assert isinstance(data["users"], list) - assert len(data["users"]) == 0 - assert "group" in data - assert data["group"]["id"] == 1 - assert data["group"]["name"] == "Org One Group" - - -@pytest.mark.anyio -async def test_put_group_user_invitation_success(default_client: AsyncClient): - body = {"user_email": "admin@test.com", "organisation_id": 1, "group_id": 1} - resp = await default_client.put("/iam/group/user/invitation", json=body) - - assert resp.status_code == 200 - data = resp.json() - assert "organisation" in data - assert isinstance(data["organisation"], dict) - assert data["organisation"]["id"] == 1 - assert data["organisation"]["name"] == "Org One" - - assert "invited_email" in data - assert isinstance(data["invited_email"], str) - assert data["invited_email"] == "admin@test.com" - - assert "group" in data - assert isinstance(data["group"], dict) - assert data["group"]["name"] == "Org One Group" - assert data["group"]["id"] == 1 - - @pytest.mark.parametrize( "body, expected_status", [ @@ -750,3 +554,284 @@ async def test_put_group_user_invitation_accept_status_checks( if resp.status_code == 401: assert resp.json()["detail"] == "Invalid JWS" + + +@pytest.mark.anyio +async def test_get_group_permissions_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/iam/group/permissions" + auth_level = "Root User" + query = "?org_id=1&group_id=1" + body = {} + expected_data = { + "organisation": {"id": 1, "name": "Org One"}, + "group": {"id": 1, "name": "Org One Group"}, + "permissions": [ + { + "id": 1, + "service_name": "Test Service", + "resource": "test_resource", + "action": "read", + } + ], + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_get_group_users_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/iam/group/users" + auth_level = "Root User" + query = "?org_id=1&group_id=1" + body = {} + expected_data = { + "organisation": {"id": 1, "name": "Org One"}, + "group": {"id": 1, "name": "Org One Group"}, + "users": [{"id": 1, "email": "admin@test.com"}], + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_post_group_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "POST" + path = "/iam/group" + auth_level = "Root User" + query = "" + body = {"name": "New Group", "organisation_id": 1} + expected_data = { + "organisation": {"id": 1, "name": "Org One"}, + "group": {"id": 4, "name": "New Group"}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_put_group_permission_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "PUT" + path = "/iam/group/permission" + auth_level = "Root User" + query = "" + body = {"permission_id": 1, "group_id": 3, "organisation_id": 1} + expected_data = { + "organisation": {"id": 1, "name": "Org One"}, + "group": {"id": 3, "name": "Org One Group Two"}, + "permissions": [ + { + "id": 1, + "service_name": "Test Service", + "resource": "test_resource", + "action": "read", + } + ], + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_put_group_user_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "PUT" + path = "/iam/group/user" + auth_level = "Root User" + query = "" + body = {"user_id": 2, "group_id": 1, "organisation_id": 1} + expected_data = { + "organisation": {"id": 1, "name": "Org One"}, + "group": {"id": 1, "name": "Org One Group"}, + "users": [ + { + "id": 1, + "first_name": "Admin", + "last_name": "Test", + "email": "admin@test.com", + }, + { + "id": 2, + "first_name": "User", + "last_name": "Test", + "email": "user@orgone.com", + }, + ], + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_get_permissions_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "GET" + path = "/iam/permissions" + auth_level = "Root User" + query = "?org_id=1" + body = {} + expected_data = { + "permissions": [ + { + "id": 1, + "service_name": "Test Service", + "resource": "test_resource", + "action": "read", + }, + { + "id": 2, + "service_name": "Test Service", + "resource": "test_resource", + "action": "move", + }, + ] + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_post_permission_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "POST" + path = "/iam/permission" + auth_level = "Super Admin" + query = "" + body = {"service_id": 1, "resource": "test_resource", "action": "create"} + expected_data = { + "permission": { + "id": 4, + "service_name": "Test Service", + "resource": "test_resource", + "action": "create", + } + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_group_permission_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/iam/group/permission" + auth_level = "Root User" + query = "?org_id=1&group_id=1&perm_id=1" + body = {} + expected_data = {"group": {"id": 1, "name": "Org One Group"}, "permissions": []} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_permission_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/iam/permission" + auth_level = "Super Admin" + query = "?perm_id=1" + body = {} + expected_data = {} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_delete_group_user_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "DELETE" + path = "/iam/group/user" + auth_level = "Root User" + query = "?org_id=1&group_id=1&user_id=1" + body = {} + expected_data = {"group": {"id": 1, "name": "Org One Group"}, "users": []} + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_put_group_user_invitation_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + method = "PUT" + path = "/iam/group/user/invitation" + auth_level = "Root User" + query = "" + body = {"user_email": "admin@test.com", "organisation_id": 1, "group_id": 1} + expected_data = { + "group": {"id": 1, "name": "Org One Group"}, + "organisation": {"id": 1, "name": "Org One"}, + "invited_email": "admin@test.com", + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) + + +@pytest.mark.anyio +async def test_put_group_user_invitation_accept_standard( + default_client: AsyncClient, route_data: dict[str, dict[str, Any]] +): + expiry_delta = timedelta(hours=24) + expiry = datetime.now(timezone.utc) + expiry_delta + claims = { + "email": "admin@test.com", + "org_id": 1, + "exp": expiry, + "type": "org_invite", + "group_id": 3, + "group_name": "Org One Group Two", + } + + token = await generate_jwt(claims) + + method = "PUT" + path = "/iam/group/user/invitation/accept" + auth_level = "User" + query = "" + body = {"jwt": token} + expected_data = { + "organisation": {"name": "Org One", "id": 1}, + "user": {"email": "admin@test.com", "id": 1}, + "group": {"details": {"id": 3, "name": "Org One Group Two"}, "permissions": []}, + } + + await standard_test( + default_client, method, path, auth_level, query, body, expected_data, route_data + ) From 6155d955a784a233344bad3a6bc06a5222966c35 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 24 Jun 2026 16:02:48 +0100 Subject: [PATCH 10/11] tests: simplified auth tests --- test/test_auth_general.py | 31 ++++++++ test/test_auth_root.py | 153 -------------------------------------- test/test_auth_su.py | 75 ------------------- test/test_auth_user.py | 28 ------- 4 files changed, 31 insertions(+), 256 deletions(-) delete mode 100644 test/test_auth_root.py delete mode 100644 test/test_auth_su.py delete mode 100644 test/test_auth_user.py diff --git a/test/test_auth_general.py b/test/test_auth_general.py index 543af7a..ea3cd69 100644 --- a/test/test_auth_general.py +++ b/test/test_auth_general.py @@ -16,3 +16,34 @@ async def test_get_org_auth_root_su(default_client: AsyncClient): assert resp.status_code != 422 assert resp.status_code == 200 assert resp.json()["organisations"][0]["name"] == "Org Two" + + +# Standardised tests verify if each endpoint has been assigned the correct auth level. +# Sample tests here verify that each auth level works. + + +@pytest.mark.anyio +async def test_get_org_auth_root(no_su_client: AsyncClient): + # Sample test. Checks if a non-root user gets blocked on a root endpoint. + resp = await no_su_client.get("/org?org_id=2") + assert resp.status_code != 422 + assert resp.status_code == 403 + assert "Must be the org's root user" in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_get_user_auth_su(no_su_client: AsyncClient): + # Sample test. Checks if a non-su user gets blocked on a su endpoint. + resp = await no_su_client.get("/user?user_id=1") + assert resp.status_code != 422 + assert resp.status_code == 403 + assert resp.json()["detail"] == "Must be super admin" + + +@pytest.mark.anyio +async def test_get_self_db_auth_user(no_user_client: AsyncClient): + # Sample test. Checks if a non-user gets blocked on a user endpoint. + resp = await no_user_client.get("/user/self/db") + assert resp.status_code != 422 + assert resp.status_code == 401 + assert resp.json()["detail"] == "Not authenticated" diff --git a/test/test_auth_root.py b/test/test_auth_root.py deleted file mode 100644 index 429bf4b..0000000 --- a/test/test_auth_root.py +++ /dev/null @@ -1,153 +0,0 @@ -""" -This module ensures root user only endpoints do return a correctly formatted 401 when user is not the root user for the org -DELETE endpoints are not tested -""" - -import pytest -from httpx import AsyncClient - -pytestmark = [ - pytest.mark.auth, - pytest.mark.root_user, -] - - -@pytest.mark.anyio -async def test_get_org_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/org?org_id=2") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_patch_org_questionnaire_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.patch( - "/org/questionnaire", - json={ - "organisation_id": 2, - "intake_questionnaire": { - "question_one": "new answer one", - "question_two": None, - "question_three": None, - }, - "partial": True, - }, - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_org_users_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/org/users?org_id=2") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_org_groups_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/org/groups?org_id=2") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_org_contact_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/org/contact?org_id=2&contact_type=billing") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_patch_org_contact_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.patch( - "/org/contact", - json={ - "organisation_id": 2, - "contact_type": "billing", - "email": "user@example.com", - }, - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_service_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/service?org_id=2") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_iam_group_permissions_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/iam/group/permissions?org_id=2&group_id=1") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_iam_group_users_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/iam/group/users?org_id=2&group_id=1") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_post_iam_group_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.post( - "/iam/group", json={"name": "New Group", "organisation_id": 2} - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.put( - "/iam/group/permission", - json={"permission_id": 1, "group_id": 2, "organisation_id": 2}, - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_put_iam_group_user_auth_root( - no_su_client: AsyncClient, -): - resp = await no_su_client.put( - "/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 2} - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_get_iam_permissions_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.get("/iam/permissions?org_id=2") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] - - -@pytest.mark.anyio -async def test_post_iam_permissions_search_auth_root(no_su_client: AsyncClient): - resp = await no_su_client.post( - "/iam/permissions/search", json={"organisation_id": 2, "action": "read"} - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be the org's root user" in resp.json()["detail"] diff --git a/test/test_auth_su.py b/test/test_auth_su.py deleted file mode 100644 index f0136bf..0000000 --- a/test/test_auth_su.py +++ /dev/null @@ -1,75 +0,0 @@ -""" -This module ensures super admin only endpoints do return a correctly formatted 401 when user is not a super admin -DELETE endpoints are not tested -""" - -import pytest -from httpx import AsyncClient - -pytestmark = [ - pytest.mark.auth, - pytest.mark.super_admin, -] - - -@pytest.mark.anyio -async def test_get_user_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.get("/user?user_id=1") - assert resp.status_code != 422 - assert resp.status_code == 403 - assert resp.json()["detail"] == "Must be super admin" - - -@pytest.mark.anyio -async def test_patch_org_status_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.patch( - "/org/status", json={"organisation_id": 1, "status": "submitted"} - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert resp.json()["detail"] == "Must be super admin" - - -@pytest.mark.anyio -async def test_patch_org_root_user_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.patch( - "/org/root_user", json={"organisation_id": 1, "user_id": 2} - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert resp.json()["detail"] == "Must be super admin" - - -@pytest.mark.anyio -async def test_patch_service_key_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.patch("/service/key", json={"service_id": 1}) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert resp.json()["detail"] == "Must be super admin" - - -@pytest.mark.anyio -async def test_post_service_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.post("/service", json={"name": "New Test Service"}) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert resp.json()["detail"] == "Must be super admin" - - -@pytest.mark.anyio -async def test_post_perm_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.post( - "/iam/permission", - json={"service_id": 1, "resource": "test_resource", "action": "create"}, - ) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert resp.json()["detail"] == "Must be super admin" - - -@pytest.mark.anyio -async def test_post_org_user_auth_su(no_su_client: AsyncClient): - resp = await no_su_client.post("/org/user", json={"organisation_id": 1, "user_id": 2}) - assert resp.status_code != 422 - assert resp.status_code == 403 - assert "Must be super admin" in resp.json()["detail"] diff --git a/test/test_auth_user.py b/test/test_auth_user.py deleted file mode 100644 index 3d2f6ce..0000000 --- a/test/test_auth_user.py +++ /dev/null @@ -1,28 +0,0 @@ -""" -This testing module removes the testing user override to verify that endpoints with only the user requirement return a 401 error when not logged in -""" - -import pytest -from httpx import AsyncClient - - -pytestmark = [ - pytest.mark.auth, - pytest.mark.user, -] - - -@pytest.mark.anyio -async def test_get_self_db_auth_user(no_user_client: AsyncClient): - resp = await no_user_client.get("/user/self/db") - assert resp.status_code != 422 - assert resp.status_code == 401 - assert resp.json()["detail"] == "Not authenticated" - - -@pytest.mark.anyio -async def test_post_org_success_auth_user(no_user_client: AsyncClient): - resp = await no_user_client.post("/org", json={"name": "New Test Org"}) - assert resp.status_code != 422 - assert resp.status_code == 401 - assert resp.json()["detail"] == "Not authenticated" From 4395b35320da6705264c27ffee498a5cf7d47c7a Mon Sep 17 00:00:00 2001 From: Renovate Date: Wed, 24 Jun 2026 15:48:45 +0000 Subject: [PATCH 11/11] chore(deps): update dependency starlette to v1.3.1 [security] --- uv.lock | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/uv.lock b/uv.lock index 147ce12..130b316 100644 --- a/uv.lock +++ b/uv.lock @@ -931,15 +931,15 @@ wheels = [ [[package]] name = "starlette" -version = "1.1.0" +version = "1.2.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/95/66/4d20cdf39a8d6a51e663b7038e3b828ff211d3891a43a713fe7e4643f3a8/starlette-1.1.0.tar.gz", hash = "sha256:e83c7fe0ddecd8719c5b840080325aec0260acec86e9832899e377b91d65e90f", size = 2660060, upload-time = "2026-05-23T16:55:41.376Z" } +sdist = { url = "https://files.pythonhosted.org/packages/25/44/ec35f1b6e83094b997da438a02c8c9b0ade2b1e84cfc48bd4656780760a6/starlette-1.2.1.tar.gz", hash = "sha256:9b9b5ebb992e67d6093741e63c2f59e4f6fff986f81163c087867bd7b924b3f6", size = 2701854, upload-time = "2026-05-31T01:07:51.847Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/93/79/920b8e0a8b20f793e8d64855095cb8febabf6175b8550b6f7a547d813891/starlette-1.1.0-py3-none-any.whl", hash = "sha256:7f0dfd38e428aad5cb6f9f667f0ca1d2d8ca3f3385dccac8305f79ec98458382", size = 72899, upload-time = "2026-05-23T16:55:39.201Z" }, + { url = "https://files.pythonhosted.org/packages/1c/54/196d0c1db10af76baa4f64894448505d60d3cdf70ef92cbb35f46a4e4c71/starlette-1.2.1-py3-none-any.whl", hash = "sha256:4de0082d08c8f6764a85a54cf1120d6939507a19905c7768acad2a9f875d2b89", size = 73350, upload-time = "2026-05-31T01:07:50.09Z" }, ] [[package]]