sr2/cloud-api
6
0
Fork 1
Code Issues 8 Pull requests 2 Projects Releases Packages Wiki Activity Actions

minor: ruff formatter

Browse source 
All changes are either:
- Correcting tabs
- Adding/removing line breaks
- Adding trailing commas
This commit is contained in:
Chris Milne 2026-06-08 15:31:37 +01:00
parent b2e5dd2ebb
commit c689ac1e10
91 changed files with 1710 additions and 689 deletions
Download patch file Download diff file Expand all files Collapse all files

39
test/conftest.py
View file

@ -37,11 +37,14 @@ def db_session():
async def default_client(db_session) -> AsyncGenerator[AsyncClient, None]:
def get_db_override():
return db_session
app.dependency_overrides[get_db] = get_db_override
app.dependency_overrides[get_current_user] = get_dev_user
app.dependency_overrides[get_super_admin_list] = testing_su_list
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://localhost:8000/api/v1") as ac:
async with AsyncClient(
transport=transport, base_url="http://localhost:8000/api/v1"
) as ac:
yield ac
app.dependency_overrides.clear()
@ -51,37 +54,58 @@ async def default_client(db_session) -> AsyncGenerator[AsyncClient, None]:
async def no_user_client(db_session) -> AsyncGenerator[AsyncClient, None]:
def get_db_override():
return db_session
app.dependency_overrides[get_db] = get_db_override
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://localhost:8000/api/v1") as ac:
async with AsyncClient(
transport=transport, base_url="http://localhost:8000/api/v1"
) as ac:
yield ac
app.dependency_overrides.clear()
@pytest.fixture
async def no_su_client(db_session) -> AsyncGenerator[AsyncClient, None]:
def get_db_override():
return db_session
app.dependency_overrides[get_db] = get_db_override
app.dependency_overrides[get_current_user] = get_dev_user
app.dependency_overrides[get_super_admin_list] = empty_su_list
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://localhost:8000/api/v1") as ac:
async with AsyncClient(
transport=transport, base_url="http://localhost:8000/api/v1"
) as ac:
yield ac
app.dependency_overrides.clear()
def _seed(db):
db.add(User(email="admin@test.com", first_name="Admin", last_name="Test", oidc_id="abcd-efgh-ijkl-mnop"))
db.add(
User(
email="admin@test.com",
first_name="Admin",
last_name="Test",
oidc_id="abcd-efgh-ijkl-mnop",
)
)
db.add(Contact(org_id=1, email="billing@test.org", phonenumber="07521539927"))
db.add(Contact(org_id=1, email="owner@test.org", phonenumber="07521539927"))
db.add(Contact(org_id=1, email="security@test.org", phonenumber="07521539927"))
db.flush()
db.add(Org(name="Test Org", root_user_id=1, billing_contact_id=1, owner_contact_id=2, security_contact_id=3,
status="approved", intake_questionnaire={"question_two": "answer two"}))
db.add(
Org(
name="Test Org",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
intake_questionnaire={"question_two": "answer two"},
)
)
db.add(Service(name="Test Service", api_key="123456789"))
db.add(Permission(service_id=1, resource="test_resource", action="read"))
db.add(Group(name="Test Group", org_id=1))
@ -131,6 +155,7 @@ 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
#

96
test/test_auth_approval.py
View file

@ -3,6 +3,7 @@ This test module checks relevant endpoints to ensure only approved orgs get acce
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
@ -27,18 +28,27 @@ async def test_get_org_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient):
resp = await default_client.patch("/org/questionnaire", json={"organisation_id": 1,
"intake_questionnaire": {"question_one": "new answer one",
"question_two": None,
"question_three": None},
"partial": True})
resp = await default_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(default_client: AsyncClient):
resp = await default_client.patch("/org/status", json={"organisation_id": 1, "status": "submitted"})
resp = await default_client.patch(
"/org/status", json={"organisation_id": 1, "status": "submitted"}
)
assert resp.status_code != 422
assert resp.status_code == 200
@ -52,22 +62,42 @@ async def test_get_org_users_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_post_org_user_auth_approval(default_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.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
resp = await default_client.post("/org/user", json={"organisation_id": 1, "user_id": 2})
resp = await default_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(default_client: AsyncClient, db_session):
db_session.add(User(email="user@test.org", first_name="User", last_name="Test", oidc_id="abcd-efgh-ijkl-1234"))
async def test_patch_org_root_user_auth_approval(
default_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 default_client.patch("/org/root_user", json={"organisation_id": 1, "user_id": 2})
resp = await default_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"]
@ -88,8 +118,14 @@ async def test_get_org_contact_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
resp = await default_client.patch("/org/contact",
json={"organisation_id": 1, "contact_type": "billing", "email": "user@example.com"})
resp = await default_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
@ -117,26 +153,44 @@ async def test_get_iam_group_users_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_post_iam_group_auth_approval(default_client: AsyncClient):
resp = await default_client.post("/iam/group", json={"name": "New Group", "organisation_id": 1})
resp = await default_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(default_client: AsyncClient, db_session):
async def test_put_iam_group_permission_auth_approval(
default_client: AsyncClient, db_session
):
db_session.add(Group(name="Test Group Two", org_id=1))
db_session.flush()
resp = await default_client.put("/iam/group/permission", json={"permission_id": 1, "group_id": 2, "organisation_id": 1})
resp = await default_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(default_client: AsyncClient, db_session):
db_session.add(User(email="user@test.org", first_name="User", last_name="Test", oidc_id="abcd-efgh-ijkl-1234"))
async def test_put_iam_group_user_auth_approval(
default_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 default_client.put("/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1})
resp = await default_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"]
@ -150,6 +204,8 @@ async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_post_iam_permissions_search_auth_approval(default_client: AsyncClient):
resp = await default_client.post("/iam/permissions/search", json={"organisation_id": 1, "action": "read"})
resp = await default_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"]

25
test/test_auth_general.py
View file

@ -1,5 +1,5 @@
"""
"""
""" """
import pytest
from httpx import AsyncClient
@ -10,11 +10,26 @@ from src.user.models import User
@pytest.mark.anyio
async def test_get_org_auth_root_su(default_client: AsyncClient, db_session):
# If a super admin can access a resource when not the root user
db_session.add(User(email="admin@test.org", first_name="Admin", last_name="Test", oidc_id="abcd-efgh-ijkl-4321"))
db_session.add(
User(
email="admin@test.org",
first_name="Admin",
last_name="Test",
oidc_id="abcd-efgh-ijkl-4321",
)
)
db_session.flush()
db_session.add(
Org(name="Test Org Two", root_user_id=2, billing_contact_id=1, owner_contact_id=2, security_contact_id=3,
status="approved", intake_questionnaire={}))
Org(
name="Test Org Two",
root_user_id=2,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
intake_questionnaire={},
)
)
db_session.flush()
resp = await default_client.get("/org?org_id=2")

93
test/test_auth_root.py
View file

@ -2,6 +2,7 @@
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
@ -12,10 +13,26 @@ from src.iam.models import Group
@pytest.fixture(autouse=True)
def add_second_org(db_session):
db_session.add(User(email="admin@test.org", first_name="Admin", last_name="Test", oidc_id="abcd-efgh-ijkl-4321"))
db_session.add(
User(
email="admin@test.org",
first_name="Admin",
last_name="Test",
oidc_id="abcd-efgh-ijkl-4321",
)
)
db_session.flush()
db_session.add(Org(name="Test Org Two", root_user_id=2, billing_contact_id=1, owner_contact_id=2, security_contact_id=3,
status="approved", intake_questionnaire={}))
db_session.add(
Org(
name="Test Org Two",
root_user_id=2,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
intake_questionnaire={},
)
)
db_session.flush()
@ -29,11 +46,18 @@ async def test_get_org_auth_root(no_su_client: AsyncClient):
@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})
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 == 401
assert "Must be the org's root user" in resp.json()["detail"]
@ -49,10 +73,19 @@ async def test_get_org_users_auth_root(no_su_client: AsyncClient):
@pytest.mark.anyio
async def test_post_org_user_auth_root(no_su_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.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
resp = await no_su_client.post("/org/user", json={"organisation_id": 2, "user_id": 2})
resp = await no_su_client.post(
"/org/user", json={"organisation_id": 2, "user_id": 2}
)
assert resp.status_code != 422
assert resp.status_code == 401
assert "Must be the org's root user" in resp.json()["detail"]
@ -76,8 +109,14 @@ async def test_get_org_contact_auth_root(no_su_client: AsyncClient):
@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"})
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 == 401
assert "Must be the org's root user" in resp.json()["detail"]
@ -109,17 +148,24 @@ async def test_get_iam_group_users_auth_root(no_su_client: AsyncClient):
@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})
resp = await no_su_client.post(
"/iam/group", json={"name": "New Group", "organisation_id": 2}
)
assert resp.status_code != 422
assert resp.status_code == 401
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, db_session):
async def test_put_iam_group_permission_auth_root(
no_su_client: AsyncClient, db_session
):
db_session.add(Group(name="Test Group Two", org_id=2))
db_session.flush()
resp = await no_su_client.put("/iam/group/permission", json={"permission_id": 1, "group_id": 2, "organisation_id": 2})
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 == 401
assert "Must be the org's root user" in resp.json()["detail"]
@ -127,10 +173,19 @@ async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient, db_
@pytest.mark.anyio
async def test_put_iam_group_user_auth_root(no_su_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.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
resp = await no_su_client.put("/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 2})
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 == 401
assert "Must be the org's root user" in resp.json()["detail"]
@ -146,7 +201,9 @@ async def test_get_iam_permissions_auth_root(no_su_client: AsyncClient):
@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"})
resp = await no_su_client.post(
"/iam/permissions/search", json={"organisation_id": 2, "action": "read"}
)
assert resp.status_code != 422
assert resp.status_code == 401
assert "Must be the org's root user" in resp.json()["detail"]

23
test/test_auth_su.py
View file

@ -2,6 +2,7 @@
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
@ -19,7 +20,9 @@ async def test_get_user_auth_su(no_su_client: AsyncClient):
@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"})
resp = await no_su_client.patch(
"/org/status", json={"organisation_id": 1, "status": "submitted"}
)
assert resp.status_code != 422
assert resp.status_code == 401
assert resp.json()["detail"] == "Must be super admin"
@ -27,12 +30,21 @@ async def test_patch_org_status_auth_su(no_su_client: AsyncClient):
@pytest.mark.anyio
async def test_patch_org_root_user_auth_su(no_su_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.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 no_su_client.patch("/org/root_user", json={"organisation_id": 1, "user_id": 2})
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 == 401
assert resp.json()["detail"] == "Must be super admin"
@ -56,7 +68,10 @@ async def test_post_service_auth_su(no_su_client: AsyncClient):
@pytest.mark.anyio
async def test_post_perm_success(no_su_client: AsyncClient, db_session):
resp = await no_su_client.post("/iam/permission", json={"service_id": 1, "resource": "test_resource", "action": "create"})
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 == 401
assert resp.json()["detail"] == "Must be super admin"

1
test/test_auth_user.py
View file

@ -1,6 +1,7 @@
"""
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

6
test/test_healthcheck.py
View file

@ -4,7 +4,7 @@ from httpx import AsyncClient
@pytest.mark.anyio
async def test_healthcheck(default_client: AsyncClient):
resp = await default_client.get("/healthcheck")
resp = await default_client.get("/healthcheck")
assert resp.status_code == 200
assert resp.json() == {"status": "ok"}
assert resp.status_code == 200
assert resp.json() == {"status": "ok"}

535
test/test_iam.py
View file

@ -1,5 +1,5 @@
"""
"""
""" """
import pytest
from httpx import AsyncClient
@ -15,13 +15,15 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
body = {
"service": "Test Service",
"organisation": "Test Org",
"resource": "test_resource"
"resource": "test_resource",
}
headers = {
"Authorization": "Bearer not_checked_when_auth_is_disabled",
"X-API-Key": "123456789"
"X-API-Key": "123456789",
}
resp = await default_client.post("/iam/can_act_on_resource?action=read", json=body, headers=headers)
resp = await default_client.post(
"/iam/can_act_on_resource?action=read", json=body, headers=headers
)
data = resp.json()
assert resp.status_code == 200
@ -30,23 +32,20 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
@pytest.mark.parametrize(
"service, api_key",
[
("Test Service", "not_the_correct_key"),
("Test Service Two", "123456789")
],
[("Test Service", "not_the_correct_key"), ("Test Service Two", "123456789")],
)
@pytest.mark.anyio
async def test_act_on_resource_wrong_key(default_client: AsyncClient, db_session, service: str, api_key: str):
body = {
"service": service,
"organisation": "Test Org",
"resource": "test_resource"
}
async def test_act_on_resource_wrong_key(
default_client: AsyncClient, db_session, service: str, api_key: str
):
body = {"service": service, "organisation": "Test Org", "resource": "test_resource"}
headers = {
"Authorization": "Bearer not_checked_when_auth_is_disabled",
"X-API-Key": api_key
"X-API-Key": api_key,
}
resp = await default_client.post("/iam/can_act_on_resource?action=read", json=body, headers=headers)
resp = await default_client.post(
"/iam/can_act_on_resource?action=read", json=body, headers=headers
)
data = resp.json()
assert resp.status_code == 401
@ -58,12 +57,12 @@ async def test_act_on_resource_missing_key(default_client: AsyncClient):
body = {
"service": "Test Service",
"organisation": "Test Org",
"resource": "test_resource"
"resource": "test_resource",
}
headers = {
"Authorization": "Bearer not_checked_when_auth_is_disabled"
}
resp = await default_client.post("/iam/can_act_on_resource?action=read", json=body, headers=headers)
headers = {"Authorization": "Bearer not_checked_when_auth_is_disabled"}
resp = await default_client.post(
"/iam/can_act_on_resource?action=read", json=body, headers=headers
)
data = resp.json()
assert resp.status_code == 401
@ -82,18 +81,17 @@ async def test_act_on_resource_missing_key(default_client: AsyncClient):
],
)
@pytest.mark.anyio
async def test_act_on_resource_endpoint_status_checks(default_client: AsyncClient, service, org, resource, action,
expected_status: int):
body = {
"service": service,
"organisation": org,
"resource": resource
}
async def test_act_on_resource_endpoint_status_checks(
default_client: AsyncClient, service, org, resource, action, expected_status: int
):
body = {"service": service, "organisation": org, "resource": resource}
headers = {
"Authorization": "Bearer not_checked_when_auth_is_disabled",
"X-API-Key": "123456789"
"X-API-Key": "123456789",
}
resp = await default_client.post(f"/iam/can_act_on_resource?action={action}", json=body, headers=headers)
resp = await default_client.post(
f"/iam/can_act_on_resource?action={action}", json=body, headers=headers
)
assert resp.status_code == expected_status
@ -108,18 +106,23 @@ async def test_act_on_resource_endpoint_status_checks(default_client: AsyncClien
],
)
@pytest.mark.anyio
async def test_act_on_resource_logic(default_client: AsyncClient, db_session, service, org, resource, action,
expected_response: bool):
body = {
"service": service,
"organisation": org,
"resource": resource
}
async def test_act_on_resource_logic(
default_client: AsyncClient,
db_session,
service,
org,
resource,
action,
expected_response: bool,
):
body = {"service": service, "organisation": org, "resource": resource}
headers = {
"Authorization": "Bearer not_checked_when_auth_is_disabled",
"X-API-Key": "123456789"
"X-API-Key": "123456789",
}
resp = await default_client.post(f"/iam/can_act_on_resource?action={action}", json=body, headers=headers)
resp = await default_client.post(
f"/iam/can_act_on_resource?action={action}", json=body, headers=headers
)
data = resp.json()
assert resp.status_code == 200
@ -140,11 +143,12 @@ async def test_get_group_permissions_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["group_id", "org_id"])
"query, expected_status", generate_query_and_status(["group_id", "org_id"])
)
@pytest.mark.anyio
async def test_get_group_permissions_status_checks(default_client: AsyncClient, db_session, query: str, expected_status: int):
async def test_get_group_permissions_status_checks(
default_client: AsyncClient, db_session, query: str, expected_status: int
):
resp = await default_client.get(f"/iam/group/permissions?{query}")
assert resp.status_code == expected_status
@ -158,8 +162,19 @@ async def test_get_group_permissions_status_checks(default_client: AsyncClient,
],
)
@pytest.mark.anyio
async def test_get_group_permissions_mismatch(default_client: AsyncClient, db_session, query: str):
db_session.add(Org(name="Another Test Org", root_user_id=1, billing_contact_id=1, owner_contact_id=2, security_contact_id=3, status="approved"))
async def test_get_group_permissions_mismatch(
default_client: AsyncClient, db_session, query: str
):
db_session.add(
Org(
name="Another Test Org",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
)
)
db_session.add(Group(name="Another Test Group", org_id=2))
db_session.flush()
resp = await default_client.get(f"/iam/group/permissions?{query}")
@ -183,11 +198,12 @@ async def test_get_group_users_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["group_id", "org_id"])
"query, expected_status", generate_query_and_status(["group_id", "org_id"])
)
@pytest.mark.anyio
async def test_get_group_users_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_group_users_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/iam/group/users?{query}")
assert resp.status_code == expected_status
@ -201,8 +217,19 @@ async def test_get_group_users_status_checks(default_client: AsyncClient, query:
],
)
@pytest.mark.anyio
async def test_get_group_users_mismatch(default_client: AsyncClient, db_session, query: str):
db_session.add(Org(name="Another Test Org", root_user_id=1, billing_contact_id=1, owner_contact_id=2, security_contact_id=3, status="approved"))
async def test_get_group_users_mismatch(
default_client: AsyncClient, db_session, query: str
):
db_session.add(
Org(
name="Another Test Org",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
)
)
db_session.add(Group(name="Another Test Group", org_id=2))
db_session.flush()
resp = await default_client.get(f"/iam/group/users?{query}")
@ -213,7 +240,9 @@ async def test_get_group_users_mismatch(default_client: AsyncClient, db_session,
@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})
resp = await default_client.post(
"/iam/group", json={"name": "New Group", "organisation_id": 1}
)
data = resp.json()
assert resp.status_code == 200
@ -227,10 +256,22 @@ async def test_post_group_success(default_client: AsyncClient):
"body, expected_status",
[
({"organisation_id": 1, "name": "Test Group"}, 409),
({"organisation_id": 2, "name": "new group"}, 404), # Non-existent organisation, valid name
({"organisation_id": "banana", "name": "new group"}, 422), # Invalid organisation ID, valid name
({"organisation_id": "", "name": "new group"}, 422), # Blank organisation ID, valid name
({"organisation_id": -1, "name": "new group"}, 422), # Negative organisation ID, valid name
(
{"organisation_id": 2, "name": "new group"},
404,
), # Non-existent organisation, valid name
(
{"organisation_id": "banana", "name": "new group"},
422,
), # Invalid organisation ID, valid name
(
{"organisation_id": "", "name": "new group"},
422,
), # Blank organisation ID, valid name
(
{"organisation_id": -1, "name": "new group"},
422,
), # Negative organisation ID, valid name
({"name": 1}, 422), # Only name
({}, 422), # Blank body
({"organisation_id": "", "name": ""}, 422), # Both blank
@ -241,7 +282,9 @@ async def test_post_group_success(default_client: AsyncClient):
],
)
@pytest.mark.anyio
async def test_post_group_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_post_group_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/iam/group", json=body)
assert resp.status_code == expected_status
@ -251,7 +294,10 @@ async def test_post_group_status_checks(default_client: AsyncClient, body: dict[
async def test_put_group_perm_success(default_client: AsyncClient, db_session):
db_session.add(Group(name="Test Group Two", org_id=1))
db_session.flush()
resp = await default_client.put("/iam/group/permission", json={"permission_id": 1, "group_id": 2, "organisation_id": 1})
resp = await default_client.put(
"/iam/group/permission",
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
)
data = resp.json()
assert resp.status_code == 200
@ -270,36 +316,71 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
@pytest.mark.parametrize(
"body, expected_status",
[
({"organisation_id": 42, "group_id": 1, "permission_id": 1}, 404), # Non-existent organisation
({"organisation_id": "banana", "group_id": 1, "permission_id": 1}, 422), # Invalid organisation ID
({"organisation_id": "", "group_id": 1, "permission_id": 1}, 422), # Blank organisation ID
({"organisation_id": -1, "group_id": 1, "permission_id": 1}, 422), # Negative organisation ID
({"organisation_id": 1, "group_id": 42, "permission_id": 1}, 404), # Non-existent group
({"organisation_id": 1, "group_id": "banana", "permission_id": 1}, 422), # Invalid group ID
({"organisation_id": 1, "group_id": "", "permission_id": 1}, 422), # Blank group ID
({"organisation_id": 1, "group_id": -1, "permission_id": 1}, 422), # Negative group ID
({"organisation_id": 1, "group_id": 1, "permission_id": 42}, 404), # Non-existent permission
({"organisation_id": 1, "group_id": 1, "permission_id": "banana"}, 422), # Invalid permission ID
({"organisation_id": 1, "group_id": 1, "permission_id": ""}, 422), # Blank permission ID
({"organisation_id": 1, "group_id": 1, "permission_id": -1}, 422), # Negative permission ID
(
{"organisation_id": 42, "group_id": 1, "permission_id": 1},
404,
), # Non-existent organisation
(
{"organisation_id": "banana", "group_id": 1, "permission_id": 1},
422,
), # Invalid organisation ID
(
{"organisation_id": "", "group_id": 1, "permission_id": 1},
422,
), # Blank organisation ID
(
{"organisation_id": -1, "group_id": 1, "permission_id": 1},
422,
), # Negative organisation ID
(
{"organisation_id": 1, "group_id": 42, "permission_id": 1},
404,
), # Non-existent group
(
{"organisation_id": 1, "group_id": "banana", "permission_id": 1},
422,
), # Invalid group ID
(
{"organisation_id": 1, "group_id": "", "permission_id": 1},
422,
), # Blank group ID
(
{"organisation_id": 1, "group_id": -1, "permission_id": 1},
422,
), # Negative group ID
(
{"organisation_id": 1, "group_id": 1, "permission_id": 42},
404,
), # Non-existent permission
(
{"organisation_id": 1, "group_id": 1, "permission_id": "banana"},
422,
), # Invalid permission ID
(
{"organisation_id": 1, "group_id": 1, "permission_id": ""},
422,
), # Blank permission ID
(
{"organisation_id": 1, "group_id": 1, "permission_id": -1},
422,
), # Negative permission ID
({}, 422), # Blank body
({"permission_id": 1}, 422), # Only permission
({"organisation_id": 1}, 422), # Only organisation
({"group_id": 1}, 422), # Only group
({"organisation_id": 1, "permission_id": 1}, 422), # Missing group
({"group_id": 1, "permission_id": 1}, 422), # Missing organisation
({"organisation_id": 1, "group_id": 1}, 422), # Missing permission
({"organisation_id": 1, "group_id": 1, "permission_id": 1}, 409), # Permission already in group
(
{"organisation_id": 1, "group_id": 1, "permission_id": 1},
409,
), # Permission already in group
],
)
@pytest.mark.anyio
async def test_put_group_perm_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_put_group_perm_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.put("/iam/group/permission", json=body)
assert resp.status_code == expected_status
@ -313,8 +394,19 @@ async def test_put_group_perm_status_checks(default_client: AsyncClient, body: d
],
)
@pytest.mark.anyio
async def test_put_group_perm_mismatch(default_client: AsyncClient, db_session, body: dict):
db_session.add(Org(name="Another Test Org", root_user_id=1, billing_contact_id=1, owner_contact_id=2, security_contact_id=3, status="approved"))
async def test_put_group_perm_mismatch(
default_client: AsyncClient, db_session, body: dict
):
db_session.add(
Org(
name="Another Test Org",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
)
)
db_session.add(Group(name="Another Test Group", org_id=2))
db_session.flush()
resp = await default_client.put("/iam/group/permission", json=body)
@ -325,10 +417,19 @@ async def test_put_group_perm_mismatch(default_client: AsyncClient, db_session,
@pytest.mark.anyio
async def test_put_group_user_success(default_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.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
resp = await default_client.put("/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1})
resp = await default_client.put(
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
)
data = resp.json()
assert resp.status_code == 200
@ -348,34 +449,58 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
@pytest.mark.parametrize(
"body, expected_status",
[
({"organisation_id": 42, "group_id": 1, "user_id": 1}, 404), # Non-existent organisation
({"organisation_id": "banana", "group_id": 1, "user_id": 1}, 422), # Invalid organisation ID
({"organisation_id": "", "group_id": 1, "user_id": 1}, 422), # Blank organisation ID
({"organisation_id": -1, "group_id": 1, "user_id": 1}, 422), # Negative organisation ID
({"organisation_id": 1, "group_id": 42, "user_id": 1}, 404), # Non-existent group
({"organisation_id": 1, "group_id": "banana", "user_id": 1}, 422), # Invalid group ID
(
{"organisation_id": 42, "group_id": 1, "user_id": 1},
404,
), # Non-existent organisation
(
{"organisation_id": "banana", "group_id": 1, "user_id": 1},
422,
), # Invalid organisation ID
(
{"organisation_id": "", "group_id": 1, "user_id": 1},
422,
), # Blank organisation ID
(
{"organisation_id": -1, "group_id": 1, "user_id": 1},
422,
), # Negative organisation ID
(
{"organisation_id": 1, "group_id": 42, "user_id": 1},
404,
), # Non-existent group
(
{"organisation_id": 1, "group_id": "banana", "user_id": 1},
422,
), # Invalid group ID
({"organisation_id": 1, "group_id": "", "user_id": 1}, 422), # Blank group ID
({"organisation_id": 1, "group_id": -1, "user_id": 1}, 422), # Negative group ID
({"organisation_id": 1, "group_id": 1, "user_id": 42}, 404), # Non-existent user
({"organisation_id": 1, "group_id": 1, "user_id": "banana"}, 422), # Invalid user ID
(
{"organisation_id": 1, "group_id": -1, "user_id": 1},
422,
), # Negative group ID
(
{"organisation_id": 1, "group_id": 1, "user_id": 42},
404,
), # Non-existent user
(
{"organisation_id": 1, "group_id": 1, "user_id": "banana"},
422,
), # Invalid user ID
({"organisation_id": 1, "group_id": 1, "user_id": ""}, 422), # Blank user ID
({"organisation_id": 1, "group_id": 1, "user_id": -1}, 422), # Negative user ID
({}, 422), # Blank body
({"user_id": 1}, 422), # Only user
({"organisation_id": 1}, 422), # Only organisation
({"group_id": 1}, 422), # Only group
({"organisation_id": 1, "user_id": 1}, 422), # Missing group
({"group_id": 1, "user_id": 1}, 422), # Missing organisation
({"organisation_id": 1, "group_id": 1}, 422), # Missing user
],
)
@pytest.mark.anyio
async def test_put_group_user_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_put_group_user_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.put("/iam/group/user", json=body)
assert resp.status_code == expected_status
@ -395,11 +520,12 @@ async def test_get_permissions_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["org_id"])
"query, expected_status", generate_query_and_status(["org_id"])
)
@pytest.mark.anyio
async def test_get_permissions_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_permissions_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/iam/permissions?{query}")
assert resp.status_code == expected_status
@ -407,7 +533,10 @@ async def test_get_permissions_status_checks(default_client: AsyncClient, query:
@pytest.mark.anyio
async def test_post_perm_success(default_client: AsyncClient, db_session):
resp = await default_client.post("/iam/permission", json={"service_id": 1, "resource": "test_resource", "action": "create"})
resp = await default_client.post(
"/iam/permission",
json={"service_id": 1, "resource": "test_resource", "action": "create"},
)
data = resp.json()
assert resp.status_code == 200
@ -418,51 +547,70 @@ async def test_post_perm_success(default_client: AsyncClient, db_session):
@pytest.mark.parametrize(
"body, expected_status",
[
# service_id tests
({"service_id": 42, "resource": "test_resource", "action": "read"}, 404), # Non-existent service
({"service_id": "banana", "resource": "test_resource", "action": "read"}, 422), # Invalid service ID
({"service_id": "", "resource": "test_resource", "action": "read"}, 422), # Blank service ID
({"service_id": -1, "resource": "test_resource", "action": "read"}, 422), # Negative service ID
# resource tests
({"service_id": 1, "resource": 42, "action": "read"}, 422), # Invalid resource type
# action tests
({"service_id": 1, "resource": "test_resource", "action": 42}, 422), # Invalid action type
# missing/partial body tests
({}, 422), # Blank body
({"resource": "test_resource"}, 422), # Only resource
({"action": "read"}, 422), # Only action
({"service_id": 1}, 422), # Only service
({"service_id": 1, "action": "read"}, 422), # Missing resource
({"service_id": 1, "resource": "test_resource"}, 422), # Missing action
({"resource": "test_resource", "action": "read"}, 422), # Missing service
],
"body, expected_status",
[
# service_id tests
(
{"service_id": 42, "resource": "test_resource", "action": "read"},
404,
), # Non-existent service
(
{"service_id": "banana", "resource": "test_resource", "action": "read"},
422,
), # Invalid service ID
(
{"service_id": "", "resource": "test_resource", "action": "read"},
422,
), # Blank service ID
(
{"service_id": -1, "resource": "test_resource", "action": "read"},
422,
), # Negative service ID
# resource tests
(
{"service_id": 1, "resource": 42, "action": "read"},
422,
), # Invalid resource type
# action tests
(
{"service_id": 1, "resource": "test_resource", "action": 42},
422,
), # Invalid action type
# missing/partial body tests
({}, 422), # Blank body
({"resource": "test_resource"}, 422), # Only resource
({"action": "read"}, 422), # Only action
({"service_id": 1}, 422), # Only service
({"service_id": 1, "action": "read"}, 422), # Missing resource
({"service_id": 1, "resource": "test_resource"}, 422), # Missing action
({"resource": "test_resource", "action": "read"}, 422), # Missing service
],
)
@pytest.mark.anyio
async def test_post_perm_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_post_perm_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/iam/permission", json=body)
assert resp.status_code == expected_status
@pytest.mark.parametrize(
"body",
[
{"organisation_id": 1, "service_id": 1, "resource": "test_resource", "action": "read"},
{"organisation_id": 1, "service_id": 1},
{"organisation_id": 1, "resource": "test_resource"},
{"organisation_id": 1, "action": "read"},
{"organisation_id": 1, "service_id": 1, "action": "read"},
{"organisation_id": 1, "service_id": 1, "resource": "test_resource"},
{"organisation_id": 1, "resource": "test_resource", "action": "read"},
],
"body",
[
{
"organisation_id": 1,
"service_id": 1,
"resource": "test_resource",
"action": "read",
},
{"organisation_id": 1, "service_id": 1},
{"organisation_id": 1, "resource": "test_resource"},
{"organisation_id": 1, "action": "read"},
{"organisation_id": 1, "service_id": 1, "action": "read"},
{"organisation_id": 1, "service_id": 1, "resource": "test_resource"},
{"organisation_id": 1, "resource": "test_resource", "action": "read"},
],
)
@pytest.mark.anyio
async def test_post_perm_search_success(default_client: AsyncClient, db_session, body):
@ -478,33 +626,96 @@ async def test_post_perm_search_success(default_client: AsyncClient, db_session,
@pytest.mark.parametrize(
"body, expected_status",
[
# organisation_id tests
({"organisation_id": 42, "service_id": 1, "resource": "test_resource", "action": "read"}, 404), # Non-existent organisation
({"organisation_id": "banana", "service_id": 1, "resource": "test_resource", "action": "read"}, 422), # Invalid organisation ID
({"organisation_id": "", "service_id": 1, "resource": "test_resource", "action": "read"}, 422), # Blank organisation ID
({"organisation_id": -1, "service_id": 1, "resource": "test_resource", "action": "read"}, 422), # Negative organisation ID
# service_id tests
({"organisation_id": 1, "service_id": "banana", "resource": "test_resource", "action": "read"}, 422), # Invalid service ID
({"organisation_id": 1, "service_id": "", "resource": "test_resource", "action": "read"}, 422), # Blank service ID
({"organisation_id": 1, "service_id": -1, "resource": "test_resource", "action": "read"}, 422), # Negative service ID
# resource tests
({"organisation_id": 1, "service_id": 1, "resource": 42, "action": "read"}, 422), # Invalid resource type
# action tests
({"organisation_id": 1, "service_id": 1, "resource": "test_resource", "action": 42}, 422), # Invalid action type
# missing/partial body tests
({}, 422), # Blank body
],
"body, expected_status",
[
# organisation_id tests
(
{
"organisation_id": 42,
"service_id": 1,
"resource": "test_resource",
"action": "read",
},
404,
), # Non-existent organisation
(
{
"organisation_id": "banana",
"service_id": 1,
"resource": "test_resource",
"action": "read",
},
422,
), # Invalid organisation ID
(
{
"organisation_id": "",
"service_id": 1,
"resource": "test_resource",
"action": "read",
},
422,
), # Blank organisation ID
(
{
"organisation_id": -1,
"service_id": 1,
"resource": "test_resource",
"action": "read",
},
422,
), # Negative organisation ID
# service_id tests
(
{
"organisation_id": 1,
"service_id": "banana",
"resource": "test_resource",
"action": "read",
},
422,
), # Invalid service ID
(
{
"organisation_id": 1,
"service_id": "",
"resource": "test_resource",
"action": "read",
},
422,
), # Blank service ID
(
{
"organisation_id": 1,
"service_id": -1,
"resource": "test_resource",
"action": "read",
},
422,
), # Negative service ID
# resource tests
(
{"organisation_id": 1, "service_id": 1, "resource": 42, "action": "read"},
422,
), # Invalid resource type
# action tests
(
{
"organisation_id": 1,
"service_id": 1,
"resource": "test_resource",
"action": 42,
},
422,
), # Invalid action type
# missing/partial body tests
({}, 422), # Blank body
],
)
@pytest.mark.anyio
async def test_post_perm_search_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_post_perm_search_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/iam/permissions/search", json=body)
assert resp.status_code == expected_status

191
test/test_organisation.py
View file

@ -1,6 +1,7 @@
"""
[DELETE] /org/ is not tested because the testing client cannot attach a body to a delete request.
"""
import pytest
from httpx import AsyncClient
@ -24,11 +25,12 @@ async def test_get_org_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["org_id"])
"query, expected_status", generate_query_and_status(["org_id"])
)
@pytest.mark.anyio
async def test_get_org_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_org_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org?{query}")
assert resp.status_code == expected_status
@ -53,18 +55,33 @@ async def test_post_org_success(default_client: AsyncClient):
],
)
@pytest.mark.anyio
async def test_post_org_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_post_org_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/org", json=body)
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient, db_session):
async def test_patch_org_questionnaire_partial_success(
default_client: AsyncClient, db_session
):
org_model = db_session.get(Organisation, 1)
org_model.status = "partial"
db_session.flush()
resp = await default_client.patch("/org/questionnaire", json={"organisation_id": 1, "intake_questionnaire": {"question_one": "new answer one", "question_two": None, "question_three": None}, "partial": True})
resp = await default_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
@ -83,24 +100,56 @@ async def test_patch_org_questionnaire_partial_success(default_client: AsyncClie
({"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),
(
{
"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_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_patch_questionnaire_partial_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/questionnaire", json=body)
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_patch_org_questionnaire_submit_success(default_client: AsyncClient, db_session):
async def test_patch_org_questionnaire_submit_success(
default_client: AsyncClient, db_session
):
org_model = db_session.get(Organisation, 1)
org_model.status = "partial"
db_session.flush()
resp = await default_client.patch("/org/questionnaire", json={"organisation_id": 1, "intake_questionnaire": {"question_one": "new answer one", "question_two": None, "question_three": None}, "partial": False})
resp = await default_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
@ -113,12 +162,13 @@ async def test_patch_org_questionnaire_submit_success(default_client: AsyncClien
@pytest.mark.parametrize(
"status",
["partial", "submitted", "remediation", "approved", "rejected", "removed"]
"status", ["partial", "submitted", "remediation", "approved", "rejected", "removed"]
)
@pytest.mark.anyio
async def test_patch_org_status_success(default_client: AsyncClient, status: str):
resp = await default_client.patch("/org/status", json={"organisation_id": 1, "status": status})
resp = await default_client.patch(
"/org/status", json={"organisation_id": 1, "status": status}
)
data = resp.json()
assert resp.status_code == 200
@ -138,7 +188,9 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
],
)
@pytest.mark.anyio
async def test_patch_org_status_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_patch_org_status_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/status", json=body)
assert resp.status_code == expected_status
@ -161,11 +213,12 @@ async def test_get_org_users_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["org_id"])
"query, expected_status", generate_query_and_status(["org_id"])
)
@pytest.mark.anyio
async def test_get_org_users_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_org_users_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org/users?{query}")
assert resp.status_code == expected_status
@ -173,10 +226,19 @@ async def test_get_org_users_status_checks(default_client: AsyncClient, query: s
@pytest.mark.anyio
async def test_post_org_user_success(default_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.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
resp = await default_client.post("/org/user", json={"organisation_id": 1, "user_id": 2})
resp = await default_client.post(
"/org/user", json={"organisation_id": 1, "user_id": 2}
)
data = resp.json()
assert resp.status_code == 200
@ -197,8 +259,17 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
],
)
@pytest.mark.anyio
async def test_post_org_user_status_checks(default_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"))
async def test_post_org_user_status_checks(
default_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 default_client.post("/org/user", json=body)
@ -208,12 +279,21 @@ async def test_post_org_user_status_checks(default_client: AsyncClient, body: di
@pytest.mark.anyio
async def test_patch_org_root_user_success(default_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.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 default_client.patch("/org/root_user", json={"organisation_id": 1, "user_id": 2})
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
)
data = resp.json()
assert resp.status_code == 200
@ -234,8 +314,17 @@ async def test_patch_org_root_user_success(default_client: AsyncClient, db_sessi
],
)
@pytest.mark.anyio
async def test_patch_root_user_status_checks(default_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"))
async def test_patch_root_user_status_checks(
default_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()
@ -247,10 +336,19 @@ async def test_patch_root_user_status_checks(default_client: AsyncClient, body:
@pytest.mark.anyio
async def test_patch_org_root_user_non_member(default_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.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
resp = await default_client.patch("/org/root_user", json={"organisation_id": 1, "user_id": 2})
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
)
data = resp.json()
assert resp.status_code == 422
@ -269,23 +367,23 @@ async def test_get_org_groups_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["org_id"])
"query, expected_status", generate_query_and_status(["org_id"])
)
@pytest.mark.anyio
async def test_get_org_groups_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_org_groups_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org/groups?{query}")
assert resp.status_code == expected_status
@pytest.mark.parametrize(
"contact_type",
["billing", "security", "owner"]
)
@pytest.mark.parametrize("contact_type", ["billing", "security", "owner"])
@pytest.mark.anyio
async def test_get_org_contact_success(default_client: AsyncClient, contact_type: str):
resp = await default_client.get(f"/org/contact?org_id=1&contact_type={contact_type}")
resp = await default_client.get(
f"/org/contact?org_id=1&contact_type={contact_type}"
)
data = resp.json()
assert resp.status_code == 200
@ -327,7 +425,9 @@ async def test_get_org_contact_success(default_client: AsyncClient, contact_type
],
)
@pytest.mark.anyio
async def test_get_org_contact_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_org_contact_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org/contact?{query}")
assert resp.status_code == expected_status
@ -348,11 +448,16 @@ async def test_get_org_contact_status_checks(default_client: AsyncClient, query:
("address_region", "Glasgow City"),
("country_code", "GB"),
("postal_code", "G1 1AA"),
]
],
)
@pytest.mark.anyio
async def test_patch_org_contact_success(default_client: AsyncClient, key: str, value: str):
resp = await default_client.patch("/org/contact", json={"organisation_id": 1, "contact_type": "billing", key: value})
async def test_patch_org_contact_success(
default_client: AsyncClient, key: str, value: str
):
resp = await default_client.patch(
"/org/contact",
json={"organisation_id": 1, "contact_type": "billing", key: value},
)
data = resp.json()
assert resp.status_code == 200
@ -379,7 +484,9 @@ async def test_patch_org_contact_success(default_client: AsyncClient, key: str,
],
)
@pytest.mark.anyio
async def test_patch_org_contact_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_patch_org_contact_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/contact", json=body)
assert resp.status_code == expected_status

16
test/test_service.py
View file

@ -1,6 +1,7 @@
"""
409 on [POST]/service/ not tested because SQLite throws a different error than Postgres
"""
import pytest
from httpx import AsyncClient
@ -19,11 +20,12 @@ async def test_get_services_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["org_id"])
"query, expected_status", generate_query_and_status(["org_id"])
)
@pytest.mark.anyio
async def test_get_services_status_checks(default_client: AsyncClient, query: str, expected_status: int):
async def test_get_services_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/service/?{query}")
assert resp.status_code == expected_status
@ -49,7 +51,9 @@ async def test_post_service_success(default_client: AsyncClient):
],
)
@pytest.mark.anyio
async def test_post_services_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_post_services_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/service/", json=body)
assert resp.status_code == expected_status
@ -77,7 +81,9 @@ async def test_patch_service_success(default_client: AsyncClient):
],
)
@pytest.mark.anyio
async def test_patch_services_status_checks(default_client: AsyncClient, body: dict[str, str], expected_status: int):
async def test_patch_services_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/service/key", json=body)
assert resp.status_code == expected_status

44
test/test_user.py
View file

@ -8,38 +8,40 @@ from httpx import AsyncClient
from .conftest import generate_query_and_status
@pytest.mark.anyio
async def test_get_self_db_success(default_client: AsyncClient):
resp = await default_client.get("/user/self/db")
data = resp.json()
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 "groups" in data
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 "groups" in data
@pytest.mark.anyio
async def test_get_user_success(default_client: AsyncClient):
resp = await default_client.get("/user/?user_id=1")
data = resp.json()
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 "groups" in data
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 "groups" in data
@pytest.mark.anyio
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["user_id"])
"query, expected_status", generate_query_and_status(["user_id"])
)
async def test_get_user_status_checks(default_client: AsyncClient, query: str, expected_status: int):
resp = await default_client.get(f"/user/?{query}")
async def test_get_user_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/user/?{query}")
assert resp.status_code == expected_status
assert resp.status_code == expected_status