From d3400f53e43e1f47b0c31a60a91884efbe376271 Mon Sep 17 00:00:00 2001 From: luxferre Date: Mon, 1 Jun 2026 15:57:00 +0100 Subject: [PATCH] tests: org router --- test/test_organisation.py | 378 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 378 insertions(+) create mode 100644 test/test_organisation.py diff --git a/test/test_organisation.py b/test/test_organisation.py new file mode 100644 index 0000000..a629d04 --- /dev/null +++ b/test/test_organisation.py @@ -0,0 +1,378 @@ +""" +[DELETE] /org/ is not tested because the testing client cannot attach a body to a delete request. +""" +import pytest +from httpx import AsyncClient + +from src.organisation.models import Organisation, OrgUsers +from src.user.models import User +from .conftest import client + + +@pytest.mark.anyio +async def test_get_org_success(client: AsyncClient): + resp = await client.get("/org/id?org_id=1") + data = resp.json() + + assert resp.status_code == 200 + assert data["name"] == "Test Org" + assert data["root_user"] == "admin@test.com" + assert data["billing_contact"] == "billing@test.org" + assert data["owner_contact"] == "owner@test.org" + assert data["security_contact"] == "security@test.org" + assert data["status"] == "approved" + + +@pytest.mark.parametrize( + "query, expected_status", + [ + ("org_id=2", 404), + ("org_id=banana", 422), + ("", 422), + ], +) +@pytest.mark.anyio +async def test_get_org_failure(client: AsyncClient, query: str, expected_status: int): + resp = await client.get(f"/org/id?{query}") + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_post_org_success(client: AsyncClient): + resp = await 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", + [ + ({"name": 42}, 422), + ({}, 422), + ({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422), + ], +) +@pytest.mark.anyio +async def test_post_org_failure(client: AsyncClient, body: dict[str, str], expected_status: int): + resp = await client.post("/org/", json=body) + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_patch_org_questionnaire_partial_success(client: AsyncClient, db_session): + org_model = db_session.get(Organisation, 1) + org_model.status = "partial" + db_session.flush() + 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}) + data = resp.json() + + assert resp.status_code == 200 + assert "intake_questionnaire" in data + assert data["name"] == "Test Org" + assert data["intake_questionnaire"]["question_one"] == "new answer one" + assert data["status"] == "partial" + # assert type(data["intake_questionnaire"]["question_two"]) == str + assert data["intake_questionnaire"]["question_three"] is None + + +@pytest.mark.parametrize( + "body, expected_status", + [ + ({"organisation_id": 42}, 404), + ({"organisation_id": "Test Org"}, 422), + ({"organisation_id": ""}, 422), + ({}, 422), + ({"organisation_id": "1", "intake_questionnaire": {"question_one": 42}, "partial": True}, 422), + ({"organisation_id": "1", "intake_questionnaire": {"question_one": "valid"}}, 422), + ({"organisation_id": "1", "intake_questionnaire": {"question_one": "valid"}, "partial": 42}, 422), + ], +) +@pytest.mark.anyio +async def test_patch_questionnaire_partial_failure(client: AsyncClient, body: dict[str, str], expected_status: int): + resp = await client.patch("/org/questionnaire", json=body) + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_patch_org_questionnaire_submit_success(client: AsyncClient, db_session): + org_model = db_session.get(Organisation, 1) + org_model.status = "partial" + db_session.flush() + resp = await client.patch("/org/questionnaire", json={"organisation_id": 1, "intake_questionnaire": {"question_one": "new answer one", "question_two": None, "question_three": None}, "partial": False}) + data = resp.json() + + assert resp.status_code == 200 + assert "intake_questionnaire" in data + assert data["name"] == "Test Org" + assert data["intake_questionnaire"]["question_one"] == "new answer one" + assert data["status"] == "submitted" + # assert type(data["intake_questionnaire"]["question_two"]) == str + assert data["intake_questionnaire"]["question_three"] is None + + +@pytest.mark.parametrize( + "status", + ["partial", "submitted", "remediation", "approved", "rejected", "removed"] +) +@pytest.mark.anyio +async def test_patch_org_status_success(client: AsyncClient, status: str): + resp = await client.patch("/org/status", json={"organisation_id": 1, "status": status}) + data = resp.json() + + assert resp.status_code == 200 + assert data["name"] == "Test Org" + assert data["status"] == status + + +@pytest.mark.parametrize( + "body, expected_status", + [ + ({"organisation_id": 42}, 404), + ({"organisation_id": "Test Org"}, 422), + ({"organisation_id": ""}, 422), + ({}, 422), + ({"organisation_id": "1", "status": True}, 422), + ({"organisation_id": "1", "status": 42}, 422), + ], +) +@pytest.mark.anyio +async def test_patch_org_status_failure(client: AsyncClient, body: dict[str, str], expected_status: int): + resp = await client.patch("/org/status", json=body) + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_get_org_users_success(client: AsyncClient): + resp = await client.get("/org/users?org_id=1") + data = resp.json() + + assert resp.status_code == 200 + assert "users" in data + assert type(data["users"]) == list + assert len(data["users"]) == 1 + assert data["users"][0] == "admin@test.com" + + +@pytest.mark.parametrize( + "query, expected_status", + [ + ("org_id=2", 404), + ("org_id=banana", 422), + ("", 422), + ], +) +@pytest.mark.anyio +async def test_get_org_users_failure(client: AsyncClient, query: str, expected_status: int): + resp = await client.get(f"/org/users?{query}") + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_post_org_user_success(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}) + data = resp.json() + + assert resp.status_code == 200 + assert "users" in data + assert type(data["users"]) == list + assert "user@test.org" in data["users"] + + +@pytest.mark.parametrize( + "body, expected_status", + [ + ({"organisation_id": 42}, 404), + ({}, 422), + ({"organisation_id": 1, "user_id": "id"}, 422), + ({"user_id": 2}, 422), + ({"organisation_id": 1, "user_id": 42}, 404), + ], +) +@pytest.mark.anyio +async def test_post_org_failure(client: AsyncClient, body: dict[str, str], expected_status: int, 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=body) + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_patch_org_root_user_success(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}) + data = resp.json() + + assert resp.status_code == 200 + assert data["name"] == "Test Org" + assert data["root_user_email"] == "user@test.org" + + +@pytest.mark.parametrize( + "body, expected_status", + [ + ({"organisation_id": 42, "user_id": 2}, 404), + ({"organisation_id": "Test Org", "user_id": 2}, 422), + ({"organisation_id": "", "user_id": 2}, 422), + ({}, 422), + ({"user_id": 2}, 422), + ({"user_id": 42}, 404), + ({"organisation_id": 1, "user_id": "Test User"}, 422), + ], +) +@pytest.mark.anyio +async def test_patch_root_user_failure(client: AsyncClient, body: dict[str, str], expected_status: int, 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=body) + + assert resp.status_code == expected_status + + +@pytest.mark.anyio +async def test_get_org_groups_success(client: AsyncClient): + resp = await client.get("/org/groups?org_id=1") + data = resp.json() + + assert resp.status_code == 200 + assert "groups" in data + assert type(data["groups"]) == list + assert "Test Group" in data["groups"] + + +@pytest.mark.parametrize( + "query, expected_status", + [ + ("org_id=2", 404), + ("org_id=banana", 422), + ("", 422), + ], +) +@pytest.mark.anyio +async def test_get_org_groups_failure(client: AsyncClient, query: str, expected_status: int): + resp = await client.get(f"/org/groups?{query}") + + assert resp.status_code == expected_status + + +@pytest.mark.parametrize( + "contact_type", + ["billing", "security", "owner"] +) +@pytest.mark.anyio +async def test_get_org_contact_success(client: AsyncClient, contact_type: str): + resp = await client.get(f"/org/contact?org_id=1&contact_type={contact_type}") + data = resp.json() + + assert resp.status_code == 200 + + attributes = [ + "email", + "first_name", + "last_name", + "phonenumber", + "vat_number", + "address", + ] + + for attribute in attributes: + assert attribute in data["contact"] + + address_attributes = [ + "post_office_box_number", + "street_address", + "street_address_line_2", + "locality", + "address_region", + "country_code", + "postal_code", + ] + + for attribute in address_attributes: + assert attribute in data["contact"]["address"] + + +@pytest.mark.parametrize( + "query, expected_status", + [ + ("org_id=42&contact_type=billing", 404), + ("org_id=banana&contact_type=billing", 422), + ("", 422), + ("org_id=1&contact_type=contact", 422), + ("contact_type=billing", 422), + ], +) +@pytest.mark.anyio +async def test_get_org_contact_failure(client: AsyncClient, query: str, expected_status: int): + resp = await client.get(f"/org/contact?{query}") + + assert resp.status_code == expected_status + + +@pytest.mark.parametrize( + "key, value", + [ + ("email", "user@example.com"), + ("first_name", "John"), + ("last_name", "Doe"), + ("phonenumber", "+441234567890"), + ("vat_number", "GB123456789"), + ("post_office_box_number", "PO Box 123"), + ("street_address", "123 Example Street"), + ("street_address_line_2", "Suite 4B"), + ("locality", "Glasgow"), + ("address_region", "Glasgow City"), + ("country_code", "GB"), + ("postal_code", "G1 1AA"), + ] +) +@pytest.mark.anyio +async def test_patch_org_contact_success(client: AsyncClient, key: str, value: str): + resp = await client.patch("/org/contact", json={"organisation_id": 1, "contact_type": "billing", key: value}) + data = resp.json() + + assert resp.status_code == 200 + if key in data["contact"]: + assert data["contact"][key] == value + elif key in data["contact"]["address"]: + assert data["contact"]["address"][key] == value + else: + pytest.fail(f"Invalid contact key: {key}") + + +@pytest.mark.parametrize( + "body, expected_status", + [ + ({"organisation_id": 42, "contact_type": "billing"}, 404), + ({"organisation_id": "Test Org", "contact_type": "billing"}, 422), + ({"organisation_id": "", "contact_type": "billing"}, 422), + ({}, 422), + ({"organisation_id": 1, "contact_type": "not_real"}, 422), + ({"organisation_id": 1, "contact_type": 42}, 422), + ({"organisation_id": 1, "contact_type": ""}, 422), + ], +) +@pytest.mark.anyio +async def test_patch_org_status_failure(client: AsyncClient, body: dict[str, str], expected_status: int): + resp = await client.patch("/org/contact", json=body) + + assert resp.status_code == expected_status