Compare commits

...

2 commits

Author SHA1 Message Date
294baadcb7 feat: more ids returned on endpoints
All checks were successful
ci / lint_and_test (push) Successful in 14s
Issue: #23
2026-06-10 13:48:59 +01:00
5a433dfe41 tests: dynamic test structure
Issue: #23
2026-06-10 12:29:15 +01:00
9 changed files with 159 additions and 42 deletions

View file

@ -21,6 +21,7 @@ from sqlalchemy.exc import IntegrityError
from src.iam.exceptions import GroupNotFoundException
from src.organisation.exceptions import OrgNotFoundException
from src.schemas import GroupSummary, OrgSummary
from src.service.exceptions import ServiceNotFoundException
from src.exceptions import ConflictException
from src.database import db_dependency
@ -127,7 +128,11 @@ async def get_group_permissions(
):
if group_model.org_id != org_model.id:
raise UnauthorizedException("Group does not belong to this organization")
return {"permissions": group_model.permission_rel}
return {
"organisation": org_model,
"group": group_model,
"permissions": group_model.permission_rel,
}
@router.get("/group/users", response_model=IAMGetGroupUsersResponse)
@ -137,7 +142,11 @@ async def get_group_users(
):
if group_model.org_id != org_model.id:
raise UnauthorizedException("Group does not belong to this organization")
return {"users": group_model.user_rel}
return {
"organisation": org_model,
"group": group_model,
"users": group_model.user_rel,
}
@router.post("/group", response_model=IAMPostGroupResponse)
@ -180,7 +189,8 @@ async def add_group_permission(
db.flush()
response = IAMPutGroupPermissionResponse(
group=GroupSchema(**group_model.__dict__),
organisation=OrgSummary(**org_model.__dict__),
group=GroupSummary(**group_model.__dict__),
permissions=group_model.permission_rel,
)
db.commit()

View file

@ -18,6 +18,9 @@ from src.schemas import (
UserIDMixin,
PermIDMixin,
GroupIDMixin,
GroupSummary,
OrgSummary,
UserSummary,
)
@ -50,11 +53,15 @@ class IAMCAoRRequest(CustomBaseModel):
class IAMGetGroupPermissionsResponse(CustomBaseModel):
organisation: OrgSummary
group: GroupSummary
permissions: list[PermissionSchema]
class IAMGetGroupUsersResponse(CustomBaseModel):
users: list[UserSchema]
organisation: OrgSummary
group: GroupSummary
users: list[UserSummary]
class IAMPostGroupRequest(OrgIDMixin):
@ -70,7 +77,8 @@ class IAMPutGroupPermissionRequest(GroupIDMixin, PermIDMixin, OrgIDMixin):
class IAMPutGroupPermissionResponse(CustomBaseModel):
group: GroupSchema
organisation: OrgSummary
group: GroupSummary
permissions: list[PermissionSchema]

View file

@ -314,7 +314,10 @@ async def add_user_to_org(
raise ConflictException(message="User already a part of this organisation")
org_model.user_rel.append(user_model)
db.flush()
response = {"users": [user.email for user in org_model.user_rel]}
response = {
"organisation": org_model,
"users": [{"id": user.id, "email": user.email} for user in org_model.user_rel],
}
db.commit()
return response
@ -437,7 +440,12 @@ async def get_org_groups(org_model: org_model_root_claim_query_dependency):
"""
Returns a list of the names of all IAM groups created by the organisation.
"""
return {"groups": [group.name for group in org_model.group_rel]}
return {
"organisation": org_model,
"groups": [
{"id": group.id, "name": group.name} for group in org_model.group_rel
],
}
@router.delete(

View file

@ -10,7 +10,14 @@ from typing import Optional
from pydantic import EmailStr, ConfigDict
from src.schemas import CustomBaseModel, OrgIDMixin, UserIDMixin
from src.schemas import (
CustomBaseModel,
OrgIDMixin,
UserIDMixin,
GroupSummary,
OrgSummary,
UserSummary,
)
from src.contact.schemas import ContactModel
from src.organisation.constants import Status, ContactType
@ -22,11 +29,6 @@ class Questionnaire(CustomBaseModel):
question_three: Optional[str] = None
class OrgSummary(CustomBaseModel):
id: int
name: str
class ContactSummary(CustomBaseModel):
id: int
email: Optional[EmailStr] = None
@ -49,6 +51,7 @@ class OrgPostOrgRequest(CustomBaseModel):
class OrgPostOrgResponse(CustomBaseModel):
id: int
name: str
status: Status
@ -59,6 +62,7 @@ class OrgPatchQuestionnaireRequest(OrgIDMixin):
class OrgPatchQuestionnaireResponse(CustomBaseModel):
id: int
name: str
intake_questionnaire: Questionnaire
status: Status
@ -69,6 +73,7 @@ class OrgPatchStatusRequest(OrgIDMixin):
class OrgPatchStatusResponse(CustomBaseModel):
id: int
name: str
status: Status
@ -95,7 +100,8 @@ class OrgPostUserRequest(OrgIDMixin, UserIDMixin):
class OrgPostUserResponse(CustomBaseModel):
users: list[str]
organisation: OrgSummary
users: list[UserSummary]
class OrgPatchRootRequest(OrgIDMixin, UserIDMixin):
@ -113,7 +119,8 @@ class OrgGetUserResponse(CustomBaseModel):
class OrgGetGroupResponse(CustomBaseModel):
groups: list[str]
organisation: OrgSummary
groups: list[GroupSummary]
class OrgGetContactResponse(CustomBaseModel):

View file

@ -40,3 +40,18 @@ class ServiceIDMixin(CustomBaseModel):
class UserIDMixin(CustomBaseModel):
user_id: int = Field(gt=0)
class OrgSummary(CustomBaseModel):
id: int
name: str
class GroupSummary(CustomBaseModel):
id: int
name: str
class UserSummary(CustomBaseModel):
id: int
email: str

View file

@ -1,7 +1,8 @@
import pytest
from typing import AsyncGenerator
from itertools import combinations
import pytest
from fastapi.routing import APIRoute
from httpx import AsyncClient, ASGITransport
from sqlalchemy.orm import sessionmaker
@ -12,7 +13,6 @@ from src.contact.models import Contact
from src.iam.models import Group, Permission
from src.auth.service import get_current_user, get_dev_user
from src.auth.dependencies import empty_su_list, get_super_admin_list, testing_su_list
from src.main import app # inited FastAPI app
from src.database import engine, Base, get_db
@ -156,25 +156,22 @@ 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
#
# def get_testable_routes():
# routes = []
#
# for route in app.routes:
# if not isinstance(route, APIRoute):
# continue
#
# for method in route.methods:
# if method in {"HEAD", "OPTIONS"}:
# continue
#
# routes.append((route.path, method))
#
# return routes
#
#
def get_testable_routes():
routes = []
for route in app.routes:
if not isinstance(route, APIRoute):
continue
for method in route.methods:
if method in {"HEAD", "OPTIONS"}:
continue
routes.append((method, route.path, route.status_code, route.response_model))
return routes
# with open("endpoints.txt", "w") as f:
# for ep in get_testable_routes():
# f.write(f"{ep[1]} {ep[0]}\n")
# f.write(f"[{ep[0]}]{ep[1]}({ep[2]}) -> {ep[2]}: {ep[3]}\n")

View file

@ -221,10 +221,18 @@ async def test_get_group_users_success(default_client: AsyncClient):
user = data["users"][0]
assert user["id"] == 1
assert user["first_name"] == "Admin"
assert user["last_name"] == "Test"
assert user["email"] == "admin@test.com"
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["id"] == 1
assert data["group"]["name"] == "Test Group"
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
@pytest.mark.parametrize(
"query, expected_status", generate_query_and_status(["group_id", "org_id"])

View file

@ -265,9 +265,16 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
assert "users" in data
assert isinstance(data["users"], list)
assert "user@test.org" in data["users"]
assert (
len([user for user in data["users"] if user["email"] == "user@test.org"]) == 1
)
@pytest.mark.parametrize(
@ -386,9 +393,17 @@ async def test_get_org_groups_success(default_client: AsyncClient):
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Test Org"
assert "groups" in data
assert isinstance(data["groups"], list)
assert "Test Group" in data["groups"]
group = data["groups"][0]
assert isinstance(group, dict)
assert group["id"] == 1
assert group["name"] == "Test Group"
@pytest.mark.parametrize(

View file

@ -5,6 +5,7 @@
import pytest
from httpx import AsyncClient
from fastapi.routing import APIRoute
from .conftest import generate_query_and_status
@ -143,3 +144,51 @@ async def test_get_self_orgs_success(default_client: AsyncClient):
assert isinstance(org["security_contact"], dict)
assert org["security_contact"]["email"] == "security@test.org"
assert org["security_contact"]["id"] == 3
@pytest.mark.anyio
async def test_get_self_orgs_dynamic(default_client: AsyncClient):
method = "GET"
path = "/user/self/orgs"
expected_data = {
"organisations": [
{
"organisation_id": 1,
"name": "Test Org",
"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},
"intake_questionnaire": {
"question_one": None,
"question_three": None,
"question_two": "answer two",
},
}
]
}
resp = await default_client.get(path)
route = next(
route
for route in default_client._transport.app.routes
if isinstance(route, APIRoute)
and path in route.path
and method in route.methods
)
assert resp.status_code == route.status_code
if route.status_code == 204:
return
expected_response_schema = route.response_model
data = resp.json()
response_model = expected_response_schema(**data)
assert isinstance(response_model, expected_response_schema)
expected_response_model = expected_response_schema(**expected_data)
assert response_model == expected_response_model