irl/cloud-api
1
0
Fork 0
forked from sr2/cloud-api
Code Pull requests Activity

minor: ruff format

Browse source 
Tabs -> spaces
This commit is contained in:
Chris Milne 2026-06-22 15:04:11 +01:00
parent b2921b73b8
commit fab228bf8f
56 changed files with 3629 additions and 3630 deletions
Download patch file Download diff file Expand all files Collapse all files

320
test/conftest.py
View file

@ -22,15 +22,15 @@ SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture()
def db_session():
CustomBase.metadata.drop_all(bind=engine)
CustomBase.metadata.create_all(bind=engine)
db = SessionLocal()
try:
_seed(db) # extracted seeding logic into a plain function
yield db
finally:
db.rollback()
db.close()
CustomBase.metadata.drop_all(bind=engine)
CustomBase.metadata.create_all(bind=engine)
db = SessionLocal()
try:
_seed(db) # extracted seeding logic into a plain function
yield db
finally:
db.rollback()
db.close()
@pytest.fixture
@ -83,176 +83,176 @@ async def no_su_client(db_session) -> AsyncGenerator[AsyncClient, None]:
def _seed(db):
db.add(
User(
email="admin@test.com",
first_name="Admin",
last_name="Test",
oidc_id="abcd-efgh-ijkl-mnop",
)
)
db.add(
User(
email="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="Org One",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
intake_questionnaire={
"metadata": {"version": 0, "submission_date": None},
"questions": {"question_two": "answer two"},
},
)
)
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(Permission(service_id=1, resource="test_resource", action="move"))
db.add(Permission(service_id=1, resource="test_resource", action="delete"))
db.add(OrgPermissions(org_id=1, permission_id=1))
db.add(OrgPermissions(org_id=1, permission_id=2))
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)
group_model.permission_rel.append(perm_model)
user_model = db.get(User, 1)
org_model = db.get(Org, 1)
org_model.user_rel.append(user_model)
org_model.group_rel.append(group_model)
db.flush()
group_model.user_rel.append(user_model)
db.commit()
db.add(
User(
email="admin@test.com",
first_name="Admin",
last_name="Test",
oidc_id="abcd-efgh-ijkl-mnop",
)
)
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="Org One",
root_user_id=1,
billing_contact_id=1,
owner_contact_id=2,
security_contact_id=3,
status="approved",
intake_questionnaire={
"metadata": {"version": 0, "submission_date": None},
"questions": {"question_two": "answer two"},
},
)
)
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(Permission(service_id=1, resource="test_resource", action="move"))
db.add(Permission(service_id=1, resource="test_resource", action="delete"))
db.add(OrgPermissions(org_id=1, permission_id=1))
db.add(OrgPermissions(org_id=1, permission_id=2))
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)
group_model.permission_rel.append(perm_model)
user_model = db.get(User, 1)
org_model = db.get(Org, 1)
org_model.user_rel.append(user_model)
org_model.group_rel.append(group_model)
db.flush()
group_model.user_rel.append(user_model)
db.commit()
def generate_query_and_status(params) -> list[tuple[str, int]]:
possible_values = [0, -1, 42, "banana", ""]
possible_values = [0, -1, 42, "banana", ""]
defaults = [f"{param}=1" for param in params]
defaults = [f"{param}=1" for param in params]
# Missing params
query_list = [
"&".join(combo)
for r in range(len(defaults) + 1)
for combo in combinations(defaults, r)
]
# Missing params
query_list = [
"&".join(combo)
for r in range(len(defaults) + 1)
for combo in combinations(defaults, r)
]
# Complete query as default for invalid checks
default_query = query_list.pop(-1)
# Complete query as default for invalid checks
default_query = query_list.pop(-1)
# Checks for each param being invalid
for param in params:
for value in possible_values:
new_value = f"&{param}={value}"
query_list.append(default_query.replace(f"{param}=1", new_value))
# Checks for each param being invalid
for param in params:
for value in possible_values:
new_value = f"&{param}={value}"
query_list.append(default_query.replace(f"{param}=1", new_value))
query_and_status = []
query_and_status = []
# Assign expected status
for query in query_list:
# ID 42 is used to represent a non-existent entry. So it should 404.
status = 404 if "42" in query else 422
# Remove leading "&" if present
query = query if len(query) > 1 and query[0] != "&" else query[1:]
query_and_status.append((query, status))
# Assign expected status
for query in query_list:
# ID 42 is used to represent a non-existent entry. So it should 404.
status = 404 if "42" in query else 422
# Remove leading "&" if present
query = query if len(query) > 1 and query[0] != "&" else query[1:]
query_and_status.append((query, status))
return query_and_status
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, "", "a"]
possible_values_int = [0, -1, 42, "banana", ""]
possible_values_str = [0, "", "a"]
defaults = [{param: 1 for param in params.keys()}]
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)
]
# 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)
# 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)
# 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 = []
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
# 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():

182
test/test_auth_approval.py
View file

@ -8,181 +8,181 @@ import pytest
from httpx import AsyncClient
pytestmark = [
pytest.mark.auth,
pytest.mark.preapproval,
pytest.mark.auth,
pytest.mark.preapproval,
]
@pytest.mark.anyio
async def test_get_org_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.get("/org?org_id=3")
assert resp.status_code != 422
assert resp.status_code == 200
resp = await no_su_client.get("/org?org_id=3")
assert resp.status_code != 422
assert resp.status_code == 200
@pytest.mark.anyio
async def test_patch_org_questionnaire_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.patch(
"/org/questionnaire",
json={
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
"question_three": None,
},
"partial": True,
},
)
assert resp.status_code != 422
assert resp.status_code == 200
resp = await no_su_client.patch(
"/org/questionnaire",
json={
"organisation_id": 3,
"intake_questionnaire": {
"question_one": "new answer one",
"question_two": None,
"question_three": None,
},
"partial": True,
},
)
assert resp.status_code != 422
assert resp.status_code == 200
@pytest.mark.anyio
async def test_get_org_users_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.get("/org/users?org_id=3")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_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_get_org_groups_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.get("/org/groups?org_id=3")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_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(no_su_client: AsyncClient):
resp = await no_su_client.get("/org/contact?org_id=3&contact_type=billing")
assert resp.status_code != 422
assert resp.status_code == 200
resp = await no_su_client.get("/org/contact?org_id=3&contact_type=billing")
assert resp.status_code != 422
assert resp.status_code == 200
@pytest.mark.anyio
async def test_patch_org_contact_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.patch(
"/org/contact",
json={
"organisation_id": 3,
"contact_type": "billing",
"email": "user@example.com",
},
)
assert resp.status_code != 422
assert resp.status_code == 200
resp = await no_su_client.patch(
"/org/contact",
json={
"organisation_id": 3,
"contact_type": "billing",
"email": "user@example.com",
},
)
assert resp.status_code != 422
assert resp.status_code == 200
@pytest.mark.anyio
async def test_get_service_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.get("/service?org_id=3")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_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(no_su_client: AsyncClient):
resp = await no_su_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"]
resp = await no_su_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(no_su_client: AsyncClient):
resp = await no_su_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"]
resp = await no_su_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"]
@pytest.mark.anyio
async def test_post_iam_group_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.post(
"/iam/group", json={"name": "New Group", "organisation_id": 3}
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_client.post(
"/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(no_su_client: AsyncClient):
resp = await no_su_client.put(
"/iam/group/permission",
json={"permission_id": 1, "group_id": 2, "organisation_id": 3},
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_client.put(
"/iam/group/permission",
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(no_su_client: AsyncClient):
resp = await no_su_client.put(
"/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"]
resp = await no_su_client.put(
"/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"]
@pytest.mark.anyio
async def test_get_iam_permissions_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.get("/iam/permissions?org_id=3")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_client.get("/iam/permissions?org_id=3")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@pytest.mark.anyio
async def test_post_iam_permissions_search_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.post(
"/iam/permissions/search", json={"organisation_id": 3, "action": "read"}
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
resp = await no_su_client.post(
"/iam/permissions/search", json={"organisation_id": 3, "action": "read"}
)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@pytest.mark.anyio
async def test_delete_org_user_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.delete("/org/user?org_id=3&user_id=1")
resp = await no_su_client.delete("/org/user?org_id=3&user_id=1")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@pytest.mark.anyio
async def test_delete_preapproval_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.delete("/org/self?org_id=3")
resp = await no_su_client.delete("/org/self?org_id=3")
assert resp.status_code != 422
assert resp.status_code == 204
assert resp.status_code != 422
assert resp.status_code == 204
@pytest.mark.anyio
async def test_post_user_invitation_auth_approval(no_su_client: AsyncClient):
body = {"user_email": "admin@test.com", "organisation_id": 3}
resp = await no_su_client.post("/user/invitation", json=body)
body = {"user_email": "admin@test.com", "organisation_id": 3}
resp = await no_su_client.post("/user/invitation", json=body)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@pytest.mark.anyio
async def test_delete_group_permissions_auth_approval(no_su_client: AsyncClient):
resp = await no_su_client.delete("/iam/group/permission?org_id=3&group_id=1&perm_id=1")
resp = await no_su_client.delete("/iam/group/permission?org_id=3&group_id=1&perm_id=1")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@pytest.mark.anyio
async def test_delete_group_users_success(no_su_client: AsyncClient):
resp = await no_su_client.delete("/iam/group/user?org_id=3&group_id=1&user_id=1")
resp = await no_su_client.delete("/iam/group/user?org_id=3&group_id=1&user_id=1")
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
@pytest.mark.anyio
async def test_put_group_user_invitation_success(no_su_client: AsyncClient):
body = {"user_email": "admin@test.com", "organisation_id": 3, "group_id": 1}
resp = await no_su_client.put("/iam/group/user/invitation", json=body)
body = {"user_email": "admin@test.com", "organisation_id": 3, "group_id": 1}
resp = await no_su_client.put("/iam/group/user/invitation", json=body)
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]
assert resp.status_code != 422
assert "has not been approved." in resp.json()["detail"]

12
test/test_auth_general.py
View file

@ -5,14 +5,14 @@ from httpx import AsyncClient
pytestmark = [
pytest.mark.auth,
pytest.mark.auth,
]
@pytest.mark.anyio
async def test_get_org_auth_root_su(default_client: AsyncClient):
# If a super admin can access a resource when not the root user
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"] == "Org Two"
# If a super admin can access a resource when not the root user
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"] == "Org Two"

172
test/test_auth_root.py
View file

@ -7,147 +7,147 @@ import pytest
from httpx import AsyncClient
pytestmark = [
pytest.mark.auth,
pytest.mark.root_user,
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"]
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"]
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"]
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"]
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"]
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"]
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"]
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"]
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"]
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"]
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"]
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,
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"]
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"]
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"]
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"]

74
test/test_auth_su.py
View file

@ -7,69 +7,69 @@ import pytest
from httpx import AsyncClient
pytestmark = [
pytest.mark.auth,
pytest.mark.super_admin,
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"
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"
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"
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"
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"
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"
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"]
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"]

20
test/test_auth_user.py
View file

@ -7,22 +7,22 @@ from httpx import AsyncClient
pytestmark = [
pytest.mark.auth,
pytest.mark.user,
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"
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"
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"

6
test/test_healthcheck.py
View file

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

1000
test/test_iam.py
View file

File diff suppressed because it is too large Load diff

644
test/test_organisation.py
View file

@ -9,506 +9,506 @@ from .conftest import generate_query_and_status
pytestmark = [
pytest.mark.org_module,
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()
resp = await default_client.get("/org?org_id=1")
data = resp.json()
assert resp.status_code == 200
assert resp.status_code == 200
org = data["organisations"][0]
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 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"
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.anyio
async def test_get_org_status_checks(
default_client: AsyncClient, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org?{query}")
resp = await default_client.get(f"/org?{query}")
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()
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"
assert resp.status_code == 201
assert data["name"] == "New Test Org"
assert data["status"] == "partial"
@pytest.mark.parametrize(
"body, expected_status",
[
({"name": 42}, 422),
({}, 422),
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
],
"body, expected_status",
[
({"name": 42}, 422),
({}, 422),
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
],
)
@pytest.mark.anyio
async def test_post_org_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/org", json=body)
resp = await default_client.post("/org", json=body)
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()
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
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(
"body, expected_status",
[
({"organisation_id": 42}, 404),
({"organisation_id": "Org One"}, 422),
({"organisation_id": ""}, 422),
({}, 422),
(
{
"organisation_id": "1",
"intake_questionnaire": {"question_one": 42},
"partial": True,
},
422,
),
(
{"organisation_id": "1", "intake_questionnaire": {"question_one": "valid"}},
422,
),
(
{
"organisation_id": "1",
"intake_questionnaire": {"question_one": "valid"},
"partial": 42,
},
422,
),
],
"body, expected_status",
[
({"organisation_id": 42}, 404),
({"organisation_id": "Org One"}, 422),
({"organisation_id": ""}, 422),
({}, 422),
(
{
"organisation_id": "1",
"intake_questionnaire": {"question_one": 42},
"partial": True,
},
422,
),
(
{"organisation_id": "1", "intake_questionnaire": {"question_one": "valid"}},
422,
),
(
{
"organisation_id": "1",
"intake_questionnaire": {"question_one": "valid"},
"partial": 42,
},
422,
),
],
)
@pytest.mark.anyio
async def test_patch_questionnaire_partial_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/questionnaire", json=body)
resp = await default_client.patch("/org/questionnaire", json=body)
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_patch_org_questionnaire_submit_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": False,
},
)
data = resp.json()
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": False,
},
)
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Org Three"
assert data["status"] == "submitted"
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 not 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
assert resp.status_code == 200
assert data["name"] == "Org Three"
assert data["status"] == "submitted"
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 not 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(
"status", ["partial", "submitted", "remediation", "approved", "rejected", "removed"]
"status", ["partial", "submitted", "remediation", "approved", "rejected", "removed"]
)
@pytest.mark.anyio
async def test_patch_org_status_success(default_client: AsyncClient, status: str):
resp = await default_client.patch(
"/org/status", json={"organisation_id": 1, "status": status}
)
data = resp.json()
resp = await default_client.patch(
"/org/status", json={"organisation_id": 1, "status": status}
)
data = resp.json()
assert resp.status_code == 200
assert data["name"] == "Org One"
assert data["status"] == status
assert resp.status_code == 200
assert data["name"] == "Org One"
assert data["status"] == status
@pytest.mark.parametrize(
"body, expected_status",
[
({"organisation_id": 42}, 404),
({"organisation_id": "Org One"}, 422),
({"organisation_id": ""}, 422),
({}, 422),
({"organisation_id": "1", "status": True}, 422),
({"organisation_id": "1", "status": 42}, 422),
],
"body, expected_status",
[
({"organisation_id": 42}, 404),
({"organisation_id": "Org One"}, 422),
({"organisation_id": ""}, 422),
({}, 422),
({"organisation_id": "1", "status": True}, 422),
({"organisation_id": "1", "status": 42}, 422),
],
)
@pytest.mark.anyio
async def test_patch_org_status_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/status", json=body)
resp = await default_client.patch("/org/status", json=body)
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()
resp = await default_client.get("/org/users?org_id=1")
data = resp.json()
assert resp.status_code == 200
assert resp.status_code == 200
assert "users" in data
assert isinstance(data["users"], list)
assert len(data["users"]) == 2
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
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
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.anyio
async def test_get_org_users_status_checks(
default_client: AsyncClient, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org/users?{query}")
resp = await default_client.get(f"/org/users?{query}")
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})
resp = await default_client.post("/org/user", json={"organisation_id": 1, "user_id": 3})
assert resp.status_code == 200
assert resp.status_code == 200
data = resp.json()
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
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
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(
"body, expected_status",
[
({"organisation_id": 42}, 404),
({}, 422),
({"organisation_id": 1, "user_id": "id"}, 422),
({"user_id": 2}, 422),
({"organisation_id": 1, "user_id": 42}, 404),
({"organisation_id": 1, "user_id": 1}, 409),
],
"body, expected_status",
[
({"organisation_id": 42}, 404),
({}, 422),
({"organisation_id": 1, "user_id": "id"}, 422),
({"user_id": 2}, 422),
({"organisation_id": 1, "user_id": 42}, 404),
({"organisation_id": 1, "user_id": 1}, 409),
],
)
@pytest.mark.anyio
async def test_post_org_user_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/org/user", json=body)
resp = await default_client.post("/org/user", json=body)
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
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
)
assert resp.status_code == 200
data = resp.json()
data = resp.json()
assert data["name"] == "Org One"
assert data["root_user_email"] == "user@orgone.com"
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": "Org One", "user_id": 2}, 422),
({"organisation_id": "", "user_id": 2}, 422),
({}, 422),
({"user_id": 2}, 422),
({"user_id": 42}, 404),
({"organisation_id": 1, "user_id": "Test User"}, 422),
],
"body, expected_status",
[
({"organisation_id": 42, "user_id": 2}, 404),
({"organisation_id": "Org One", "user_id": 2}, 422),
({"organisation_id": "", "user_id": 2}, 422),
({}, 422),
({"user_id": 2}, 422),
({"user_id": 42}, 404),
({"organisation_id": 1, "user_id": "Test User"}, 422),
],
)
@pytest.mark.anyio
async def test_patch_root_user_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/root_user", json=body)
resp = await default_client.patch("/org/root_user", json=body)
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.anyio
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": 3}
)
data = resp.json()
resp = await default_client.patch(
"/org/root_user", json={"organisation_id": 1, "user_id": 3}
)
data = resp.json()
assert resp.status_code == 422
assert data["detail"] == "This user does not belong to your organisation."
assert resp.status_code == 422
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
resp = await default_client.get("/org/groups?org_id=1")
assert resp.status_code == 200
data = resp.json()
data = resp.json()
assert "organisation" in data
assert isinstance(data["organisation"], dict)
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
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"
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.anyio
async def test_get_org_groups_status_checks(
default_client: AsyncClient, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org/groups?{query}")
resp = await default_client.get(f"/org/groups?{query}")
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.parametrize("contact_type", ["billing", "security", "owner"])
@pytest.mark.anyio
async def test_get_org_contact_success(default_client: AsyncClient, contact_type: str):
resp = await default_client.get(f"/org/contact?org_id=1&contact_type={contact_type}")
data = resp.json()
resp = await default_client.get(f"/org/contact?org_id=1&contact_type={contact_type}")
data = resp.json()
assert resp.status_code == 200
assert resp.status_code == 200
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
attributes = [
"email",
"first_name",
"last_name",
"phonenumber",
"vat_number",
"address",
]
attributes = [
"email",
"first_name",
"last_name",
"phonenumber",
"vat_number",
"address",
]
for attribute in attributes:
assert attribute in data["contact"]
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",
]
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"]
for attribute in address_attributes:
assert attribute in data["contact"]["address"]
@pytest.mark.parametrize(
"query, expected_status",
[
("org_id=42&contact_type=billing", 404),
("org_id=banana&contact_type=billing", 422),
("", 422),
("org_id=1&contact_type=contact", 422),
("contact_type=billing", 422),
],
"query, expected_status",
[
("org_id=42&contact_type=billing", 404),
("org_id=banana&contact_type=billing", 422),
("", 422),
("org_id=1&contact_type=contact", 422),
("contact_type=billing", 422),
],
)
@pytest.mark.anyio
async def test_get_org_contact_status_checks(
default_client: AsyncClient, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/org/contact?{query}")
resp = await default_client.get(f"/org/contact?{query}")
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.parametrize(
"key, value",
[
("email", "user@example.com"),
("first_name", "John"),
("last_name", "Doe"),
("phonenumber", "+441234567890"),
("vat_number", "GB123456789"),
("post_office_box_number", "PO Box 123"),
("street_address", "123 Example Street"),
("street_address_line_2", "Suite 4B"),
("locality", "Glasgow"),
("address_region", "Glasgow City"),
("country_code", "GB"),
("postal_code", "G1 1AA"),
],
"key, value",
[
("email", "user@example.com"),
("first_name", "John"),
("last_name", "Doe"),
("phonenumber", "+441234567890"),
("vat_number", "GB123456789"),
("post_office_box_number", "PO Box 123"),
("street_address", "123 Example Street"),
("street_address_line_2", "Suite 4B"),
("locality", "Glasgow"),
("address_region", "Glasgow City"),
("country_code", "GB"),
("postal_code", "G1 1AA"),
],
)
@pytest.mark.anyio
async def test_patch_org_contact_success(default_client: AsyncClient, key: str, value: str):
resp = await default_client.patch(
"/org/contact",
json={"organisation_id": 1, "contact_type": "billing", key: value},
)
assert resp.status_code == 200
resp = await default_client.patch(
"/org/contact",
json={"organisation_id": 1, "contact_type": "billing", key: value},
)
assert resp.status_code == 200
data = resp.json()
data = resp.json()
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
assert "organisation" in data
assert data["organisation"]["id"] == 1
assert data["organisation"]["name"] == "Org One"
attributes = [
"email",
"first_name",
"last_name",
"phonenumber",
"vat_number",
"address",
]
attributes = [
"email",
"first_name",
"last_name",
"phonenumber",
"vat_number",
"address",
]
for attribute in attributes:
assert attribute in data["contact"]
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",
]
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"]
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"]:
assert data["contact"]["address"][key] == value
else:
pytest.fail(f"Invalid contact key: {key}")
if key in data["contact"]:
assert data["contact"][key] == value
elif key in data["contact"]["address"]:
assert data["contact"]["address"][key] == value
else:
pytest.fail(f"Invalid contact key: {key}")
@pytest.mark.parametrize(
"body, expected_status",
[
({"organisation_id": 42, "contact_type": "billing"}, 404),
({"organisation_id": 1, "contact_type": "security"}, 200),
({"organisation_id": 1, "contact_type": "owner"}, 200),
({"organisation_id": "Org One", "contact_type": "billing"}, 422),
({"organisation_id": "", "contact_type": "billing"}, 422),
({}, 422),
({"organisation_id": 1, "contact_type": "not_real"}, 422),
({"organisation_id": 1, "contact_type": 42}, 422),
({"organisation_id": 1, "contact_type": ""}, 422),
],
"body, expected_status",
[
({"organisation_id": 42, "contact_type": "billing"}, 404),
({"organisation_id": 1, "contact_type": "security"}, 200),
({"organisation_id": 1, "contact_type": "owner"}, 200),
({"organisation_id": "Org One", "contact_type": "billing"}, 422),
({"organisation_id": "", "contact_type": "billing"}, 422),
({}, 422),
({"organisation_id": 1, "contact_type": "not_real"}, 422),
({"organisation_id": 1, "contact_type": 42}, 422),
({"organisation_id": 1, "contact_type": ""}, 422),
],
)
@pytest.mark.anyio
async def test_patch_org_contact_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/org/contact", json=body)
resp = await default_client.patch("/org/contact", json=body)
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_delete_org_success(default_client: AsyncClient):
resp = await default_client.delete("/org?org_id=1")
resp = await default_client.delete("/org?org_id=1")
assert resp.status_code == 204
assert resp.status_code == 204
@pytest.mark.anyio
async def test_delete_org_users_success(default_client: AsyncClient):
resp = await default_client.delete("/org/user?org_id=1&user_id=2")
resp = await default_client.delete("/org/user?org_id=1&user_id=2")
assert resp.status_code == 204
assert resp.status_code == 204
@pytest.mark.anyio
async def test_delete_preapproval_org_success(default_client: AsyncClient):
resp = await default_client.delete("/org/self?org_id=3")
resp = await default_client.delete("/org/self?org_id=3")
assert resp.status_code == 204
assert resp.status_code == 204

78
test/test_service.py
View file

@ -8,90 +8,90 @@ from httpx import AsyncClient
from .conftest import generate_query_and_status, generate_body_and_status
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()
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"
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.anyio
async def test_get_services_status_checks(
default_client: AsyncClient, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/service?{query}")
resp = await default_client.get(f"/service?{query}")
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_post_service_success(default_client: AsyncClient):
resp = await default_client.post("/service", json={"name": "New Test Service"})
data = resp.json()
resp = await default_client.post("/service", json={"name": "New Test Service"})
data = resp.json()
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)
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)
@pytest.mark.parametrize("body, expected_status", generate_body_and_status({"name": "str"}))
@pytest.mark.anyio
async def test_post_service_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.post("/service", json=body)
resp = await default_client.post("/service", json=body)
assert resp.status_code == expected_status
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"})
resp = await default_client.post("/service", json={"name": "Test Service"})
assert resp.status_code == 409
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})
data = resp.json()
resp = await default_client.patch("/service/key", json={"service_id": 1})
data = resp.json()
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)
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)
@pytest.mark.parametrize(
"body, expected_status",
generate_body_and_status({"service_id": "int"}),
"body, expected_status",
generate_body_and_status({"service_id": "int"}),
)
@pytest.mark.anyio
async def test_patch_services_status_checks(
default_client: AsyncClient, body: dict[str, str], expected_status: int
default_client: AsyncClient, body: dict[str, str], expected_status: int
):
resp = await default_client.patch("/service/key", json=body)
resp = await default_client.patch("/service/key", json=body)
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.anyio
async def test_delete_service_success(default_client: AsyncClient):
resp = await default_client.delete("/service?service_id=1")
resp = await default_client.delete("/service?service_id=1")
assert resp.status_code == 204
assert resp.status_code == 204

242
test/test_user.py
View file

@ -11,191 +11,191 @@ from .conftest import generate_query_and_status
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()
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)
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()
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)
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.parametrize("query, expected_status", generate_query_and_status(["user_id"]))
async def test_get_user_status_checks(
default_client: AsyncClient, query: str, expected_status: int
default_client: AsyncClient, query: str, expected_status: int
):
resp = await default_client.get(f"/user?{query}")
resp = await default_client.get(f"/user?{query}")
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")
resp = await default_client.delete("/user?user_id=1")
assert resp.status_code == 204
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)
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 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 "invited_email" in data
assert isinstance(data["invited_email"], str)
assert data["invited_email"] == "admin@test.com"
@pytest.mark.parametrize(
"body, expected_status",
[
({"organisation_id": 42, "user_email": "admin@test.com"}, 404),
({"organisation_id": "Test Org", "user_email": "admin@test.com"}, 422),
({"organisation_id": "", "user_email": "admin@test.com"}, 422),
({}, 422),
({"user_email": 42}, 422),
({"organisation_id": 1, "user_email": "Test User"}, 422),
],
"body, expected_status",
[
({"organisation_id": 42, "user_email": "admin@test.com"}, 404),
({"organisation_id": "Test Org", "user_email": "admin@test.com"}, 422),
({"organisation_id": "", "user_email": "admin@test.com"}, 422),
({}, 422),
({"user_email": 42}, 422),
({"organisation_id": 1, "user_email": "Test User"}, 422),
],
)
@pytest.mark.anyio
async def test_post_user_invitation_status_checks(
default_client: AsyncClient, body, expected_status
default_client: AsyncClient, body, expected_status
):
resp = await default_client.post("/user/invitation", json=body)
resp = await default_client.post("/user/invitation", json=body)
assert resp.status_code == expected_status
assert resp.status_code == expected_status
@pytest.mark.parametrize(
"body, expected_status",
[
({"jwt": "invalid"}, 401),
({"jwt": ""}, 401),
({"jwt": None}, 422),
({"jwt": 42}, 422),
],
"body, expected_status",
[
({"jwt": "invalid"}, 401),
({"jwt": ""}, 401),
({"jwt": None}, 422),
({"jwt": 42}, 422),
],
)
@pytest.mark.anyio
async def test_post_user_invitation_accept_status_checks(
default_client: AsyncClient, body, expected_status
default_client: AsyncClient, body, expected_status
):
resp = await default_client.post("/user/invitation/accept", json=body)
resp = await default_client.post("/user/invitation/accept", json=body)
assert resp.status_code == expected_status
assert resp.status_code == expected_status
if resp.status_code == 401:
assert resp.json()["detail"] == "Invalid JWS"
if resp.status_code == 401:
assert resp.json()["detail"] == "Invalid JWS"
@pytest.mark.anyio
async def test_get_self_orgs_success(default_client: AsyncClient):
resp = await default_client.get("/user/self/orgs")
assert resp.status_code == 200
resp = await default_client.get("/user/self/orgs")
assert resp.status_code == 200
data = resp.json()
data = resp.json()
assert "organisations" in data
assert isinstance(data["organisations"], list)
assert len(data["organisations"]) > 0
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)
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["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["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
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"
path = "/user/self/orgs"
expected_data = {
"organisations": [
{
"organisation_id": 1,
"name": "Org One",
"status": "approved",
"root_user_email": "admin@test.com",
"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,
"question_three": None,
"question_two": "answer two",
},
"metadata": {"version": 0, "submission_date": None},
},
}
]
}
method = "GET"
path = "/user/self/orgs"
expected_data = {
"organisations": [
{
"organisation_id": 1,
"name": "Org One",
"status": "approved",
"root_user_email": "admin@test.com",
"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,
"question_three": None,
"question_two": "answer two",
},
"metadata": {"version": 0, "submission_date": None},
},
}
]
}
resp = await default_client.get(path)
resp = await default_client.get(path)
route = next(
route
for route in default_client._transport.app.routes # ty:ignore[unresolved-attribute]
if isinstance(route, APIRoute) and path in route.path and method in route.methods
)
route = next(
route
for route in default_client._transport.app.routes # ty:ignore[unresolved-attribute]
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
assert resp.status_code == route.status_code
if route.status_code == 204:
return
expected_response_schema = route.response_model
data = resp.json()
expected_response_schema = route.response_model
data = resp.json()
response_model = expected_response_schema(**data)
assert isinstance(response_model, expected_response_schema)
response_model = expected_response_schema(**data)
assert isinstance(response_model, expected_response_schema)
expected_response_model = expected_response_schema(**expected_data)
expected_response_model = expected_response_schema(**expected_data)
assert response_model == expected_response_model
assert response_model == expected_response_model