Compare commits

...

6 commits

Author SHA1 Message Date
76e889d836 tests: expanded assertions
Some checks failed
ci / lint_and_test (push) Failing after 8s
New assertions added for new data being delivered (eg IDs on endpoints not previously serving IDs)
2026-06-09 14:43:52 +01:00
607f736453 feat: user ids return on get org users 2026-06-09 14:42:37 +01:00
a215d11df9 feat: id returned with permission details 2026-06-09 14:04:21 +01:00
bace6388aa tests: pytest module markers 2026-06-09 13:58:08 +01:00
4ff9edf6d1 feat: schema mixins moved to project level
Resolves circular dependency issues.
2026-06-09 13:17:31 +01:00
e9fe405e06 feat: add org user by id requires su
Part of the "sensical user adding" changes.
2026-06-09 13:07:43 +01:00
18 changed files with 242 additions and 103 deletions

View file

@ -272,6 +272,7 @@ async def create_new_permission(
):
raise ConflictException(message="Permission already exists")
response = {
"id": perm_model.id,
"service_name": perm_model.service_name,
"resource": perm_model.resource,
"action": perm_model.action,

View file

@ -3,7 +3,6 @@ Pydantic models for the IAM module
Models follow the nomenclature of:
- Sub-models: "<Resource><Opt:>Schema"
- Mixins: "<Attribute>Mixin"
- Models: "<Module><Method><Resource><Opt:Resource><Direction>" ie "IAMGetGroupPermissionsResponse"
"""
@ -11,10 +10,14 @@ from typing import Optional, Annotated
from pydantic import EmailStr, ConfigDict, Field
from src.service.schemas import ServiceIDMixin
from src.organisation.schemas import OrgIDMixin
from src.schemas import CustomBaseModel
from src.user.schemas import UserIDMixin
from src.schemas import (
CustomBaseModel,
ServiceIDMixin,
OrgIDMixin,
UserIDMixin,
PermIDMixin,
GroupIDMixin,
)
class UserSchema(CustomBaseModel):
@ -29,6 +32,7 @@ class UserSchema(CustomBaseModel):
class PermissionSchema(CustomBaseModel):
model_config = ConfigDict(from_attributes=True, extra="ignore")
id: int
service_name: str
resource: str
action: str
@ -39,14 +43,6 @@ class GroupSchema(CustomBaseModel):
name: str
class GroupIDMixin(CustomBaseModel):
group_id: int = Field(gt=0)
class PermIDMixin(CustomBaseModel):
permission_id: int = Field(gt=0)
class IAMGetGroupPermissionsResponse(CustomBaseModel):
permissions: list[PermissionSchema]

View file

@ -265,7 +265,7 @@ async def get_users(org_model: org_model_root_claim_query_dependency):
Returns a list of the email addresses of all users of the organisation.
"""
return {
"users": [user.email for user in org_model.user_rel],
"users": [{"email": user.email, "id": user.id} for user in org_model.user_rel],
"organisation": org_model,
}
@ -292,8 +292,9 @@ async def get_users(org_model: org_model_root_claim_query_dependency):
)
async def add_user_to_org(
db: db_dependency,
org_model: org_model_root_claim_body_dependency,
org_model: org_model_body_dependency,
user_model: user_model_body_dependency,
su: super_admin_dependency,
request_model: OrgPostUserRequest,
):
"""

View file

@ -3,25 +3,19 @@ Pydantic models for organisation module
Models follow the nomenclature of:
- Sub-models: "<Resource><Opt:>Schema"
- Mixins: "<Attribute>Mixin"
- Models: "<Module><Method><Resource><Opt:Resource><Direction>" ie "OrgPostOrgRequest"
"""
from typing import Optional
from pydantic import EmailStr, ConfigDict, Field
from pydantic import EmailStr, ConfigDict
from src.schemas import CustomBaseModel
from src.schemas import CustomBaseModel, OrgIDMixin, UserIDMixin
from src.contact.schemas import ContactModel
from src.user.schemas import UserIDMixin
from src.organisation.constants import Status, ContactType
class OrgIDMixin(CustomBaseModel):
organisation_id: int = Field(gt=0)
class Questionnaire(CustomBaseModel):
question_one: Optional[str] = None
question_two: Optional[str] = None
@ -98,7 +92,7 @@ class OrgPatchRootResponse(CustomBaseModel):
class OrgGetUserResponse(CustomBaseModel):
users: list[str]
users: list[dict[str, str | int]]
organisation: OrgSchema

View file

@ -6,7 +6,7 @@ Exports:
- ResourceName
"""
from pydantic import BaseModel
from pydantic import BaseModel, Field
from typing import Optional
@ -19,3 +19,24 @@ class ResourceName(CustomBaseModel):
organisation: str
resource: str
instance: Optional[str] = None
### Mixins ###
class OrgIDMixin(CustomBaseModel):
organisation_id: int = Field(gt=0)
class GroupIDMixin(CustomBaseModel):
group_id: int = Field(gt=0)
class PermIDMixin(CustomBaseModel):
permission_id: int = Field(gt=0)
class ServiceIDMixin(CustomBaseModel):
service_id: int = Field(gt=0)
class UserIDMixin(CustomBaseModel):
user_id: int = Field(gt=0)

View file

@ -3,17 +3,12 @@ Pydantic models for service module
Models follow the nomenclature of:
- Sub-models: "<Resource><Opt:>Schema"
- Mixins: "<Attribute>Mixin"
- Models: "<Module><Method><Resource><Opt:Resource><Direction>" ie "ServiceGetServiceResponse"
"""
from pydantic import ConfigDict, Field
from pydantic import ConfigDict
from src.schemas import CustomBaseModel
class ServiceIDMixin(CustomBaseModel):
service_id: int = Field(gt=0)
from src.schemas import CustomBaseModel, ServiceIDMixin
class ServiceSchema(CustomBaseModel):

View file

@ -15,7 +15,7 @@ from src.user.models import User
from src.auth.service import claims_dependency
from src.database import db_dependency
from src.user.schemas import UserIDMixin
from src.schemas import UserIDMixin
async def get_user_model_claims(claims: claims_dependency, db: db_dependency):

View file

@ -3,13 +3,9 @@ Pydantic models for the user module
"""
from typing import Optional
from pydantic import Field, EmailStr
from pydantic import EmailStr
from src.schemas import CustomBaseModel
class UserIDMixin(CustomBaseModel):
user_id: int = Field(gt=0)
from src.schemas import CustomBaseModel, OrgIDMixin
class OIDCClaims(CustomBaseModel):
@ -52,13 +48,7 @@ class UserResponse(CustomBaseModel):
groups: Optional[dict[str, list[dict[str, str | int]]]] = None
class OrgResponse(CustomBaseModel):
org_id: int
name: str
class UserPostInvitationRequest(CustomBaseModel):
organisation_id: int
class UserPostInvitationRequest(OrgIDMixin):
user_email: EmailStr

12
test/pytest.toml Normal file
View file

@ -0,0 +1,12 @@
[tool.pytest]
markers = [
"iam_module",
"org_module",
"service_module",
"user_module",
"auth",
"root_user",
"super_admin",
"user",
"preapproval"
]

View file

@ -12,6 +12,12 @@ 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)

View file

@ -7,6 +7,11 @@ from src.organisation.models import Organisation as Org
from src.user.models import User
pytestmark = [
pytest.mark.auth,
]
@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

View file

@ -11,6 +11,12 @@ 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(
@ -71,26 +77,6 @@ async def test_get_org_users_auth_root(no_su_client: AsyncClient):
assert "Must be the org's root user" in resp.json()["detail"]
@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.flush()
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"]
@pytest.mark.anyio
async def test_get_org_groups_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/org/groups?org_id=2")

View file

@ -10,6 +10,12 @@ from src.organisation.models import OrgUsers
from src.user.models import User
pytestmark = [
pytest.mark.auth,
pytest.mark.super_admin,
]
@pytest.mark.anyio
async def test_get_user_auth_su(no_su_client: AsyncClient):
resp = await no_su_client.get("/user/?user_id=1")
@ -67,7 +73,7 @@ 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):
async def test_post_perm_auth_su(no_su_client: AsyncClient, db_session):
resp = await no_su_client.post(
"/iam/permission",
json={"service_id": 1, "resource": "test_resource", "action": "create"},
@ -75,3 +81,23 @@ async def test_post_perm_success(no_su_client: AsyncClient, db_session):
assert resp.status_code != 422
assert resp.status_code == 401
assert resp.json()["detail"] == "Must be super admin"
@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()
resp = await no_su_client.post(
"/org/user", json={"organisation_id": 1, "user_id": 2}
)
assert resp.status_code != 422
assert resp.status_code == 401
assert "Must be super admin" in resp.json()["detail"]

View file

@ -6,6 +6,12 @@ import pytest
from httpx import AsyncClient
pytestmark = [
pytest.mark.auth,
pytest.mark.user,
]
@pytest.mark.anyio
async def test_get_self_db_auth_user(no_user_client: AsyncClient):
resp = await no_user_client.get("/user/self/db")

View file

@ -10,6 +10,11 @@ from src.iam.models import Group
from .conftest import generate_query_and_status
pytestmark = [
pytest.mark.iam_module,
]
@pytest.mark.anyio
async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient):
body = {
@ -132,14 +137,18 @@ async def test_act_on_resource_logic(
@pytest.mark.anyio
async def test_get_group_permissions_success(default_client: AsyncClient):
resp = await default_client.get("/iam/group/permissions?org_id=1&group_id=1")
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "permissions" in data
assert isinstance(data["permissions"], list)
assert data["permissions"][0]["service_name"] == "Test Service"
assert data["permissions"][0]["resource"] == "test_resource"
assert data["permissions"][0]["action"] == "read"
permission = data["permissions"][0]
assert permission["id"] == 1
assert permission["service_name"] == "Test Service"
assert permission["resource"] == "test_resource"
assert permission["action"] == "read"
@pytest.mark.parametrize(
@ -186,15 +195,18 @@ async def test_get_group_permissions_mismatch(
@pytest.mark.anyio
async def test_get_group_users_success(default_client: AsyncClient):
resp = await default_client.get("/iam/group/users?org_id=1&group_id=1")
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "users" in data
assert isinstance(data["users"], list)
assert data["users"][0]["id"] == 1
assert data["users"][0]["first_name"] == "Admin"
assert data["users"][0]["last_name"] == "Test"
assert data["users"][0]["email"] == "admin@test.com"
user = data["users"][0]
assert user["id"] == 1
assert user["first_name"] == "Admin"
assert user["last_name"] == "Test"
assert user["email"] == "admin@test.com"
@pytest.mark.parametrize(
@ -243,9 +255,10 @@ async def test_post_group_success(default_client: AsyncClient):
resp = await default_client.post(
"/iam/group", json={"name": "New Group", "organisation_id": 1}
)
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "New Group"
@ -298,9 +311,10 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
"/iam/group/permission",
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
)
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Test Group Two"
@ -308,9 +322,12 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
assert "permissions" in data
assert isinstance(data["permissions"], list)
assert data["permissions"][0]["service_name"] == "Test Service"
assert data["permissions"][0]["resource"] == "test_resource"
assert data["permissions"][0]["action"] == "read"
permission = data["permissions"][0]
assert permission["id"] == 1
assert permission["service_name"] == "Test Service"
assert permission["resource"] == "test_resource"
assert permission["action"] == "read"
@pytest.mark.parametrize(
@ -430,9 +447,10 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
resp = await default_client.put(
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
)
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Test Group"
@ -440,10 +458,12 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
assert "users" in data
assert isinstance(data["users"], list)
assert data["users"][1]["id"] == 2
assert data["users"][1]["first_name"] == "User"
assert data["users"][1]["last_name"] == "Test"
assert data["users"][1]["email"] == "user@test.org"
user = data["users"][1]
assert user["id"] == 2
assert user["first_name"] == "User"
assert user["last_name"] == "Test"
assert user["email"] == "user@test.org"
@pytest.mark.parametrize(
@ -514,9 +534,12 @@ async def test_get_permissions_success(default_client: AsyncClient):
assert resp.status_code == 200
assert "permissions" in data
assert isinstance(data["permissions"], list)
assert data["permissions"][0]["service_name"] == "Test Service"
assert data["permissions"][0]["resource"] == "test_resource"
assert data["permissions"][0]["action"] == "read"
permission = data["permissions"][0]
assert permission["id"] == 1
assert permission["service_name"] == "Test Service"
assert permission["resource"] == "test_resource"
assert permission["action"] == "read"
@pytest.mark.parametrize(
@ -537,10 +560,14 @@ async def test_post_perm_success(default_client: AsyncClient, db_session):
"/iam/permission",
json={"service_id": 1, "resource": "test_resource", "action": "create"},
)
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "permission" in data
assert isinstance(data["permission"], dict)
assert data["permission"]["id"] == 2
assert data["permission"]["service_name"] == "Test Service"
assert data["permission"]["resource"] == "test_resource"
assert data["permission"]["action"] == "create"
@ -624,9 +651,12 @@ async def test_post_perm_search_success(default_client: AsyncClient, db_session,
assert resp.status_code == 200
assert "permissions" in data
assert isinstance(data["permissions"], list)
assert data["permissions"][0]["service_name"] == "Test Service"
assert data["permissions"][0]["resource"] == "test_resource"
assert data["permissions"][0]["action"] == "read"
permission = data["permissions"][0]
assert permission["id"] == 1
assert permission["service_name"] == "Test Service"
assert permission["resource"] == "test_resource"
assert permission["action"] == "read"
@pytest.mark.parametrize(

View file

@ -10,18 +10,26 @@ from src.user.models import User
from .conftest import generate_query_and_status
pytestmark = [
pytest.mark.org_module,
]
@pytest.mark.anyio
async def test_get_org_success(default_client: AsyncClient):
resp = await default_client.get("/org?org_id=1")
data = resp.json()
assert resp.status_code == 200
assert data["id"] == 1
assert data["name"] == "Test Org"
assert data["status"] == "approved"
assert data["root_user"] == "admin@test.com"
assert data["billing_contact"] == "billing@test.org"
assert data["owner_contact"] == "owner@test.org"
assert data["security_contact"] == "security@test.org"
assert data["status"] == "approved"
assert "intake_questionnaire" in data
assert isinstance(data["intake_questionnaire"], dict)
@pytest.mark.parametrize(
@ -86,10 +94,11 @@ async def test_patch_org_questionnaire_partial_success(
data = resp.json()
assert resp.status_code == 200
assert "intake_questionnaire" in data
assert data["name"] == "Test Org"
assert data["intake_questionnaire"]["question_one"] == "new answer one"
assert data["status"] == "partial"
assert "intake_questionnaire" in data
assert isinstance(data["intake_questionnaire"], dict)
assert data["intake_questionnaire"]["question_one"] == "new answer one"
assert data["intake_questionnaire"]["question_two"] == "answer two"
assert data["intake_questionnaire"]["question_three"] is None
@ -154,10 +163,11 @@ async def test_patch_org_questionnaire_submit_success(
data = resp.json()
assert resp.status_code == 200
assert "intake_questionnaire" in data
assert data["name"] == "Test Org"
assert data["intake_questionnaire"]["question_one"] == "new answer one"
assert data["status"] == "submitted"
assert "intake_questionnaire" in data
assert isinstance(data["intake_questionnaire"], dict)
assert data["intake_questionnaire"]["question_one"] == "new answer one"
assert data["intake_questionnaire"]["question_two"] == "answer two"
assert data["intake_questionnaire"]["question_three"] is None
@ -203,10 +213,15 @@ async def test_get_org_users_success(default_client: AsyncClient):
data = resp.json()
assert resp.status_code == 200
assert "users" in data
assert isinstance(data["users"], list)
assert len(data["users"]) == 1
assert data["users"][0] == "admin@test.com"
user = data["users"][0]
assert isinstance(user, dict)
assert user["email"] == "admin@test.com"
assert user["id"] == 1
assert "organisation" in data
assert data["organisation"]["name"] == "Test Org"
@ -240,9 +255,11 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
resp = await default_client.post(
"/org/user", json={"organisation_id": 1, "user_id": 2}
)
data = resp.json()
assert resp.status_code == 200
data = resp.json()
assert "users" in data
assert isinstance(data["users"], list)
assert "user@test.org" in data["users"]
@ -295,9 +312,10 @@ async def test_patch_org_root_user_success(default_client: AsyncClient, db_sessi
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
)
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Test Org"
assert data["root_user_email"] == "user@test.org"
@ -359,9 +377,10 @@ async def test_patch_org_root_user_non_member(default_client: AsyncClient, db_se
@pytest.mark.anyio
async def test_get_org_groups_success(default_client: AsyncClient):
resp = await default_client.get("/org/groups?org_id=1")
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "groups" in data
assert isinstance(data["groups"], list)
assert "Test Group" in data["groups"]
@ -389,6 +408,10 @@ async def test_get_org_contact_success(default_client: AsyncClient, contact_type
assert resp.status_code == 200
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
attributes = [
"email",
"first_name",
@ -459,9 +482,39 @@ async def test_patch_org_contact_success(
"/org/contact",
json={"organisation_id": 1, "contact_type": "billing", key: value},
)
assert resp.status_code == 200
data = resp.json()
assert resp.status_code == 200
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
attributes = [
"email",
"first_name",
"last_name",
"phonenumber",
"vat_number",
"address",
]
for attribute in attributes:
assert attribute in data["contact"]
address_attributes = [
"post_office_box_number",
"street_address",
"street_address_line_2",
"locality",
"address_region",
"country_code",
"postal_code",
]
for attribute in address_attributes:
assert attribute in data["contact"]["address"]
if key in data["contact"]:
assert data["contact"][key] == value
elif key in data["contact"]["address"]:

View file

@ -8,6 +8,11 @@ from httpx import AsyncClient
from .conftest import generate_query_and_status
pytestmark = [
pytest.mark.service_module,
]
@pytest.mark.anyio
async def test_get_services_success(default_client: AsyncClient):
resp = await default_client.get("/service/?org_id=1")
@ -15,6 +20,7 @@ async def test_get_services_success(default_client: AsyncClient):
assert resp.status_code == 200
assert "services" in data
assert isinstance(data["services"], list)
assert data["services"][0]["id"] == 1
assert data["services"][0]["name"] == "Test Service"
@ -38,6 +44,7 @@ async def test_post_service_success(default_client: AsyncClient):
assert resp.status_code == 200
assert "service" in data
assert isinstance(data["service"], dict)
assert data["service"]["name"] == "New Test Service"
assert data["service"]["id"] == 2
assert isinstance(data["service"]["api_key"], str)
@ -67,6 +74,7 @@ async def test_patch_service_success(default_client: AsyncClient):
assert resp.status_code == 200
assert "service" in data
assert isinstance(data["service"], dict)
assert data["service"]["name"] == "Test Service"
assert data["service"]["id"] == 1
assert isinstance(data["service"]["api_key"], str)

View file

@ -9,6 +9,11 @@ from httpx import AsyncClient
from .conftest import generate_query_and_status
pytestmark = [
pytest.mark.user_module,
]
@pytest.mark.anyio
async def test_get_self_db_success(default_client: AsyncClient):
resp = await default_client.get("/user/self/db")
@ -19,7 +24,9 @@ async def test_get_self_db_success(default_client: AsyncClient):
assert data["last_name"] == "Test"
assert data["email"] == "admin@test.com"
assert "organisations" in data
assert isinstance(data["organisations"], list)
assert "groups" in data
assert isinstance(data["groups"], dict)
@pytest.mark.anyio
@ -32,7 +39,9 @@ async def test_get_user_success(default_client: AsyncClient):
assert data["last_name"] == "Test"
assert data["email"] == "admin@test.com"
assert "organisations" in data
assert isinstance(data["organisations"], list)
assert "groups" in data
assert isinstance(data["groups"], dict)
@pytest.mark.anyio