From a907506ec11c6ff5553953e8d1c470a8d3dff748 Mon Sep 17 00:00:00 2001 From: luxferre Date: Wed, 3 Jun 2026 11:58:32 +0100 Subject: [PATCH] tests: org approval auth tests A module of tests which verifies relevant endpoints either do or do not allow unapproved orgs. Issue: #18 --- test/test_auth_approval.py | 157 +++++++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) create mode 100644 test/test_auth_approval.py diff --git a/test/test_auth_approval.py b/test/test_auth_approval.py new file mode 100644 index 0000000..df72976 --- /dev/null +++ b/test/test_auth_approval.py @@ -0,0 +1,157 @@ +""" +This test module checks relevant endpoints to ensure only approved orgs get access or, for pre-approval endpoints, that they are not blocked. +Endpoints not checked here are endpoints that do not require an org check. +Delete endpoints are currently skipped because the testing system cannot use bodies in deletes. +""" +import pytest +from httpx import AsyncClient + +from .conftest import client + +from src.organisation.models import Organisation as Org, OrgUsers +from src.user.models import User +from src.iam.models import Group + + +@pytest.fixture(autouse=True) +def set_org_partial(db_session): + org_model = db_session.get(Org, 1) + org_model.status = "partial" + db_session.flush() + + +@pytest.mark.anyio +async def test_get_org_auth_approval(client: AsyncClient): + resp = await client.get("/org?org_id=1") + assert resp.status_code != 422 + assert resp.status_code == 200 + + +@pytest.mark.anyio +async def test_patch_org_questionnaire_auth_approval(client: AsyncClient): + resp = await client.patch("/org/questionnaire", json={"organisation_id": 1, + "intake_questionnaire": {"question_one": "new answer one", + "question_two": None, + "question_three": None}, + "partial": True}) + assert resp.status_code != 422 + assert resp.status_code == 200 + + +@pytest.mark.anyio +async def test_patch_org_status_auth_approval(client: AsyncClient): + resp = await client.patch("/org/status", json={"organisation_id": 1, "status": "submitted"}) + assert resp.status_code != 422 + assert resp.status_code == 200 + + +@pytest.mark.anyio +async def test_get_org_users_auth_approval(client: AsyncClient): + resp = await client.get("/org/users?org_id=1") + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_post_org_user_auth_approval(client: AsyncClient, db_session): + db_session.add(User(email="user@test.org", first_name="User", last_name="Test", oidc_id="abcd-efgh-ijkl-1234")) + db_session.flush() + + resp = await client.post("/org/user", json={"organisation_id": 1, "user_id": 2}) + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_patch_org_root_user_auth_approval(client: AsyncClient, db_session): + db_session.add(User(email="user@test.org", first_name="User", last_name="Test", oidc_id="abcd-efgh-ijkl-1234")) + db_session.flush() + db_session.add(OrgUsers(org_id=1, user_id=2)) + db_session.flush() + + resp = await client.patch("/org/root_user", json={"organisation_id": 1, "user_id": 2}) + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_get_org_groups_auth_approval(client: AsyncClient): + resp = await client.get("/org/groups?org_id=1") + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_get_org_contact_auth_approval(client: AsyncClient): + resp = await client.get(f"/org/contact?org_id=1&contact_type=billing") + assert resp.status_code != 422 + assert resp.status_code == 200 + + +@pytest.mark.anyio +async def test_patch_org_contact_auth_approval(client: AsyncClient): + resp = await client.patch("/org/contact", + json={"organisation_id": 1, "contact_type": "billing", "email": "user@example.com"}) + assert resp.status_code != 422 + assert resp.status_code == 200 + + +@pytest.mark.anyio +async def test_get_service_auth_approval(client: AsyncClient): + resp = await client.get("/service/?org_id=1") + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_get_iam_group_permissions_auth_approval(client: AsyncClient): + resp = await client.get("/iam/group/permissions?org_id=1&group_id=1") + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_get_iam_group_users_auth_approval(client: AsyncClient): + resp = await client.get("/iam/group/users?org_id=1&group_id=1") + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_post_iam_group_auth_approval(client: AsyncClient): + resp = await client.post("/iam/group", json={"name": "New Group", "organisation_id": 1}) + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_put_iam_group_permission_auth_approval(client: AsyncClient, db_session): + db_session.add(Group(name="Test Group Two", org_id=1)) + db_session.flush() + resp = await client.put("/iam/group/permission", json={"permission_id": 1, "group_id": 2, "organisation_id": 1}) + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_put_iam_group_user_auth_approval(client: AsyncClient, db_session): + db_session.add(User(email="user@test.org", first_name="User", last_name="Test", oidc_id="abcd-efgh-ijkl-1234")) + db_session.flush() + + resp = await client.put("/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}) + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_get_iam_permissions_auth_approval(client: AsyncClient): + resp = await client.get("/iam/permissions?org_id=1") + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"] + + +@pytest.mark.anyio +async def test_post_iam_permissions_search_auth_approval(client: AsyncClient): + resp = await client.post("/iam/permissions/search", json={"organisation_id": 1, "action": "read"}) + assert resp.status_code != 422 + assert "has not been approved." in resp.json()["detail"]