Compare commits

...

5 commits

Author SHA1 Message Date
a655eaf543 tests: delete_group_perm perm not in group
All checks were successful
ci / lint_and_test (push) Successful in 14s
Issue: #24
2026-06-12 13:17:07 +01:00
fe2171df44 tests: delete group perm status checks
Issue: #24
2026-06-12 13:08:58 +01:00
c0b9763669 tests: caor status checks match new model
Issue: #24
2026-06-12 13:01:04 +01:00
fc9d7f8536 tests: body param generator
Issue: #24
2026-06-12 12:54:48 +01:00
778f1dbece tests: remove db modifications from individual tests
All db seeding now down in conftest
2026-06-12 11:29:42 +01:00
9 changed files with 268 additions and 547 deletions

View file

@ -8,7 +8,7 @@ from sqlalchemy.orm import sessionmaker
from src.user.models import User
from src.service.models import Service
from src.organisation.models import Organisation as Org
from src.organisation.models import Organisation as Org, OrgUsers
from src.contact.models import Contact
from src.iam.models import Group, Permission
from src.auth.service import get_current_user, get_dev_user
@ -91,13 +91,35 @@ def _seed(db):
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.add(
User(
email="user@orgone.com",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-qwer",
)
)
db.add(
User(
email="root@orgtwo.com",
first_name="Root",
last_name="Test",
oidc_id="abcd-efgh-ijkl-hjkl",
)
)
db.add(Contact(org_id=1, email="billing@orgone.com", phonenumber="07521539927"))
db.add(Contact(org_id=1, email="owner@orgone.com", phonenumber="07521539927"))
db.add(Contact(org_id=1, email="security@orgone.com", phonenumber="07521539927"))
db.add(Contact(org_id=2, email="billing@orgtwo.com", phonenumber="07521539927"))
db.add(Contact(org_id=2, email="owner@orgtwo.com", phonenumber="07521539927"))
db.add(Contact(org_id=2, email="security@orgtwo.com", phonenumber="07521539927"))
db.add(Contact(org_id=3, email="billing@orgthree.com", phonenumber="07521539927"))
db.add(Contact(org_id=3, email="owner@orgthree.com", phonenumber="07521539927"))
db.add(Contact(org_id=3, email="security@orgthree.com", phonenumber="07521539927"))
db.flush()
db.add(
Org(
name="Test Org",
name="Org One",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
@ -109,9 +131,41 @@ def _seed(db):
},
)
)
db.add(
Org(
name="Org Two",
root_user_id=3,
billing_contact_id=4,
owner_contact_id=5,
security_contact_id=6,
status="approved",
intake_questionnaire={
"metadata": {"version": 0, "submission_date": None},
"questions": {"question_two": "answer two"},
},
)
)
db.add(
Org(
name="Org Three",
root_user_id=1,
billing_contact_id=7,
owner_contact_id=8,
security_contact_id=9,
status="partial",
intake_questionnaire={
"metadata": {"version": 0, "submission_date": None},
"questions": {"question_two": "answer two"},
},
)
)
db.add(OrgUsers(org_id=1, user_id=2))
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))
db.add(Permission(service_id=1, resource="test_resource", action="move"))
db.add(Group(name="Org One Group", org_id=1))
db.add(Group(name="Org Two Group", org_id=2))
db.add(Group(name="Org One Group Two", org_id=1))
db.flush()
group_model = db.get(Group, 1)
perm_model = db.get(Permission, 1)
@ -159,6 +213,45 @@ def generate_query_and_status(params) -> list[tuple[str, int]]:
return query_and_status
def generate_body_and_status(params: dict[str, str]) -> list[tuple[dict, int]]:
possible_values_int = [0, -1, 42, "banana", ""]
possible_values_str = [0]
defaults = [{param: 1 for param in params.keys()}]
# Missing params
body_list = [
{key: ("valid string" if params[key] == "str" else 1) for key in combo}
for r in range(len(defaults[0].keys()) + 1)
for combo in combinations(defaults[0].keys(), r)
]
# Complete body as default for generating invalid checks
default_body = body_list.pop(-1)
# Generates checks for each param being invalid
for param, typ in params.items():
if typ == "int":
possible_values = possible_values_int
elif typ == "str":
possible_values = possible_values_str
else:
raise TypeError(f"Unknown type {typ}")
for value in possible_values:
new_record = default_body.copy()
new_record[param] = value
body_list.append(new_record)
body_and_status = []
# Assign expected status
for body in body_list:
# ID 42 is used to represent a non-existent entry. So it should 404.
status = 404 if 42 in body.values() else 422
body_and_status.append((body, status))
return body_and_status
def get_testable_routes():
routes = []

View file

@ -7,27 +7,15 @@ Delete endpoints are currently skipped because the testing system cannot use bod
import pytest
from httpx import AsyncClient
from src.organisation.models import Organisation as Org, OrgUsers
from src.user.models import User
from src.iam.models import Group
pytestmark = [
pytest.mark.auth,
pytest.mark.preapproval,
]
@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(default_client: AsyncClient):
resp = await default_client.get("/org?org_id=1")
resp = await default_client.get("/org?org_id=3")
assert resp.status_code != 422
assert resp.status_code == 200
@ -37,7 +25,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
resp = await default_client.patch(
"/org/questionnaire",
json={
"organisation_id": 1,
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
@ -53,7 +41,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
@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"}
"/org/status", json={"organisation_id": 3, "status": "submitted"}
)
assert resp.status_code != 422
assert resp.status_code == 200
@ -61,48 +49,24 @@ async def test_patch_org_status_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_get_org_users_auth_approval(default_client: AsyncClient):
resp = await default_client.get("/org/users?org_id=1")
resp = await default_client.get("/org/users?org_id=3")
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(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()
async def test_post_org_user_auth_approval(default_client: AsyncClient):
resp = await default_client.post(
"/org/user", json={"organisation_id": 1, "user_id": 2}
"/org/user", json={"organisation_id": 3, "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",
)
)
db_session.flush()
db_session.add(OrgUsers(org_id=1, user_id=2))
db_session.flush()
async def test_patch_org_root_user_auth_approval(default_client: AsyncClient):
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
"/org/root_user", json={"organisation_id": 3, "user_id": 2}
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@ -110,14 +74,14 @@ async def test_patch_org_root_user_auth_approval(
@pytest.mark.anyio
async def test_get_org_groups_auth_approval(default_client: AsyncClient):
resp = await default_client.get("/org/groups?org_id=1")
resp = await default_client.get("/org/groups?org_id=3")
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(default_client: AsyncClient):
resp = await default_client.get("/org/contact?org_id=1&contact_type=billing")
resp = await default_client.get("/org/contact?org_id=3&contact_type=billing")
assert resp.status_code != 422
assert resp.status_code == 200
@ -127,7 +91,7 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
resp = await default_client.patch(
"/org/contact",
json={
"organisation_id": 1,
"organisation_id": 3,
"contact_type": "billing",
"email": "user@example.com",
},
@ -138,21 +102,21 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
@pytest.mark.anyio
async def test_get_service_auth_approval(default_client: AsyncClient):
resp = await default_client.get("/service?org_id=1")
resp = await default_client.get("/service?org_id=3")
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(default_client: AsyncClient):
resp = await default_client.get("/iam/group/permissions?org_id=1&group_id=1")
resp = await default_client.get("/iam/group/permissions?org_id=3&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(default_client: AsyncClient):
resp = await default_client.get("/iam/group/users?org_id=1&group_id=1")
resp = await default_client.get("/iam/group/users?org_id=3&group_id=1")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@ -160,42 +124,26 @@ 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}
"/iam/group", json={"name": "New Group", "organisation_id": 3}
)
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
):
db_session.add(Group(name="Test Group Two", org_id=1))
db_session.flush()
async def test_put_iam_group_permission_auth_approval(default_client: AsyncClient):
resp = await default_client.put(
"/iam/group/permission",
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
json={"permission_id": 1, "group_id": 2, "organisation_id": 3},
)
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",
)
)
db_session.flush()
async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
resp = await default_client.put(
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 3}
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@ -203,7 +151,7 @@ async def test_put_iam_group_user_auth_approval(
@pytest.mark.anyio
async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
resp = await default_client.get("/iam/permissions?org_id=1")
resp = await default_client.get("/iam/permissions?org_id=3")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@ -211,7 +159,7 @@ 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"}
"/iam/permissions/search", json={"organisation_id": 3, "action": "read"}
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]

View file

@ -3,9 +3,6 @@
import pytest
from httpx import AsyncClient
from src.organisation.models import Organisation as Org
from src.user.models import User
pytestmark = [
pytest.mark.auth,
@ -13,34 +10,9 @@ pytestmark = [
@pytest.mark.anyio
async def test_get_org_auth_root_su(default_client: AsyncClient, db_session):
async def test_get_org_auth_root_su(default_client: AsyncClient):
# 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.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={
"metadata": {"version": 0, "submission_date": None},
"questions": {},
},
)
)
db_session.flush()
resp = await default_client.get("/org?org_id=2")
assert resp.status_code != 422
assert resp.status_code == 200
assert resp.json()["organisations"][0]["name"] == "Test Org Two"
assert resp.json()["organisations"][0]["name"] == "Org Two"

View file

@ -6,42 +6,12 @@ DELETE endpoints are not tested
import pytest
from httpx import AsyncClient
from src.organisation.models import Organisation as Org
from src.user.models import User
from src.iam.models import Group
pytestmark = [
pytest.mark.auth,
pytest.mark.root_user,
]
@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.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.flush()
@pytest.mark.anyio
async def test_get_org_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/org?org_id=2")
@ -143,11 +113,7 @@ async def test_post_iam_group_auth_root(no_su_client: AsyncClient):
@pytest.mark.anyio
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()
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},
@ -158,17 +124,9 @@ async def test_put_iam_group_permission_auth_root(
@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.flush()
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}
)

View file

@ -6,10 +6,6 @@ DELETE endpoints are not tested
import pytest
from httpx import AsyncClient
from src.organisation.models import OrgUsers
from src.user.models import User
pytestmark = [
pytest.mark.auth,
pytest.mark.super_admin,
@ -35,19 +31,7 @@ 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.flush()
db_session.add(OrgUsers(org_id=1, user_id=2))
db_session.flush()
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}
)
@ -73,7 +57,7 @@ async def test_post_service_auth_su(no_su_client: AsyncClient):
@pytest.mark.anyio
async def test_post_perm_auth_su(no_su_client: AsyncClient, db_session):
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"},
@ -84,17 +68,7 @@ async def test_post_perm_auth_su(no_su_client: AsyncClient, db_session):
@pytest.mark.anyio
async def test_post_org_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.flush()
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}
)

View file

@ -3,12 +3,7 @@
import pytest
from httpx import AsyncClient
from src.user.models import User
from src.organisation.models import Organisation as Org, OrgUsers
from src.iam.models import Group
from .conftest import generate_query_and_status
from .conftest import generate_query_and_status, generate_body_and_status
pytestmark = [
pytest.mark.iam_module,
@ -20,7 +15,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
body = {
"rn": {
"service": "Test Service",
"organisation": "Test Org",
"organisation": "Org One",
"resource": "test_resource",
"instance": None,
},
@ -45,7 +40,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
)
@pytest.mark.anyio
async def test_act_on_resource_wrong_key(
default_client: AsyncClient, db_session, service: str, api_key: str
default_client: AsyncClient, service: str, api_key: str
):
body = {
"rn": {
@ -97,19 +92,24 @@ async def test_act_on_resource_missing_key(default_client: AsyncClient):
("Test Service", 42, "test_resource", "read", 422),
("Test Service", "Test Org", None, "read", 422),
("Test Service", "Test Org", 42, "read", 422),
("Test Service", "Test Org", "test_resource", None, 422),
("Test Service", "Test Org", "test_resource", 42, 422),
],
)
@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}
body = {
"rn": {"service": service, "organisation": org, "resource": resource},
"action": action,
}
headers = {
"Authorization": "Bearer not_checked_when_auth_is_disabled",
"X-API-Key": "123456789",
}
resp = await default_client.post(
f"/iam/can_act_on_resource?action={action}", json=body, headers=headers
"/iam/can_act_on_resource", json=body, headers=headers
)
assert resp.status_code == expected_status
@ -118,16 +118,15 @@ async def test_act_on_resource_endpoint_status_checks(
@pytest.mark.parametrize(
"service, org, resource, action, expected_response",
[
("Test Service", "Test Org", "test_resource", "read", True),
("Test Service", "Test Org", "test_resource", "create", False),
("Test Service", "Test Org", "no_access_here", "read", False),
("Test Service", "Test Org Two", "test_resource", "read", False),
("Test Service", "Org One", "test_resource", "read", True),
("Test Service", "Org One", "test_resource", "create", False),
("Test Service", "Org One", "no_access_here", "read", False),
("Test Service", "Org Two", "test_resource", "read", False),
],
)
@pytest.mark.anyio
async def test_act_on_resource_logic(
default_client: AsyncClient,
db_session,
service,
org,
resource,
@ -173,7 +172,7 @@ async def test_get_group_permissions_success(default_client: AsyncClient):
)
@pytest.mark.anyio
async def test_get_group_permissions_status_checks(
default_client: AsyncClient, db_session, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/iam/group/permissions?{query}")
@ -188,21 +187,7 @@ async def test_get_group_permissions_status_checks(
],
)
@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",
)
)
db_session.add(Group(name="Another Test Group", org_id=2))
db_session.flush()
async def test_get_group_permissions_mismatch(default_client: AsyncClient, query: str):
resp = await default_client.get(f"/iam/group/permissions?{query}")
assert resp.status_code == 403
@ -226,12 +211,12 @@ async def test_get_group_users_success(default_client: AsyncClient):
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["id"] == 1
assert data["group"]["name"] == "Test Group"
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"] == "Test Org"
assert data["organisation"]["name"] == "Org One"
@pytest.mark.parametrize(
@ -254,21 +239,7 @@ async def test_get_group_users_status_checks(
],
)
@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",
)
)
db_session.add(Group(name="Another Test Group", org_id=2))
db_session.flush()
async def test_get_group_users_mismatch(default_client: AsyncClient, query: str):
resp = await default_client.get(f"/iam/group/users?{query}")
assert resp.status_code == 403
@ -287,37 +258,12 @@ async def test_post_group_success(default_client: AsyncClient):
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "New Group"
assert data["group"]["id"] == 2
assert data["group"]["id"] == 4
@pytest.mark.parametrize(
"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
({"name": 1}, 422), # Only name
({}, 422), # Blank body
({"organisation_id": "", "name": ""}, 422), # Both blank
({"organisation_id": 1, "name": 42}, 422), # Valid organisation, invalid name
({"organisation_id": 1, "name": ""}, 422), # Valid organisation, blank name
({"organisation_id": 1, "name": "hi"}, 422), # Valid organisation, small name
({"organisation_id": 1}, 422), # Only organisation ID
],
generate_body_and_status({"organisation_id": "int", "name": "str"}),
)
@pytest.mark.anyio
async def test_post_group_status_checks(
@ -329,12 +275,19 @@ async def test_post_group_status_checks(
@pytest.mark.anyio
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()
async def test_post_group_conflict(default_client: AsyncClient):
resp = await default_client.post(
"/iam/group", json={"organisation_id": 1, "name": "Org One Group"}
)
assert resp.status_code == 409
@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": 2, "organisation_id": 1},
json={"permission_id": 1, "group_id": 3, "organisation_id": 1},
)
assert resp.status_code == 200
@ -342,8 +295,8 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Test Group Two"
assert data["group"]["id"] == 2
assert data["group"]["name"] == "Org One Group Two"
assert data["group"]["id"] == 3
assert "permissions" in data
assert isinstance(data["permissions"], list)
@ -357,67 +310,9 @@ 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
({}, 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
],
generate_body_and_status(
{"organisation_id": "int", "group_id": "int", "permission_id": "int"}
),
)
@pytest.mark.anyio
async def test_put_group_perm_status_checks(
@ -428,6 +323,16 @@ async def test_put_group_perm_status_checks(
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_put_group_perm_conflict(default_client: AsyncClient):
resp = await default_client.put(
"/iam/group/permission",
json={"organisation_id": 1, "group_id": 1, "permission_id": 1},
)
assert resp.status_code == 409
@pytest.mark.parametrize(
"body",
[
@ -436,21 +341,7 @@ async def test_put_group_perm_status_checks(
],
)
@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",
)
)
db_session.add(Group(name="Another Test Group", org_id=2))
db_session.flush()
async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
resp = await default_client.put("/iam/group/permission", json=body)
assert resp.status_code == 403
@ -458,19 +349,7 @@ async def test_put_group_perm_mismatch(
@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.flush()
db_session.add(OrgUsers(user_id=2, org_id=1))
db_session.flush()
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}
)
@ -480,7 +359,7 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Test Group"
assert data["group"]["name"] == "Org One Group"
assert data["group"]["id"] == 1
assert "users" in data
@ -490,59 +369,14 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
assert user["id"] == 2
assert user["first_name"] == "User"
assert user["last_name"] == "Test"
assert user["email"] == "user@test.org"
assert user["email"] == "user@orgone.com"
@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": 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": ""}, 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
],
generate_body_and_status(
{"organisation_id": "int", "group_id": "int", "user_id": "int"}
),
)
@pytest.mark.anyio
async def test_put_group_user_status_checks(
@ -582,7 +416,7 @@ async def test_get_permissions_status_checks(
@pytest.mark.anyio
async def test_post_perm_success(default_client: AsyncClient, db_session):
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"},
@ -594,7 +428,7 @@ async def test_post_perm_success(default_client: AsyncClient, db_session):
assert "permission" in data
assert isinstance(data["permission"], dict)
assert data["permission"]["id"] == 2
assert data["permission"]["id"] == 3
assert data["permission"]["service_name"] == "Test Service"
assert data["permission"]["resource"] == "test_resource"
assert data["permission"]["action"] == "create"
@ -671,7 +505,7 @@ async def test_post_perm_status_checks(
],
)
@pytest.mark.anyio
async def test_post_perm_search_success(default_client: AsyncClient, db_session, body):
async def test_post_perm_search_success(default_client: AsyncClient, body):
resp = await default_client.post("/iam/permissions/search", json=body)
data = resp.json()
@ -679,7 +513,11 @@ async def test_post_perm_search_success(default_client: AsyncClient, db_session,
assert "permissions" in data
assert isinstance(data["permissions"], list)
permission = data["permissions"][0]
permissions_filtered = [
permission for permission in data["permissions"] if permission["id"] == 1
]
assert len(permissions_filtered) == 1
permission = permissions_filtered[0]
assert permission["id"] == 1
assert permission["service_name"] == "Test Service"
assert permission["resource"] == "test_resource"
@ -795,7 +633,27 @@ async def test_delete_group_permissions_success(default_client: AsyncClient):
assert len(data["permissions"]) == 0
assert "group" in data
assert data["group"]["id"] == 1
assert data["group"]["name"] == "Test Group"
assert data["group"]["name"] == "Org One Group"
@pytest.mark.parametrize(
"query, expected_status",
generate_query_and_status(["group_id", "org_id", "perm_id"]),
)
@pytest.mark.anyio
async def test_delete_group_permissions_status_checks(
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.delete(f"/iam/group/permission?{query}")
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_delete_group_permissions_not_in_group(default_client: AsyncClient):
resp = await default_client.delete(
"/iam/group/permission?org_id=1&group_id=1&perm_id=2"
)
assert resp.status_code == 422
@pytest.mark.anyio
@ -816,4 +674,4 @@ async def test_delete_group_users_success(default_client: AsyncClient):
assert len(data["users"]) == 0
assert "group" in data
assert data["group"]["id"] == 1
assert data["group"]["name"] == "Test Group"
assert data["group"]["name"] == "Org One Group"

View file

@ -5,8 +5,6 @@
import pytest
from httpx import AsyncClient
from src.organisation.models import Organisation, OrgUsers
from src.user.models import User
from .conftest import generate_query_and_status
@ -26,15 +24,15 @@ async def test_get_org_success(default_client: AsyncClient):
assert isinstance(org, dict)
assert org["organisation_id"] == 1
assert org["name"] == "Test Org"
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@test.org"
assert org["owner_contact"]["email"] == "owner@test.org"
assert org["security_contact"]["email"] == "security@test.org"
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(
@ -62,7 +60,7 @@ async def test_post_org_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"body, expected_status",
[
({"name": "Test Org"}, 409),
({"name": "Org One"}, 409),
({"name": 42}, 422),
({}, 422),
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
@ -78,16 +76,11 @@ async def test_post_org_status_checks(
@pytest.mark.anyio
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()
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient):
resp = await default_client.patch(
"/org/questionnaire",
json={
"organisation_id": 1,
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
@ -99,7 +92,7 @@ async def test_patch_org_questionnaire_partial_success(
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Test Org"
assert data["name"] == "Org Three"
assert data["status"] == "partial"
assert "intake_questionnaire" in data
assert isinstance(data["intake_questionnaire"], dict)
@ -116,7 +109,7 @@ async def test_patch_org_questionnaire_partial_success(
"body, expected_status",
[
({"organisation_id": 42}, 404),
({"organisation_id": "Test Org"}, 422),
({"organisation_id": "Org One"}, 422),
({"organisation_id": ""}, 422),
({}, 422),
(
@ -151,16 +144,11 @@ async def test_patch_questionnaire_partial_status_checks(
@pytest.mark.anyio
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()
async def test_patch_org_questionnaire_submit_success(default_client: AsyncClient):
resp = await default_client.patch(
"/org/questionnaire",
json={
"organisation_id": 1,
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
@ -172,7 +160,7 @@ async def test_patch_org_questionnaire_submit_success(
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Test Org"
assert data["name"] == "Org Three"
assert data["status"] == "submitted"
assert "intake_questionnaire" in data
assert isinstance(data["intake_questionnaire"], dict)
@ -196,7 +184,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Test Org"
assert data["name"] == "Org One"
assert data["status"] == status
@ -204,7 +192,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
"body, expected_status",
[
({"organisation_id": 42}, 404),
({"organisation_id": "Test Org"}, 422),
({"organisation_id": "Org One"}, 422),
({"organisation_id": ""}, 422),
({}, 422),
({"organisation_id": "1", "status": True}, 422),
@ -229,7 +217,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
assert "users" in data
assert isinstance(data["users"], list)
assert len(data["users"]) == 1
assert len(data["users"]) == 2
user = data["users"][0]
assert isinstance(user, dict)
@ -237,7 +225,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
assert user["id"] == 1
assert "organisation" in data
assert data["organisation"]["name"] == "Test Org"
assert data["organisation"]["name"] == "Org One"
assert data["organisation"]["id"] == 1
@ -254,19 +242,9 @@ async def test_get_org_users_status_checks(
@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.flush()
async def test_post_org_user_success(default_client: AsyncClient):
resp = await default_client.post(
"/org/user", json={"organisation_id": 1, "user_id": 2}
"/org/user", json={"organisation_id": 1, "user_id": 3}
)
assert resp.status_code == 200
@ -276,12 +254,12 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
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"] == "user@test.org"]) == 1
len([user for user in data["users"] if user["email"] == "root@orgtwo.com"]) == 1
)
@ -298,37 +276,15 @@ 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
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
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)
assert resp.status_code == expected_status
@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.flush()
db_session.add(OrgUsers(org_id=1, user_id=2))
db_session.flush()
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}
)
@ -336,15 +292,15 @@ async def test_patch_org_root_user_success(default_client: AsyncClient, db_sessi
data = resp.json()
assert data["name"] == "Test Org"
assert data["root_user_email"] == "user@test.org"
assert data["name"] == "Org One"
assert data["root_user_email"] == "user@orgone.com"
@pytest.mark.parametrize(
"body, expected_status",
[
({"organisation_id": 42, "user_id": 2}, 404),
({"organisation_id": "Test Org", "user_id": 2}, 422),
({"organisation_id": "Org One", "user_id": 2}, 422),
({"organisation_id": "", "user_id": 2}, 422),
({}, 422),
({"user_id": 2}, 422),
@ -354,39 +310,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
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
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=body)
assert resp.status_code == expected_status
@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.flush()
async def test_patch_org_root_user_non_member(default_client: AsyncClient):
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
"/org/root_user", json={"organisation_id": 1, "user_id": 3}
)
data = resp.json()
@ -404,14 +338,14 @@ async def test_get_org_groups_success(default_client: AsyncClient):
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
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"] == "Test Group"
assert group["name"] == "Org One Group"
@pytest.mark.parametrize(
@ -438,7 +372,7 @@ async def test_get_org_contact_success(default_client: AsyncClient, contact_type
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
assert data["organisation"]["name"] == "Org One"
attributes = [
"email",
@ -516,7 +450,7 @@ async def test_patch_org_contact_success(
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
assert data["organisation"]["name"] == "Org One"
attributes = [
"email",
@ -557,7 +491,7 @@ async def test_patch_org_contact_success(
({"organisation_id": 42, "contact_type": "billing"}, 404),
({"organisation_id": 1, "contact_type": "security"}, 200),
({"organisation_id": 1, "contact_type": "owner"}, 200),
({"organisation_id": "Test Org", "contact_type": "billing"}, 422),
({"organisation_id": "Org One", "contact_type": "billing"}, 422),
({"organisation_id": "", "contact_type": "billing"}, 422),
({}, 422),
({"organisation_id": 1, "contact_type": "not_real"}, 422),
@ -582,26 +516,14 @@ async def test_delete_org_success(default_client: AsyncClient):
@pytest.mark.anyio
async def test_delete_org_users_success(db_session, default_client: AsyncClient):
db_session.add(
User(
email="user@test.org",
first_name="User",
last_name="Test",
oidc_id="abcd-efgh-ijkl-1234",
)
)
db_session.flush()
async def test_delete_org_users_success(default_client: AsyncClient):
resp = await default_client.delete("/org/user?org_id=1&user_id=2")
assert resp.status_code == 204
@pytest.mark.anyio
async def test_delete_preapproval_org_success(db_session, default_client: AsyncClient):
org_model = db_session.get(Organisation, 1)
org_model.status = "partial"
db_session.flush()
resp = await default_client.delete("/org/self?org_id=1")
async def test_delete_preapproval_org_success(default_client: AsyncClient):
resp = await default_client.delete("/org/self?org_id=3")
assert resp.status_code == 204

View file

@ -5,8 +5,7 @@
import pytest
from httpx import AsyncClient
from .conftest import generate_query_and_status
from .conftest import generate_query_and_status, generate_body_and_status
pytestmark = [
pytest.mark.service_module,
@ -51,12 +50,7 @@ async def test_post_service_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"body, expected_status",
[
({"name": "Test Service"}, 409),
({"name": 42}, 422),
({}, 422),
],
"body, expected_status", generate_body_and_status({"name": "str"})
)
@pytest.mark.anyio
async def test_post_service_status_checks(
@ -67,6 +61,13 @@ async def test_post_service_status_checks(
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_post_service_conflict(default_client: AsyncClient):
resp = await default_client.post("/service", json={"name": "Test Service"})
assert resp.status_code == 409
@pytest.mark.anyio
async def test_patch_service_success(default_client: AsyncClient):
resp = await default_client.patch("/service/key", json={"service_id": 1})
@ -82,12 +83,7 @@ async def test_patch_service_success(default_client: AsyncClient):
@pytest.mark.parametrize(
"body, expected_status",
[
({"service_id": 42}, 404),
({"service_id": "Test Service"}, 422),
({"service_id": ""}, 422),
({}, 422),
],
generate_body_and_status({"service_id": "int"}),
)
@pytest.mark.anyio
async def test_patch_services_status_checks(

View file

@ -74,7 +74,7 @@ async def test_post_user_invitation_success(default_client: AsyncClient):
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
assert data["organisation"]["name"] == "Org One"
assert "invited_email" in data
assert isinstance(data["invited_email"], str)
@ -135,22 +135,22 @@ async def test_get_self_orgs_success(default_client: AsyncClient):
org = data["organisations"][0]
assert org["organisation_id"] == 1
assert org["name"] == "Test Org"
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@test.org"
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@test.org"
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@test.org"
assert org["security_contact"]["email"] == "security@orgone.com"
assert org["security_contact"]["id"] == 3
@ -162,12 +162,12 @@ async def test_get_self_orgs_dynamic(default_client: AsyncClient):
"organisations": [
{
"organisation_id": 1,
"name": "Test Org",
"name": "Org One",
"status": "approved",
"root_user_email": "admin@test.com",
"owner_contact": {"email": "owner@test.org", "id": 2},
"security_contact": {"email": "security@test.org", "id": 3},
"billing_contact": {"email": "billing@test.org", "id": 1},
"owner_contact": {"email": "owner@orgone.com", "id": 2},
"security_contact": {"email": "security@orgone.com", "id": 3},
"billing_contact": {"email": "billing@orgone.com", "id": 1},
"intake_questionnaire": {
"questions": {
"question_one": None,