Compare commits

..

1 commit

9 changed files with 729 additions and 771 deletions

View file

@ -1,7 +1,7 @@
from fastapi.dependencies.models import Dependant from fastapi.dependencies.models import Dependant
import pytest import pytest
from typing import AsyncGenerator, Any from typing import AsyncGenerator
from itertools import combinations from itertools import combinations
from fastapi.routing import APIRoute, iter_route_contexts from fastapi.routing import APIRoute, iter_route_contexts
from httpx import AsyncClient, ASGITransport from httpx import AsyncClient, ASGITransport
@ -256,94 +256,6 @@ def generate_body_and_status(params: dict[str, str]) -> list[tuple[dict, int]]:
return body_and_status return body_and_status
@pytest.fixture
async def route_data():
routes: dict[str, dict] = {}
contexts = list(iter_route_contexts(app.routes))
for route in contexts:
if not route.methods:
continue
if not isinstance(route.route, APIRoute):
continue
dep_func_names = set()
unchecked = [route.route.dependant]
while unchecked:
dependant = unchecked.pop(0)
ck = dependant.cache_key[0]
if hasattr(ck, "__name__"):
dep_func_names.add(ck.__name__)
unchecked += [
dep for dep in dependant.dependencies if isinstance(dep, Dependant)
]
auth_level = None
if "get_current_user" in dep_func_names:
auth_level = "User"
if (
"org_body_root_claims" in dep_func_names
or "org_query_root_claims" in dep_func_names
):
auth_level = "Root User"
if "user_model_super_admin" in dep_func_names:
auth_level = "Super Admin"
if "valid_service_key" in dep_func_names:
auth_level = "API Key"
for method in route.methods:
if method in {"HEAD", "OPTIONS"}:
continue
route_key = f"{method.upper()}{route.route.path}"
routes[route_key] = {
"status_code": route.route.status_code,
"response_model": route.route.response_model,
"auth_level": auth_level,
}
return routes
async def standard_test(
client: AsyncClient,
method: str,
path: str,
auth_level: str,
query: str,
body: dict,
expected_data: dict,
route_data: dict[str, dict[str, Any]],
):
route = route_data.get(f"{method.upper()}{path}")
assert route is not None
req_func = getattr(client, method.lower())
if method in ["GET", "DELETE"]:
resp = await req_func(url=f"{path}{query}")
else:
resp = await req_func(url=f"{path}{query}", json=body)
assert route["auth_level"] == auth_level
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
def get_testable_routes(): def get_testable_routes():
routes = [] routes = []

View file

@ -16,34 +16,3 @@ async def test_get_org_auth_root_su(default_client: AsyncClient):
assert resp.status_code != 422 assert resp.status_code != 422
assert resp.status_code == 200 assert resp.status_code == 200
assert resp.json()["organisations"][0]["name"] == "Org Two" assert resp.json()["organisations"][0]["name"] == "Org Two"
# Standardised tests verify if each endpoint has been assigned the correct auth level.
# Sample tests here verify that each auth level works.
@pytest.mark.anyio
async def test_get_org_auth_root(no_su_client: AsyncClient):
# Sample test. Checks if a non-root user gets blocked on a root endpoint.
resp = await no_su_client.get("/org?org_id=2")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_user_auth_su(no_su_client: AsyncClient):
# Sample test. Checks if a non-su user gets blocked on a su endpoint.
resp = await no_su_client.get("/user?user_id=1")
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@pytest.mark.anyio
async def test_get_self_db_auth_user(no_user_client: AsyncClient):
# Sample test. Checks if a non-user gets blocked on a user endpoint.
resp = await no_user_client.get("/user/self/db")
assert resp.status_code != 422
assert resp.status_code == 401
assert resp.json()["detail"] == "Not authenticated"

153
test/test_auth_root.py Normal file
View file

@ -0,0 +1,153 @@
"""
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
pytestmark = [
pytest.mark.auth,
pytest.mark.root_user,
]
@pytest.mark.anyio
async def test_get_org_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/org?org_id=2")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@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,
},
)
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_org_users_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/org/users?org_id=2")
assert resp.status_code != 422
assert resp.status_code == 403
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")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_org_contact_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/org/contact?org_id=2&contact_type=billing")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@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",
},
)
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_service_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/service?org_id=2")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_iam_group_permissions_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/iam/group/permissions?org_id=2&group_id=1")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_iam_group_users_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/iam/group/users?org_id=2&group_id=1")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@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}
)
assert resp.status_code != 422
assert resp.status_code == 403
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):
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 == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
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}
)
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@pytest.mark.anyio
async def test_get_iam_permissions_auth_root(no_su_client: AsyncClient):
resp = await no_su_client.get("/iam/permissions?org_id=2")
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]
@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"}
)
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be the org's root user" in resp.json()["detail"]

75
test/test_auth_su.py Normal file
View file

@ -0,0 +1,75 @@
"""
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
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")
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@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"}
)
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@pytest.mark.anyio
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}
)
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@pytest.mark.anyio
async def test_patch_service_key_auth_su(no_su_client: AsyncClient):
resp = await no_su_client.patch("/service/key", json={"service_id": 1})
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@pytest.mark.anyio
async def test_post_service_auth_su(no_su_client: AsyncClient):
resp = await no_su_client.post("/service", json={"name": "New Test Service"})
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@pytest.mark.anyio
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"},
)
assert resp.status_code != 422
assert resp.status_code == 403
assert resp.json()["detail"] == "Must be super admin"
@pytest.mark.anyio
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})
assert resp.status_code != 422
assert resp.status_code == 403
assert "Must be super admin" in resp.json()["detail"]

28
test/test_auth_user.py Normal file
View file

@ -0,0 +1,28 @@
"""
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
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")
assert resp.status_code != 422
assert resp.status_code == 401
assert resp.json()["detail"] == "Not authenticated"
@pytest.mark.anyio
async def test_post_org_success_auth_user(no_user_client: AsyncClient):
resp = await no_user_client.post("/org", json={"name": "New Test Org"})
assert resp.status_code != 422
assert resp.status_code == 401
assert resp.json()["detail"] == "Not authenticated"

View file

@ -1,13 +1,9 @@
""" """ """ """
from datetime import timedelta, datetime, timezone
from typing import Any
import pytest import pytest
from httpx import AsyncClient from httpx import AsyncClient
from src.utils import generate_jwt from .conftest import generate_query_and_status, generate_body_and_status
from .conftest import generate_query_and_status, generate_body_and_status, standard_test
pytestmark = [ pytestmark = [
pytest.mark.iam_module, pytest.mark.iam_module,
@ -148,6 +144,23 @@ async def test_act_on_resource_logic(
assert data["allowed"] == expected_response assert data["allowed"] == expected_response
@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 "permissions" in data
assert isinstance(data["permissions"], list)
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( @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"])
) )
@ -175,6 +188,31 @@ async def test_get_group_permissions_mismatch(default_client: AsyncClient, query
assert resp.json()["detail"] == "Group does not belong to this organization" assert resp.json()["detail"] == "Group does not belong to this organization"
@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 "users" in data
assert isinstance(data["users"], list)
user = data["users"][0]
assert user["id"] == 1
assert user["email"] == "admin@test.com"
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["id"] == 1
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"] == "Org One"
@pytest.mark.parametrize( @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"])
) )
@ -202,6 +240,21 @@ async def test_get_group_users_mismatch(default_client: AsyncClient, query: str)
assert resp.json()["detail"] == "Group does not belong to this organization" assert resp.json()["detail"] == "Group does not belong to this organization"
@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}
)
assert resp.status_code == 201
data = resp.json()
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "New Group"
assert data["group"]["id"] == 4
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
generate_body_and_status({"organisation_id": "int", "name": "str"}), generate_body_and_status({"organisation_id": "int", "name": "str"}),
@ -233,6 +286,31 @@ async def test_post_group_non_conflict(default_client: AsyncClient):
assert resp.status_code == 201 assert resp.status_code == 201
@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": 3, "organisation_id": 1},
)
assert resp.status_code == 200
data = resp.json()
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Org One Group Two"
assert data["group"]["id"] == 3
assert "permissions" in data
assert isinstance(data["permissions"], list)
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( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
generate_body_and_status( generate_body_and_status(
@ -273,6 +351,30 @@ async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
assert resp.json()["detail"] == "Group does not belong to this organization" assert resp.json()["detail"] == "Group does not belong to this organization"
@pytest.mark.anyio
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}
)
assert resp.status_code == 200
data = resp.json()
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Org One Group"
assert data["group"]["id"] == 1
assert "users" in data
assert isinstance(data["users"], list)
user = data["users"][1]
assert user["id"] == 2
assert user["first_name"] == "User"
assert user["last_name"] == "Test"
assert user["email"] == "user@orgone.com"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
generate_body_and_status( generate_body_and_status(
@ -288,6 +390,22 @@ async def test_put_group_user_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_get_permissions_success(default_client: AsyncClient):
resp = await default_client.get("/iam/permissions?org_id=1")
data = resp.json()
assert resp.status_code == 200
assert "permissions" in data
assert isinstance(data["permissions"], list)
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("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"]))
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_permissions_status_checks( async def test_get_permissions_status_checks(
@ -298,6 +416,25 @@ async def test_get_permissions_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
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"},
)
assert resp.status_code == 201
data = resp.json()
assert "permission" in data
assert isinstance(data["permission"], dict)
assert data["permission"]["id"] == 4
assert data["permission"]["service_name"] == "Test Service"
assert data["permission"]["resource"] == "test_resource"
assert data["permission"]["action"] == "create"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -484,6 +621,22 @@ async def test_post_perm_search_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_delete_group_permissions_success(default_client: AsyncClient):
resp = await default_client.delete(
"/iam/group/permission?org_id=1&group_id=1&perm_id=1"
)
data = resp.json()
assert resp.status_code == 200
assert "permissions" in data
assert isinstance(data["permissions"], list)
assert len(data["permissions"]) == 0
assert "group" in data
assert data["group"]["id"] == 1
assert data["group"]["name"] == "Org One Group"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"query, expected_status", "query, expected_status",
generate_query_and_status(["group_id", "org_id", "perm_id"]), generate_query_and_status(["group_id", "org_id", "perm_id"]),
@ -504,6 +657,49 @@ async def test_delete_group_permissions_not_in_group(default_client: AsyncClient
assert resp.status_code == 422 assert resp.status_code == 422
@pytest.mark.anyio
async def test_delete_permissions_success(default_client: AsyncClient):
resp = await default_client.delete("/iam/permission?perm_id=1")
assert resp.status_code == 204
@pytest.mark.anyio
async def test_delete_group_users_success(default_client: AsyncClient):
resp = await default_client.delete("/iam/group/user?org_id=1&group_id=1&user_id=1")
data = resp.json()
assert resp.status_code == 200
assert "users" in data
assert isinstance(data["users"], list)
assert len(data["users"]) == 0
assert "group" in data
assert data["group"]["id"] == 1
assert data["group"]["name"] == "Org One Group"
@pytest.mark.anyio
async def test_put_group_user_invitation_success(default_client: AsyncClient):
body = {"user_email": "admin@test.com", "organisation_id": 1, "group_id": 1}
resp = await default_client.put("/iam/group/user/invitation", json=body)
assert resp.status_code == 200
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
assert "invited_email" in data
assert isinstance(data["invited_email"], str)
assert data["invited_email"] == "admin@test.com"
assert "group" in data
assert isinstance(data["group"], dict)
assert data["group"]["name"] == "Org One Group"
assert data["group"]["id"] == 1
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -554,284 +750,3 @@ async def test_put_group_user_invitation_accept_status_checks(
if resp.status_code == 401: if resp.status_code == 401:
assert resp.json()["detail"] == "Invalid JWS" assert resp.json()["detail"] == "Invalid JWS"
@pytest.mark.anyio
async def test_get_group_permissions_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/iam/group/permissions"
auth_level = "Root User"
query = "?org_id=1&group_id=1"
body = {}
expected_data = {
"organisation": {"id": 1, "name": "Org One"},
"group": {"id": 1, "name": "Org One Group"},
"permissions": [
{
"id": 1,
"service_name": "Test Service",
"resource": "test_resource",
"action": "read",
}
],
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_get_group_users_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/iam/group/users"
auth_level = "Root User"
query = "?org_id=1&group_id=1"
body = {}
expected_data = {
"organisation": {"id": 1, "name": "Org One"},
"group": {"id": 1, "name": "Org One Group"},
"users": [{"id": 1, "email": "admin@test.com"}],
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_post_group_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "POST"
path = "/iam/group"
auth_level = "Root User"
query = ""
body = {"name": "New Group", "organisation_id": 1}
expected_data = {
"organisation": {"id": 1, "name": "Org One"},
"group": {"id": 4, "name": "New Group"},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_put_group_permission_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "PUT"
path = "/iam/group/permission"
auth_level = "Root User"
query = ""
body = {"permission_id": 1, "group_id": 3, "organisation_id": 1}
expected_data = {
"organisation": {"id": 1, "name": "Org One"},
"group": {"id": 3, "name": "Org One Group Two"},
"permissions": [
{
"id": 1,
"service_name": "Test Service",
"resource": "test_resource",
"action": "read",
}
],
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_put_group_user_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "PUT"
path = "/iam/group/user"
auth_level = "Root User"
query = ""
body = {"user_id": 2, "group_id": 1, "organisation_id": 1}
expected_data = {
"organisation": {"id": 1, "name": "Org One"},
"group": {"id": 1, "name": "Org One Group"},
"users": [
{
"id": 1,
"first_name": "Admin",
"last_name": "Test",
"email": "admin@test.com",
},
{
"id": 2,
"first_name": "User",
"last_name": "Test",
"email": "user@orgone.com",
},
],
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_get_permissions_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/iam/permissions"
auth_level = "Root User"
query = "?org_id=1"
body = {}
expected_data = {
"permissions": [
{
"id": 1,
"service_name": "Test Service",
"resource": "test_resource",
"action": "read",
},
{
"id": 2,
"service_name": "Test Service",
"resource": "test_resource",
"action": "move",
},
]
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_post_permission_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "POST"
path = "/iam/permission"
auth_level = "Super Admin"
query = ""
body = {"service_id": 1, "resource": "test_resource", "action": "create"}
expected_data = {
"permission": {
"id": 4,
"service_name": "Test Service",
"resource": "test_resource",
"action": "create",
}
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_group_permission_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/iam/group/permission"
auth_level = "Root User"
query = "?org_id=1&group_id=1&perm_id=1"
body = {}
expected_data = {"group": {"id": 1, "name": "Org One Group"}, "permissions": []}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_permission_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/iam/permission"
auth_level = "Super Admin"
query = "?perm_id=1"
body = {}
expected_data = {}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_group_user_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/iam/group/user"
auth_level = "Root User"
query = "?org_id=1&group_id=1&user_id=1"
body = {}
expected_data = {"group": {"id": 1, "name": "Org One Group"}, "users": []}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_put_group_user_invitation_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "PUT"
path = "/iam/group/user/invitation"
auth_level = "Root User"
query = ""
body = {"user_email": "admin@test.com", "organisation_id": 1, "group_id": 1}
expected_data = {
"group": {"id": 1, "name": "Org One Group"},
"organisation": {"id": 1, "name": "Org One"},
"invited_email": "admin@test.com",
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_put_group_user_invitation_accept_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
expiry_delta = timedelta(hours=24)
expiry = datetime.now(timezone.utc) + expiry_delta
claims = {
"email": "admin@test.com",
"org_id": 1,
"exp": expiry,
"type": "org_invite",
"group_id": 3,
"group_name": "Org One Group Two",
}
token = await generate_jwt(claims)
method = "PUT"
path = "/iam/group/user/invitation/accept"
auth_level = "User"
query = ""
body = {"jwt": token}
expected_data = {
"organisation": {"name": "Org One", "id": 1},
"user": {"email": "admin@test.com", "id": 1},
"group": {"details": {"id": 3, "name": "Org One Group Two"}, "permissions": []},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)

View file

@ -2,12 +2,10 @@
[DELETE] /org/ is not tested because the testing client cannot attach a body to a delete request. [DELETE] /org/ is not tested because the testing client cannot attach a body to a delete request.
""" """
from typing import Any
import pytest import pytest
from httpx import AsyncClient from httpx import AsyncClient
from .conftest import generate_query_and_status, standard_test from .conftest import generate_query_and_status
pytestmark = [ pytestmark = [
@ -15,6 +13,28 @@ pytestmark = [
] ]
@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
org = data["organisations"][0]
assert isinstance(org, dict)
assert org["organisation_id"] == 1
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@orgone.com"
assert org["owner_contact"]["email"] == "owner@orgone.com"
assert org["security_contact"]["email"] == "security@orgone.com"
@pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"]))
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_org_status_checks( async def test_get_org_status_checks(
@ -25,6 +45,16 @@ async def test_get_org_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_post_org_success(default_client: AsyncClient):
resp = await default_client.post("/org", json={"name": "New Test Org"})
data = resp.json()
assert resp.status_code == 201
assert data["name"] == "New Test Org"
assert data["status"] == "partial"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -42,6 +72,36 @@ async def test_post_org_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient):
resp = await default_client.patch(
"/org/questionnaire",
json={
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
"question_three": None,
},
"partial": True,
},
)
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Org Three"
assert data["status"] == "partial"
assert "intake_questionnaire" in data
assert isinstance(data["intake_questionnaire"], dict)
metadata = data["intake_questionnaire"]["metadata"]
assert metadata["version"] == 0
assert metadata["submission_date"] is None
questions = data["intake_questionnaire"]["questions"]
assert questions["question_one"] == "new answer one"
assert questions["question_two"] == "answer two"
assert questions["question_three"] is None
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -145,6 +205,27 @@ async def test_patch_org_status_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_get_org_users_success(default_client: AsyncClient):
resp = await default_client.get("/org/users?org_id=1")
data = resp.json()
assert resp.status_code == 200
assert "users" in data
assert isinstance(data["users"], list)
assert len(data["users"]) == 2
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"] == "Org One"
assert data["organisation"]["id"] == 1
@pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"]))
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_org_users_status_checks( async def test_get_org_users_status_checks(
@ -155,6 +236,24 @@ async def test_get_org_users_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_post_org_user_success(default_client: AsyncClient):
resp = await default_client.post("/org/user", json={"organisation_id": 1, "user_id": 3})
assert resp.status_code == 200
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
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"] == "root@orgtwo.com"]) == 1
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -175,6 +274,19 @@ async def test_post_org_user_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
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}
)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "Org One"
assert data["root_user_email"] == "user@orgone.com"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -207,6 +319,26 @@ async def test_patch_org_root_user_non_member(default_client: AsyncClient):
assert data["detail"] == "This user does not belong to your organisation." assert data["detail"] == "This user does not belong to your organisation."
@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 "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
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"] == "Org One Group"
@pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"]))
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_org_groups_status_checks( async def test_get_org_groups_status_checks(
@ -362,219 +494,21 @@ async def test_patch_org_contact_status_checks(
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_org_standard( async def test_delete_org_success(default_client: AsyncClient):
default_client: AsyncClient, route_data: dict[str, dict[str, Any]] resp = await default_client.delete("/org?org_id=1")
):
method = "GET"
path = "/org"
auth_level = "Root User"
query = "?org_id=1"
body = {}
expected_data = {
"organisations": [
{
"organisation_id": 1,
"name": "Org One",
"status": "approved",
"root_user_email": "admin@test.com",
"intake_questionnaire": {
"metadata": {"version": 0, "submission_date": None},
"questions": {
"question_one": None,
"question_two": "answer two",
"question_three": None,
},
},
"billing_contact": {"id": 1, "email": "billing@orgone.com"},
"owner_contact": {"id": 2, "email": "owner@orgone.com"},
"security_contact": {"id": 3, "email": "security@orgone.com"},
}
]
}
await standard_test( assert resp.status_code == 204
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio @pytest.mark.anyio
async def test_post_org_standard( async def test_delete_org_users_success(default_client: AsyncClient):
default_client: AsyncClient, route_data: dict[str, dict[str, Any]] resp = await default_client.delete("/org/user?org_id=1&user_id=2")
):
method = "POST"
path = "/org"
auth_level = "User"
query = ""
body = {"name": "New Test Org"}
expected_data = {"id": 4, "name": "New Test Org", "status": "partial"}
await standard_test( assert resp.status_code == 204
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio @pytest.mark.anyio
async def test_patch_org_questionnaire_partial_standard( async def test_delete_preapproval_org_success(default_client: AsyncClient):
default_client: AsyncClient, route_data: dict[str, dict[str, Any]] resp = await default_client.delete("/org/self?org_id=3")
):
method = "PATCH"
path = "/org/questionnaire"
auth_level = "Root User"
query = ""
body = {
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
"question_three": None,
},
"partial": True,
}
expected_data = {
"id": 3,
"name": "Org Three",
"status": "partial",
"intake_questionnaire": {
"metadata": {"version": 0, "submission_date": None},
"questions": {
"question_one": "new answer one",
"question_two": "answer two",
"question_three": None,
},
},
}
await standard_test( assert resp.status_code == 204
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_get_org_users_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/org/users"
auth_level = "Root User"
query = "?org_id=1"
body = {}
expected_data = {
"users": [
{"email": "admin@test.com", "id": 1},
{"email": "user@orgone.com", "id": 2},
],
"organisation": {"id": 1, "name": "Org One"},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_post_org_user_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "POST"
path = "/org/user"
auth_level = "Super Admin"
query = ""
body = {"organisation_id": 1, "user_id": 3}
expected_data = {
"users": [
{"email": "admin@test.com", "id": 1},
{"email": "user@orgone.com", "id": 2},
{"email": "root@orgtwo.com", "id": 3},
],
"organisation": {"id": 1, "name": "Org One"},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_patch_org_root_user_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "PATCH"
path = "/org/root_user"
auth_level = "Super Admin"
query = ""
body = {"organisation_id": 1, "user_id": 2}
expected_data = {"name": "Org One", "root_user_email": "user@orgone.com"}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_get_org_groups_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/org/groups"
auth_level = "Root User"
query = "?org_id=1"
body = {}
expected_data = {
"groups": [
{"name": "Org One Group", "id": 1},
{"name": "Org One Group Two", "id": 3},
],
"organisation": {"id": 1, "name": "Org One"},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_org_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/org"
auth_level = "Super Admin"
query = "?org_id=1"
body = {}
expected_data = {}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_org_users_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/org/user"
auth_level = "Root User"
query = "?org_id=1&user_id=2"
body = {}
expected_data = {}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_preapproval_org_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/org/self"
auth_level = "Root User"
query = "?org_id=3"
body = {}
expected_data = {}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)

View file

@ -2,18 +2,28 @@
409 on [POST]/service/ not tested because SQLite throws a different error than Postgres 409 on [POST]/service/ not tested because SQLite throws a different error than Postgres
""" """
from typing import Any
import pytest import pytest
from httpx import AsyncClient from httpx import AsyncClient
from .conftest import generate_query_and_status, generate_body_and_status, standard_test from .conftest import generate_query_and_status, generate_body_and_status
pytestmark = [ pytestmark = [
pytest.mark.service_module, 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")
data = resp.json()
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"
@pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"])) @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["org_id"]))
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_services_status_checks( async def test_get_services_status_checks(
@ -81,36 +91,7 @@ async def test_patch_services_status_checks(
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_services_standard( async def test_delete_service_success(default_client: AsyncClient):
default_client: AsyncClient, route_data: dict[str, dict[str, Any]] resp = await default_client.delete("/service?service_id=1")
):
method = "GET"
path = "/service"
auth_level = "Root User"
query = "?org_id=1"
body = {}
expected_data = {
"services": [
{"id": 1, "name": "Test Service"},
]
}
await standard_test( assert resp.status_code == 204
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_service_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/service"
auth_level = "Super Admin"
query = "?service_id=1"
body = {}
expected_data = {}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)

View file

@ -1,22 +1,50 @@
""" """
[GET]/user/self/claims is not tested because it requires OIDC authentication. [GET]/user/self/claims is not tested because it requires OIDC authentication.
[DELETE]/user/ is not tested because the testing client cannot attach a body to a delete request. [DELETE/user/ is not tested because the testing client cannot attach a body to a delete request.
""" """
from datetime import timedelta, datetime, timezone
from typing import Any
import pytest import pytest
from httpx import AsyncClient from httpx import AsyncClient
from fastapi.routing import APIRoute, iter_route_contexts
from .conftest import generate_query_and_status
from src.utils import generate_jwt
from .conftest import generate_query_and_status, standard_test
pytestmark = [ pytestmark = [
pytest.mark.user_module, 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")
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 isinstance(data["organisations"], list)
assert "groups" in data
assert isinstance(data["groups"], dict)
@pytest.mark.anyio
async def test_get_user_success(default_client: AsyncClient):
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 isinstance(data["organisations"], list)
assert "groups" in data
assert isinstance(data["groups"], dict)
@pytest.mark.anyio @pytest.mark.anyio
@pytest.mark.parametrize("query, expected_status", generate_query_and_status(["user_id"])) @pytest.mark.parametrize("query, expected_status", generate_query_and_status(["user_id"]))
async def test_get_user_status_checks( async def test_get_user_status_checks(
@ -27,6 +55,30 @@ async def test_get_user_status_checks(
assert resp.status_code == expected_status assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_delete_user_success(default_client: AsyncClient):
resp = await default_client.delete("/user?user_id=1")
assert resp.status_code == 204
@pytest.mark.anyio
async def test_post_user_invitation_success(default_client: AsyncClient):
body = {"user_email": "admin@test.com", "organisation_id": 1}
resp = await default_client.post("/user/invitation", json=body)
assert resp.status_code == 200
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
assert "invited_email" in data
assert isinstance(data["invited_email"], str)
assert data["invited_email"] == "admin@test.com"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"body, expected_status", "body, expected_status",
[ [
@ -69,14 +121,41 @@ async def test_post_user_invitation_accept_status_checks(
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_self_orgs_standard( async def test_get_self_orgs_success(default_client: AsyncClient):
default_client: AsyncClient, route_data: dict[str, dict[str, Any]] resp = await default_client.get("/user/self/orgs")
): assert resp.status_code == 200
data = resp.json()
assert "organisations" in data
assert isinstance(data["organisations"], list)
assert len(data["organisations"]) > 0
org = data["organisations"][0]
assert org["organisation_id"] == 1
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@orgone.com"
assert org["billing_contact"]["id"] == 1
assert isinstance(org["owner_contact"], dict)
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@orgone.com"
assert org["security_contact"]["id"] == 3
@pytest.mark.anyio
async def test_get_self_orgs_dynamic(default_client: AsyncClient):
method = "GET" method = "GET"
path = "/user/self/orgs" path = "/user/self/orgs"
auth_level = "User"
query = ""
body = {}
expected_data = { expected_data = {
"organisations": [ "organisations": [
{ {
@ -99,117 +178,29 @@ async def test_get_self_orgs_standard(
] ]
} }
await standard_test( resp = await default_client.get(path)
default_client, method, path, auth_level, query, body, expected_data, route_data
contexts = list(iter_route_contexts(default_client._transport.app.routes)) # ty:ignore[unresolved-attribute]
route = next(
route.route
for route in contexts
if isinstance(route.route, APIRoute)
and path in route.route.path
and isinstance(route.methods, set)
and method in route.methods
) )
assert resp.status_code == route.status_code
if route.status_code == 204:
return
@pytest.mark.anyio expected_response_schema = route.response_model
async def test_get_self_db_standard( data = resp.json()
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/user/self/db"
auth_level = "User"
query = ""
body = {}
expected_data = {
"id": 1,
"first_name": "Admin",
"last_name": "Test",
"email": "admin@test.com",
"organisations": [{"name": "Org One", "id": 1}],
"groups": {"Org One": [{"name": "Org One Group", "id": 1}]},
}
await standard_test( response_model = expected_response_schema(**data)
default_client, method, path, auth_level, query, body, expected_data, route_data assert isinstance(response_model, expected_response_schema)
)
expected_response_model = expected_response_schema(**expected_data)
@pytest.mark.anyio assert response_model == expected_response_model
async def test_get_user_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "GET"
path = "/user"
auth_level = "Super Admin"
query = "?user_id=1"
body = {}
expected_data = {
"id": 1,
"first_name": "Admin",
"last_name": "Test",
"email": "admin@test.com",
"organisations": [{"name": "Org One", "id": 1}],
"groups": {"Org One": [{"name": "Org One Group", "id": 1}]},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_delete_user_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "DELETE"
path = "/user"
auth_level = "Super Admin"
query = "?user_id=1"
body = {}
expected_data = {}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_post_user_invitation_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
method = "POST"
path = "/user/invitation"
auth_level = "Root User"
query = ""
body = {"user_email": "admin@test.com", "organisation_id": 1}
expected_data = {
"invited_email": "admin@test.com",
"organisation": {"name": "Org One", "id": 1},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)
@pytest.mark.anyio
async def test_post_user_invitation_accept_standard(
default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
expiry_delta = timedelta(hours=24)
expiry = datetime.now(timezone.utc) + expiry_delta
claims = {
"email": "admin@test.com",
"org_id": 2,
"exp": expiry,
"type": "org_invite",
}
token = await generate_jwt(claims)
method = "POST"
path = "/user/invitation/accept"
auth_level = "User"
query = ""
body = {"jwt": token}
expected_data = {
"organisation": {"name": "Org Two", "id": 2},
"user": {"email": "admin@test.com", "id": 1},
}
await standard_test(
default_client, method, path, auth_level, query, body, expected_data, route_data
)