Compare commits
No commits in common. "a655eaf543d56dc13a6210e6aa360f3f2d7e8b26" and "8b89595531e5c6bc4d8b88dbb33cb50642a138fa" have entirely different histories.
a655eaf543
...
8b89595531
9 changed files with 547 additions and 268 deletions
105
test/conftest.py
105
test/conftest.py
|
|
@ -8,7 +8,7 @@ from sqlalchemy.orm import sessionmaker
|
||||||
|
|
||||||
from src.user.models import User
|
from src.user.models import User
|
||||||
from src.service.models import Service
|
from src.service.models import Service
|
||||||
from src.organisation.models import Organisation as Org, OrgUsers
|
from src.organisation.models import Organisation as Org
|
||||||
from src.contact.models import Contact
|
from src.contact.models import Contact
|
||||||
from src.iam.models import Group, Permission
|
from src.iam.models import Group, Permission
|
||||||
from src.auth.service import get_current_user, get_dev_user
|
from src.auth.service import get_current_user, get_dev_user
|
||||||
|
|
@ -91,35 +91,13 @@ def _seed(db):
|
||||||
oidc_id="abcd-efgh-ijkl-mnop",
|
oidc_id="abcd-efgh-ijkl-mnop",
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
db.add(
|
db.add(Contact(org_id=1, email="billing@test.org", phonenumber="07521539927"))
|
||||||
User(
|
db.add(Contact(org_id=1, email="owner@test.org", phonenumber="07521539927"))
|
||||||
email="user@orgone.com",
|
db.add(Contact(org_id=1, email="security@test.org", phonenumber="07521539927"))
|
||||||
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.flush()
|
||||||
db.add(
|
db.add(
|
||||||
Org(
|
Org(
|
||||||
name="Org One",
|
name="Test Org",
|
||||||
root_user_id=1,
|
root_user_id=1,
|
||||||
billing_contact_id=1,
|
billing_contact_id=1,
|
||||||
owner_contact_id=2,
|
owner_contact_id=2,
|
||||||
|
|
@ -131,41 +109,9 @@ def _seed(db):
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
db.add(
|
|
||||||
Org(
|
|
||||||
name="Org Two",
|
|
||||||
root_user_id=3,
|
|
||||||
billing_contact_id=4,
|
|
||||||
owner_contact_id=5,
|
|
||||||
security_contact_id=6,
|
|
||||||
status="approved",
|
|
||||||
intake_questionnaire={
|
|
||||||
"metadata": {"version": 0, "submission_date": None},
|
|
||||||
"questions": {"question_two": "answer two"},
|
|
||||||
},
|
|
||||||
)
|
|
||||||
)
|
|
||||||
db.add(
|
|
||||||
Org(
|
|
||||||
name="Org Three",
|
|
||||||
root_user_id=1,
|
|
||||||
billing_contact_id=7,
|
|
||||||
owner_contact_id=8,
|
|
||||||
security_contact_id=9,
|
|
||||||
status="partial",
|
|
||||||
intake_questionnaire={
|
|
||||||
"metadata": {"version": 0, "submission_date": None},
|
|
||||||
"questions": {"question_two": "answer two"},
|
|
||||||
},
|
|
||||||
)
|
|
||||||
)
|
|
||||||
db.add(OrgUsers(org_id=1, user_id=2))
|
|
||||||
db.add(Service(name="Test Service", api_key="123456789"))
|
db.add(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="read"))
|
||||||
db.add(Permission(service_id=1, resource="test_resource", action="move"))
|
db.add(Group(name="Test Group", org_id=1))
|
||||||
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()
|
db.flush()
|
||||||
group_model = db.get(Group, 1)
|
group_model = db.get(Group, 1)
|
||||||
perm_model = db.get(Permission, 1)
|
perm_model = db.get(Permission, 1)
|
||||||
|
|
@ -213,45 +159,6 @@ def generate_query_and_status(params) -> list[tuple[str, int]]:
|
||||||
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]
|
|
||||||
|
|
||||||
defaults = [{param: 1 for param in params.keys()}]
|
|
||||||
|
|
||||||
# Missing params
|
|
||||||
body_list = [
|
|
||||||
{key: ("valid string" if params[key] == "str" else 1) for key in combo}
|
|
||||||
for r in range(len(defaults[0].keys()) + 1)
|
|
||||||
for combo in combinations(defaults[0].keys(), r)
|
|
||||||
]
|
|
||||||
|
|
||||||
# Complete body as default for generating invalid checks
|
|
||||||
default_body = body_list.pop(-1)
|
|
||||||
|
|
||||||
# Generates checks for each param being invalid
|
|
||||||
for param, typ in params.items():
|
|
||||||
if typ == "int":
|
|
||||||
possible_values = possible_values_int
|
|
||||||
elif typ == "str":
|
|
||||||
possible_values = possible_values_str
|
|
||||||
else:
|
|
||||||
raise TypeError(f"Unknown type {typ}")
|
|
||||||
for value in possible_values:
|
|
||||||
new_record = default_body.copy()
|
|
||||||
new_record[param] = value
|
|
||||||
body_list.append(new_record)
|
|
||||||
|
|
||||||
body_and_status = []
|
|
||||||
|
|
||||||
# Assign expected status
|
|
||||||
for body in body_list:
|
|
||||||
# ID 42 is used to represent a non-existent entry. So it should 404.
|
|
||||||
status = 404 if 42 in body.values() else 422
|
|
||||||
body_and_status.append((body, status))
|
|
||||||
return body_and_status
|
|
||||||
|
|
||||||
|
|
||||||
def get_testable_routes():
|
def get_testable_routes():
|
||||||
routes = []
|
routes = []
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -7,15 +7,27 @@ Delete endpoints are currently skipped because the testing system cannot use bod
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
from src.organisation.models import Organisation as Org, OrgUsers
|
||||||
|
from src.user.models import User
|
||||||
|
from src.iam.models import Group
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.auth,
|
pytest.mark.auth,
|
||||||
pytest.mark.preapproval,
|
pytest.mark.preapproval,
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def set_org_partial(db_session):
|
||||||
|
org_model = db_session.get(Org, 1)
|
||||||
|
org_model.status = "partial"
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_auth_approval(default_client: AsyncClient):
|
async def test_get_org_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/org?org_id=3")
|
resp = await default_client.get("/org?org_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
|
|
@ -25,7 +37,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/questionnaire",
|
"/org/questionnaire",
|
||||||
json={
|
json={
|
||||||
"organisation_id": 3,
|
"organisation_id": 1,
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"question_one": "new answer one",
|
"question_one": "new answer one",
|
||||||
"question_two": None,
|
"question_two": None,
|
||||||
|
|
@ -41,7 +53,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_status_auth_approval(default_client: AsyncClient):
|
async def test_patch_org_status_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/status", json={"organisation_id": 3, "status": "submitted"}
|
"/org/status", json={"organisation_id": 1, "status": "submitted"}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
@ -49,24 +61,48 @@ async def test_patch_org_status_auth_approval(default_client: AsyncClient):
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_users_auth_approval(default_client: AsyncClient):
|
async def test_get_org_users_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/org/users?org_id=3")
|
resp = await default_client.get("/org/users?org_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_org_user_auth_approval(default_client: AsyncClient):
|
async def test_post_org_user_auth_approval(default_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.post(
|
resp = await default_client.post(
|
||||||
"/org/user", json={"organisation_id": 3, "user_id": 2}
|
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_root_user_auth_approval(default_client: AsyncClient):
|
async def test_patch_org_root_user_auth_approval(
|
||||||
|
default_client: AsyncClient, db_session
|
||||||
|
):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/root_user", json={"organisation_id": 3, "user_id": 2}
|
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
@ -74,14 +110,14 @@ async def test_patch_org_root_user_auth_approval(default_client: AsyncClient):
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_groups_auth_approval(default_client: AsyncClient):
|
async def test_get_org_groups_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/org/groups?org_id=3")
|
resp = await default_client.get("/org/groups?org_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_contact_auth_approval(default_client: AsyncClient):
|
async def test_get_org_contact_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/org/contact?org_id=3&contact_type=billing")
|
resp = await default_client.get("/org/contact?org_id=1&contact_type=billing")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
|
|
@ -91,7 +127,7 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/contact",
|
"/org/contact",
|
||||||
json={
|
json={
|
||||||
"organisation_id": 3,
|
"organisation_id": 1,
|
||||||
"contact_type": "billing",
|
"contact_type": "billing",
|
||||||
"email": "user@example.com",
|
"email": "user@example.com",
|
||||||
},
|
},
|
||||||
|
|
@ -102,21 +138,21 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_service_auth_approval(default_client: AsyncClient):
|
async def test_get_service_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/service?org_id=3")
|
resp = await default_client.get("/service?org_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_iam_group_permissions_auth_approval(default_client: AsyncClient):
|
async def test_get_iam_group_permissions_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/iam/group/permissions?org_id=3&group_id=1")
|
resp = await default_client.get("/iam/group/permissions?org_id=1&group_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_iam_group_users_auth_approval(default_client: AsyncClient):
|
async def test_get_iam_group_users_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/iam/group/users?org_id=3&group_id=1")
|
resp = await default_client.get("/iam/group/users?org_id=1&group_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
@ -124,26 +160,42 @@ async def test_get_iam_group_users_auth_approval(default_client: AsyncClient):
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_iam_group_auth_approval(default_client: AsyncClient):
|
async def test_post_iam_group_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.post(
|
resp = await default_client.post(
|
||||||
"/iam/group", json={"name": "New Group", "organisation_id": 3}
|
"/iam/group", json={"name": "New Group", "organisation_id": 1}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_iam_group_permission_auth_approval(default_client: AsyncClient):
|
async def test_put_iam_group_permission_auth_approval(
|
||||||
|
default_client: AsyncClient, db_session
|
||||||
|
):
|
||||||
|
db_session.add(Group(name="Test Group Two", org_id=1))
|
||||||
|
db_session.flush()
|
||||||
resp = await default_client.put(
|
resp = await default_client.put(
|
||||||
"/iam/group/permission",
|
"/iam/group/permission",
|
||||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 3},
|
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
|
async def test_put_iam_group_user_auth_approval(
|
||||||
|
default_client: AsyncClient, db_session
|
||||||
|
):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.put(
|
resp = await default_client.put(
|
||||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 3}
|
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
@ -151,7 +203,7 @@ async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
|
async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.get("/iam/permissions?org_id=3")
|
resp = await default_client.get("/iam/permissions?org_id=1")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
||||||
|
|
@ -159,7 +211,7 @@ async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_iam_permissions_search_auth_approval(default_client: AsyncClient):
|
async def test_post_iam_permissions_search_auth_approval(default_client: AsyncClient):
|
||||||
resp = await default_client.post(
|
resp = await default_client.post(
|
||||||
"/iam/permissions/search", json={"organisation_id": 3, "action": "read"}
|
"/iam/permissions/search", json={"organisation_id": 1, "action": "read"}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert "has not been approved." in resp.json()["detail"]
|
assert "has not been approved." in resp.json()["detail"]
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,9 @@
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
from src.organisation.models import Organisation as Org
|
||||||
|
from src.user.models import User
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.auth,
|
pytest.mark.auth,
|
||||||
|
|
@ -10,9 +13,34 @@ pytestmark = [
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_auth_root_su(default_client: AsyncClient):
|
async def test_get_org_auth_root_su(default_client: AsyncClient, db_session):
|
||||||
# If a super admin can access a resource when not the root user
|
# If a super admin can access a resource when not the root user
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="admin@test.org",
|
||||||
|
first_name="Admin",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-4321",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(
|
||||||
|
Org(
|
||||||
|
name="Test Org Two",
|
||||||
|
root_user_id=2,
|
||||||
|
billing_contact_id=1,
|
||||||
|
owner_contact_id=2,
|
||||||
|
security_contact_id=3,
|
||||||
|
status="approved",
|
||||||
|
intake_questionnaire={
|
||||||
|
"metadata": {"version": 0, "submission_date": None},
|
||||||
|
"questions": {},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.get("/org?org_id=2")
|
resp = await default_client.get("/org?org_id=2")
|
||||||
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"] == "Test Org Two"
|
||||||
|
|
|
||||||
|
|
@ -6,12 +6,42 @@ DELETE endpoints are not tested
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
from src.organisation.models import Organisation as Org
|
||||||
|
from src.user.models import User
|
||||||
|
from src.iam.models import Group
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.auth,
|
pytest.mark.auth,
|
||||||
pytest.mark.root_user,
|
pytest.mark.root_user,
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def add_second_org(db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="admin@test.org",
|
||||||
|
first_name="Admin",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-4321",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(
|
||||||
|
Org(
|
||||||
|
name="Test Org Two",
|
||||||
|
root_user_id=2,
|
||||||
|
billing_contact_id=1,
|
||||||
|
owner_contact_id=2,
|
||||||
|
security_contact_id=3,
|
||||||
|
status="approved",
|
||||||
|
intake_questionnaire={},
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_auth_root(no_su_client: AsyncClient):
|
async def test_get_org_auth_root(no_su_client: AsyncClient):
|
||||||
resp = await no_su_client.get("/org?org_id=2")
|
resp = await no_su_client.get("/org?org_id=2")
|
||||||
|
|
@ -113,7 +143,11 @@ async def test_post_iam_group_auth_root(no_su_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient):
|
async def test_put_iam_group_permission_auth_root(
|
||||||
|
no_su_client: AsyncClient, db_session
|
||||||
|
):
|
||||||
|
db_session.add(Group(name="Test Group Two", org_id=2))
|
||||||
|
db_session.flush()
|
||||||
resp = await no_su_client.put(
|
resp = await no_su_client.put(
|
||||||
"/iam/group/permission",
|
"/iam/group/permission",
|
||||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 2},
|
json={"permission_id": 1, "group_id": 2, "organisation_id": 2},
|
||||||
|
|
@ -124,9 +158,17 @@ async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_iam_group_user_auth_root(
|
async def test_put_iam_group_user_auth_root(no_su_client: AsyncClient, db_session):
|
||||||
no_su_client: AsyncClient,
|
db_session.add(
|
||||||
):
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await no_su_client.put(
|
resp = await no_su_client.put(
|
||||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 2}
|
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 2}
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -6,6 +6,10 @@ DELETE endpoints are not tested
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
from src.organisation.models import OrgUsers
|
||||||
|
from src.user.models import User
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.auth,
|
pytest.mark.auth,
|
||||||
pytest.mark.super_admin,
|
pytest.mark.super_admin,
|
||||||
|
|
@ -31,7 +35,19 @@ async def test_patch_org_status_auth_su(no_su_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_root_user_auth_su(no_su_client: AsyncClient):
|
async def test_patch_org_root_user_auth_su(no_su_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await no_su_client.patch(
|
resp = await no_su_client.patch(
|
||||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
|
|
@ -57,7 +73,7 @@ async def test_post_service_auth_su(no_su_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_perm_auth_su(no_su_client: AsyncClient):
|
async def test_post_perm_auth_su(no_su_client: AsyncClient, db_session):
|
||||||
resp = await no_su_client.post(
|
resp = await no_su_client.post(
|
||||||
"/iam/permission",
|
"/iam/permission",
|
||||||
json={"service_id": 1, "resource": "test_resource", "action": "create"},
|
json={"service_id": 1, "resource": "test_resource", "action": "create"},
|
||||||
|
|
@ -68,7 +84,17 @@ async def test_post_perm_auth_su(no_su_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_org_user_auth_su(no_su_client: AsyncClient):
|
async def test_post_org_user_auth_su(no_su_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await no_su_client.post(
|
resp = await no_su_client.post(
|
||||||
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
|
|
|
||||||
310
test/test_iam.py
310
test/test_iam.py
|
|
@ -3,7 +3,12 @@
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
from .conftest import generate_query_and_status, generate_body_and_status
|
from src.user.models import User
|
||||||
|
from src.organisation.models import Organisation as Org, OrgUsers
|
||||||
|
from src.iam.models import Group
|
||||||
|
|
||||||
|
from .conftest import generate_query_and_status
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.iam_module,
|
pytest.mark.iam_module,
|
||||||
|
|
@ -15,7 +20,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
|
||||||
body = {
|
body = {
|
||||||
"rn": {
|
"rn": {
|
||||||
"service": "Test Service",
|
"service": "Test Service",
|
||||||
"organisation": "Org One",
|
"organisation": "Test Org",
|
||||||
"resource": "test_resource",
|
"resource": "test_resource",
|
||||||
"instance": None,
|
"instance": None,
|
||||||
},
|
},
|
||||||
|
|
@ -40,7 +45,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_act_on_resource_wrong_key(
|
async def test_act_on_resource_wrong_key(
|
||||||
default_client: AsyncClient, service: str, api_key: str
|
default_client: AsyncClient, db_session, service: str, api_key: str
|
||||||
):
|
):
|
||||||
body = {
|
body = {
|
||||||
"rn": {
|
"rn": {
|
||||||
|
|
@ -92,24 +97,19 @@ async def test_act_on_resource_missing_key(default_client: AsyncClient):
|
||||||
("Test Service", 42, "test_resource", "read", 422),
|
("Test Service", 42, "test_resource", "read", 422),
|
||||||
("Test Service", "Test Org", None, "read", 422),
|
("Test Service", "Test Org", None, "read", 422),
|
||||||
("Test Service", "Test Org", 42, "read", 422),
|
("Test Service", "Test Org", 42, "read", 422),
|
||||||
("Test Service", "Test Org", "test_resource", None, 422),
|
|
||||||
("Test Service", "Test Org", "test_resource", 42, 422),
|
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_act_on_resource_endpoint_status_checks(
|
async def test_act_on_resource_endpoint_status_checks(
|
||||||
default_client: AsyncClient, service, org, resource, action, expected_status: int
|
default_client: AsyncClient, service, org, resource, action, expected_status: int
|
||||||
):
|
):
|
||||||
body = {
|
body = {"service": service, "organisation": org, "resource": resource}
|
||||||
"rn": {"service": service, "organisation": org, "resource": resource},
|
|
||||||
"action": action,
|
|
||||||
}
|
|
||||||
headers = {
|
headers = {
|
||||||
"Authorization": "Bearer not_checked_when_auth_is_disabled",
|
"Authorization": "Bearer not_checked_when_auth_is_disabled",
|
||||||
"X-API-Key": "123456789",
|
"X-API-Key": "123456789",
|
||||||
}
|
}
|
||||||
resp = await default_client.post(
|
resp = await default_client.post(
|
||||||
"/iam/can_act_on_resource", json=body, headers=headers
|
f"/iam/can_act_on_resource?action={action}", json=body, headers=headers
|
||||||
)
|
)
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
assert resp.status_code == expected_status
|
||||||
|
|
@ -118,15 +118,16 @@ async def test_act_on_resource_endpoint_status_checks(
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"service, org, resource, action, expected_response",
|
"service, org, resource, action, expected_response",
|
||||||
[
|
[
|
||||||
("Test Service", "Org One", "test_resource", "read", True),
|
("Test Service", "Test Org", "test_resource", "read", True),
|
||||||
("Test Service", "Org One", "test_resource", "create", False),
|
("Test Service", "Test Org", "test_resource", "create", False),
|
||||||
("Test Service", "Org One", "no_access_here", "read", False),
|
("Test Service", "Test Org", "no_access_here", "read", False),
|
||||||
("Test Service", "Org Two", "test_resource", "read", False),
|
("Test Service", "Test Org Two", "test_resource", "read", False),
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_act_on_resource_logic(
|
async def test_act_on_resource_logic(
|
||||||
default_client: AsyncClient,
|
default_client: AsyncClient,
|
||||||
|
db_session,
|
||||||
service,
|
service,
|
||||||
org,
|
org,
|
||||||
resource,
|
resource,
|
||||||
|
|
@ -172,7 +173,7 @@ async def test_get_group_permissions_success(default_client: AsyncClient):
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_group_permissions_status_checks(
|
async def test_get_group_permissions_status_checks(
|
||||||
default_client: AsyncClient, query: str, expected_status: int
|
default_client: AsyncClient, db_session, query: str, expected_status: int
|
||||||
):
|
):
|
||||||
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
||||||
|
|
||||||
|
|
@ -187,7 +188,21 @@ async def test_get_group_permissions_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_group_permissions_mismatch(default_client: AsyncClient, query: str):
|
async def test_get_group_permissions_mismatch(
|
||||||
|
default_client: AsyncClient, db_session, query: str
|
||||||
|
):
|
||||||
|
db_session.add(
|
||||||
|
Org(
|
||||||
|
name="Another Test Org",
|
||||||
|
root_user_id=1,
|
||||||
|
billing_contact_id=1,
|
||||||
|
owner_contact_id=2,
|
||||||
|
security_contact_id=3,
|
||||||
|
status="approved",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.add(Group(name="Another Test Group", org_id=2))
|
||||||
|
db_session.flush()
|
||||||
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
||||||
|
|
||||||
assert resp.status_code == 403
|
assert resp.status_code == 403
|
||||||
|
|
@ -211,12 +226,12 @@ async def test_get_group_users_success(default_client: AsyncClient):
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert isinstance(data["group"], dict)
|
assert isinstance(data["group"], dict)
|
||||||
assert data["group"]["id"] == 1
|
assert data["group"]["id"] == 1
|
||||||
assert data["group"]["name"] == "Org One Group"
|
assert data["group"]["name"] == "Test Group"
|
||||||
|
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert isinstance(data["organisation"], dict)
|
assert isinstance(data["organisation"], dict)
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -239,7 +254,21 @@ async def test_get_group_users_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_group_users_mismatch(default_client: AsyncClient, query: str):
|
async def test_get_group_users_mismatch(
|
||||||
|
default_client: AsyncClient, db_session, query: str
|
||||||
|
):
|
||||||
|
db_session.add(
|
||||||
|
Org(
|
||||||
|
name="Another Test Org",
|
||||||
|
root_user_id=1,
|
||||||
|
billing_contact_id=1,
|
||||||
|
owner_contact_id=2,
|
||||||
|
security_contact_id=3,
|
||||||
|
status="approved",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.add(Group(name="Another Test Group", org_id=2))
|
||||||
|
db_session.flush()
|
||||||
resp = await default_client.get(f"/iam/group/users?{query}")
|
resp = await default_client.get(f"/iam/group/users?{query}")
|
||||||
|
|
||||||
assert resp.status_code == 403
|
assert resp.status_code == 403
|
||||||
|
|
@ -258,12 +287,37 @@ async def test_post_group_success(default_client: AsyncClient):
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert isinstance(data["group"], dict)
|
assert isinstance(data["group"], dict)
|
||||||
assert data["group"]["name"] == "New Group"
|
assert data["group"]["name"] == "New Group"
|
||||||
assert data["group"]["id"] == 4
|
assert data["group"]["id"] == 2
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
generate_body_and_status({"organisation_id": "int", "name": "str"}),
|
[
|
||||||
|
({"organisation_id": 1, "name": "Test Group"}, 409),
|
||||||
|
(
|
||||||
|
{"organisation_id": 2, "name": "new group"},
|
||||||
|
404,
|
||||||
|
), # Non-existent organisation, valid name
|
||||||
|
(
|
||||||
|
{"organisation_id": "banana", "name": "new group"},
|
||||||
|
422,
|
||||||
|
), # Invalid organisation ID, valid name
|
||||||
|
(
|
||||||
|
{"organisation_id": "", "name": "new group"},
|
||||||
|
422,
|
||||||
|
), # Blank organisation ID, valid name
|
||||||
|
(
|
||||||
|
{"organisation_id": -1, "name": "new group"},
|
||||||
|
422,
|
||||||
|
), # Negative organisation ID, valid name
|
||||||
|
({"name": 1}, 422), # Only name
|
||||||
|
({}, 422), # Blank body
|
||||||
|
({"organisation_id": "", "name": ""}, 422), # Both blank
|
||||||
|
({"organisation_id": 1, "name": 42}, 422), # Valid organisation, invalid name
|
||||||
|
({"organisation_id": 1, "name": ""}, 422), # Valid organisation, blank name
|
||||||
|
({"organisation_id": 1, "name": "hi"}, 422), # Valid organisation, small name
|
||||||
|
({"organisation_id": 1}, 422), # Only organisation ID
|
||||||
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_group_status_checks(
|
async def test_post_group_status_checks(
|
||||||
|
|
@ -275,19 +329,12 @@ async def test_post_group_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_group_conflict(default_client: AsyncClient):
|
async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
||||||
resp = await default_client.post(
|
db_session.add(Group(name="Test Group Two", org_id=1))
|
||||||
"/iam/group", json={"organisation_id": 1, "name": "Org One Group"}
|
db_session.flush()
|
||||||
)
|
|
||||||
|
|
||||||
assert resp.status_code == 409
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_put_group_perm_success(default_client: AsyncClient):
|
|
||||||
resp = await default_client.put(
|
resp = await default_client.put(
|
||||||
"/iam/group/permission",
|
"/iam/group/permission",
|
||||||
json={"permission_id": 1, "group_id": 3, "organisation_id": 1},
|
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
|
||||||
)
|
)
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
|
|
@ -295,8 +342,8 @@ async def test_put_group_perm_success(default_client: AsyncClient):
|
||||||
|
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert isinstance(data["group"], dict)
|
assert isinstance(data["group"], dict)
|
||||||
assert data["group"]["name"] == "Org One Group Two"
|
assert data["group"]["name"] == "Test Group Two"
|
||||||
assert data["group"]["id"] == 3
|
assert data["group"]["id"] == 2
|
||||||
|
|
||||||
assert "permissions" in data
|
assert "permissions" in data
|
||||||
assert isinstance(data["permissions"], list)
|
assert isinstance(data["permissions"], list)
|
||||||
|
|
@ -310,9 +357,67 @@ async def test_put_group_perm_success(default_client: AsyncClient):
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
generate_body_and_status(
|
[
|
||||||
{"organisation_id": "int", "group_id": "int", "permission_id": "int"}
|
(
|
||||||
),
|
{"organisation_id": 42, "group_id": 1, "permission_id": 1},
|
||||||
|
404,
|
||||||
|
), # Non-existent organisation
|
||||||
|
(
|
||||||
|
{"organisation_id": "banana", "group_id": 1, "permission_id": 1},
|
||||||
|
422,
|
||||||
|
), # Invalid organisation ID
|
||||||
|
(
|
||||||
|
{"organisation_id": "", "group_id": 1, "permission_id": 1},
|
||||||
|
422,
|
||||||
|
), # Blank organisation ID
|
||||||
|
(
|
||||||
|
{"organisation_id": -1, "group_id": 1, "permission_id": 1},
|
||||||
|
422,
|
||||||
|
), # Negative organisation ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 42, "permission_id": 1},
|
||||||
|
404,
|
||||||
|
), # Non-existent group
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": "banana", "permission_id": 1},
|
||||||
|
422,
|
||||||
|
), # Invalid group ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": "", "permission_id": 1},
|
||||||
|
422,
|
||||||
|
), # Blank group ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": -1, "permission_id": 1},
|
||||||
|
422,
|
||||||
|
), # Negative group ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "permission_id": 42},
|
||||||
|
404,
|
||||||
|
), # Non-existent permission
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "permission_id": "banana"},
|
||||||
|
422,
|
||||||
|
), # Invalid permission ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "permission_id": ""},
|
||||||
|
422,
|
||||||
|
), # Blank permission ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "permission_id": -1},
|
||||||
|
422,
|
||||||
|
), # Negative permission ID
|
||||||
|
({}, 422), # Blank body
|
||||||
|
({"permission_id": 1}, 422), # Only permission
|
||||||
|
({"organisation_id": 1}, 422), # Only organisation
|
||||||
|
({"group_id": 1}, 422), # Only group
|
||||||
|
({"organisation_id": 1, "permission_id": 1}, 422), # Missing group
|
||||||
|
({"group_id": 1, "permission_id": 1}, 422), # Missing organisation
|
||||||
|
({"organisation_id": 1, "group_id": 1}, 422), # Missing permission
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "permission_id": 1},
|
||||||
|
409,
|
||||||
|
), # Permission already in group
|
||||||
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_perm_status_checks(
|
async def test_put_group_perm_status_checks(
|
||||||
|
|
@ -323,16 +428,6 @@ async def test_put_group_perm_status_checks(
|
||||||
assert resp.status_code == expected_status
|
assert resp.status_code == expected_status
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_put_group_perm_conflict(default_client: AsyncClient):
|
|
||||||
resp = await default_client.put(
|
|
||||||
"/iam/group/permission",
|
|
||||||
json={"organisation_id": 1, "group_id": 1, "permission_id": 1},
|
|
||||||
)
|
|
||||||
|
|
||||||
assert resp.status_code == 409
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body",
|
"body",
|
||||||
[
|
[
|
||||||
|
|
@ -341,7 +436,21 @@ async def test_put_group_perm_conflict(default_client: AsyncClient):
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
|
async def test_put_group_perm_mismatch(
|
||||||
|
default_client: AsyncClient, db_session, body: dict
|
||||||
|
):
|
||||||
|
db_session.add(
|
||||||
|
Org(
|
||||||
|
name="Another Test Org",
|
||||||
|
root_user_id=1,
|
||||||
|
billing_contact_id=1,
|
||||||
|
owner_contact_id=2,
|
||||||
|
security_contact_id=3,
|
||||||
|
status="approved",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.add(Group(name="Another Test Group", org_id=2))
|
||||||
|
db_session.flush()
|
||||||
resp = await default_client.put("/iam/group/permission", json=body)
|
resp = await default_client.put("/iam/group/permission", json=body)
|
||||||
|
|
||||||
assert resp.status_code == 403
|
assert resp.status_code == 403
|
||||||
|
|
@ -349,7 +458,19 @@ async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_user_success(default_client: AsyncClient):
|
async def test_put_group_user_success(default_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(OrgUsers(user_id=2, org_id=1))
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.put(
|
resp = await default_client.put(
|
||||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
|
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
|
||||||
)
|
)
|
||||||
|
|
@ -359,7 +480,7 @@ async def test_put_group_user_success(default_client: AsyncClient):
|
||||||
|
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert isinstance(data["group"], dict)
|
assert isinstance(data["group"], dict)
|
||||||
assert data["group"]["name"] == "Org One Group"
|
assert data["group"]["name"] == "Test Group"
|
||||||
assert data["group"]["id"] == 1
|
assert data["group"]["id"] == 1
|
||||||
|
|
||||||
assert "users" in data
|
assert "users" in data
|
||||||
|
|
@ -369,14 +490,59 @@ async def test_put_group_user_success(default_client: AsyncClient):
|
||||||
assert user["id"] == 2
|
assert user["id"] == 2
|
||||||
assert user["first_name"] == "User"
|
assert user["first_name"] == "User"
|
||||||
assert user["last_name"] == "Test"
|
assert user["last_name"] == "Test"
|
||||||
assert user["email"] == "user@orgone.com"
|
assert user["email"] == "user@test.org"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
generate_body_and_status(
|
[
|
||||||
{"organisation_id": "int", "group_id": "int", "user_id": "int"}
|
(
|
||||||
),
|
{"organisation_id": 42, "group_id": 1, "user_id": 1},
|
||||||
|
404,
|
||||||
|
), # Non-existent organisation
|
||||||
|
(
|
||||||
|
{"organisation_id": "banana", "group_id": 1, "user_id": 1},
|
||||||
|
422,
|
||||||
|
), # Invalid organisation ID
|
||||||
|
(
|
||||||
|
{"organisation_id": "", "group_id": 1, "user_id": 1},
|
||||||
|
422,
|
||||||
|
), # Blank organisation ID
|
||||||
|
(
|
||||||
|
{"organisation_id": -1, "group_id": 1, "user_id": 1},
|
||||||
|
422,
|
||||||
|
), # Negative organisation ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 42, "user_id": 1},
|
||||||
|
404,
|
||||||
|
), # Non-existent group
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": "banana", "user_id": 1},
|
||||||
|
422,
|
||||||
|
), # Invalid group ID
|
||||||
|
({"organisation_id": 1, "group_id": "", "user_id": 1}, 422), # Blank group ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": -1, "user_id": 1},
|
||||||
|
422,
|
||||||
|
), # Negative group ID
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "user_id": 42},
|
||||||
|
404,
|
||||||
|
), # Non-existent user
|
||||||
|
(
|
||||||
|
{"organisation_id": 1, "group_id": 1, "user_id": "banana"},
|
||||||
|
422,
|
||||||
|
), # Invalid user ID
|
||||||
|
({"organisation_id": 1, "group_id": 1, "user_id": ""}, 422), # Blank user ID
|
||||||
|
({"organisation_id": 1, "group_id": 1, "user_id": -1}, 422), # Negative user ID
|
||||||
|
({}, 422), # Blank body
|
||||||
|
({"user_id": 1}, 422), # Only user
|
||||||
|
({"organisation_id": 1}, 422), # Only organisation
|
||||||
|
({"group_id": 1}, 422), # Only group
|
||||||
|
({"organisation_id": 1, "user_id": 1}, 422), # Missing group
|
||||||
|
({"group_id": 1, "user_id": 1}, 422), # Missing organisation
|
||||||
|
({"organisation_id": 1, "group_id": 1}, 422), # Missing user
|
||||||
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_user_status_checks(
|
async def test_put_group_user_status_checks(
|
||||||
|
|
@ -416,7 +582,7 @@ async def test_get_permissions_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_perm_success(default_client: AsyncClient):
|
async def test_post_perm_success(default_client: AsyncClient, db_session):
|
||||||
resp = await default_client.post(
|
resp = await default_client.post(
|
||||||
"/iam/permission",
|
"/iam/permission",
|
||||||
json={"service_id": 1, "resource": "test_resource", "action": "create"},
|
json={"service_id": 1, "resource": "test_resource", "action": "create"},
|
||||||
|
|
@ -428,7 +594,7 @@ async def test_post_perm_success(default_client: AsyncClient):
|
||||||
assert "permission" in data
|
assert "permission" in data
|
||||||
assert isinstance(data["permission"], dict)
|
assert isinstance(data["permission"], dict)
|
||||||
|
|
||||||
assert data["permission"]["id"] == 3
|
assert data["permission"]["id"] == 2
|
||||||
assert data["permission"]["service_name"] == "Test Service"
|
assert data["permission"]["service_name"] == "Test Service"
|
||||||
assert data["permission"]["resource"] == "test_resource"
|
assert data["permission"]["resource"] == "test_resource"
|
||||||
assert data["permission"]["action"] == "create"
|
assert data["permission"]["action"] == "create"
|
||||||
|
|
@ -505,7 +671,7 @@ async def test_post_perm_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_perm_search_success(default_client: AsyncClient, body):
|
async def test_post_perm_search_success(default_client: AsyncClient, db_session, body):
|
||||||
resp = await default_client.post("/iam/permissions/search", json=body)
|
resp = await default_client.post("/iam/permissions/search", json=body)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
|
|
@ -513,11 +679,7 @@ async def test_post_perm_search_success(default_client: AsyncClient, body):
|
||||||
assert "permissions" in data
|
assert "permissions" in data
|
||||||
assert isinstance(data["permissions"], list)
|
assert isinstance(data["permissions"], list)
|
||||||
|
|
||||||
permissions_filtered = [
|
permission = data["permissions"][0]
|
||||||
permission for permission in data["permissions"] if permission["id"] == 1
|
|
||||||
]
|
|
||||||
assert len(permissions_filtered) == 1
|
|
||||||
permission = permissions_filtered[0]
|
|
||||||
assert permission["id"] == 1
|
assert permission["id"] == 1
|
||||||
assert permission["service_name"] == "Test Service"
|
assert permission["service_name"] == "Test Service"
|
||||||
assert permission["resource"] == "test_resource"
|
assert permission["resource"] == "test_resource"
|
||||||
|
|
@ -633,27 +795,7 @@ async def test_delete_group_permissions_success(default_client: AsyncClient):
|
||||||
assert len(data["permissions"]) == 0
|
assert len(data["permissions"]) == 0
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert data["group"]["id"] == 1
|
assert data["group"]["id"] == 1
|
||||||
assert data["group"]["name"] == "Org One Group"
|
assert data["group"]["name"] == "Test Group"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"query, expected_status",
|
|
||||||
generate_query_and_status(["group_id", "org_id", "perm_id"]),
|
|
||||||
)
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_delete_group_permissions_status_checks(
|
|
||||||
default_client: AsyncClient, query: str, expected_status: int
|
|
||||||
):
|
|
||||||
resp = await default_client.delete(f"/iam/group/permission?{query}")
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_delete_group_permissions_not_in_group(default_client: AsyncClient):
|
|
||||||
resp = await default_client.delete(
|
|
||||||
"/iam/group/permission?org_id=1&group_id=1&perm_id=2"
|
|
||||||
)
|
|
||||||
assert resp.status_code == 422
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
|
|
@ -674,4 +816,4 @@ async def test_delete_group_users_success(default_client: AsyncClient):
|
||||||
assert len(data["users"]) == 0
|
assert len(data["users"]) == 0
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert data["group"]["id"] == 1
|
assert data["group"]["id"] == 1
|
||||||
assert data["group"]["name"] == "Org One Group"
|
assert data["group"]["name"] == "Test Group"
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,8 @@
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
from src.organisation.models import Organisation, OrgUsers
|
||||||
|
from src.user.models import User
|
||||||
from .conftest import generate_query_and_status
|
from .conftest import generate_query_and_status
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -24,15 +26,15 @@ async def test_get_org_success(default_client: AsyncClient):
|
||||||
|
|
||||||
assert isinstance(org, dict)
|
assert isinstance(org, dict)
|
||||||
assert org["organisation_id"] == 1
|
assert org["organisation_id"] == 1
|
||||||
assert org["name"] == "Org One"
|
assert org["name"] == "Test Org"
|
||||||
assert org["status"] == "approved"
|
assert org["status"] == "approved"
|
||||||
assert org["root_user_email"] == "admin@test.com"
|
assert org["root_user_email"] == "admin@test.com"
|
||||||
assert "intake_questionnaire" in org
|
assert "intake_questionnaire" in org
|
||||||
assert isinstance(org["intake_questionnaire"], dict)
|
assert isinstance(org["intake_questionnaire"], dict)
|
||||||
|
|
||||||
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
assert org["billing_contact"]["email"] == "billing@test.org"
|
||||||
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
assert org["owner_contact"]["email"] == "owner@test.org"
|
||||||
assert org["security_contact"]["email"] == "security@orgone.com"
|
assert org["security_contact"]["email"] == "security@test.org"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -60,7 +62,7 @@ async def test_post_org_success(default_client: AsyncClient):
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
[
|
||||||
({"name": "Org One"}, 409),
|
({"name": "Test Org"}, 409),
|
||||||
({"name": 42}, 422),
|
({"name": 42}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
|
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
|
||||||
|
|
@ -76,11 +78,16 @@ async def test_post_org_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient):
|
async def test_patch_org_questionnaire_partial_success(
|
||||||
|
default_client: AsyncClient, db_session
|
||||||
|
):
|
||||||
|
org_model = db_session.get(Organisation, 1)
|
||||||
|
org_model.status = "partial"
|
||||||
|
db_session.flush()
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/questionnaire",
|
"/org/questionnaire",
|
||||||
json={
|
json={
|
||||||
"organisation_id": 3,
|
"organisation_id": 1,
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"question_one": "new answer one",
|
"question_one": "new answer one",
|
||||||
"question_two": None,
|
"question_two": None,
|
||||||
|
|
@ -92,7 +99,7 @@ async def test_patch_org_questionnaire_partial_success(default_client: AsyncClie
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
assert data["name"] == "Org Three"
|
assert data["name"] == "Test Org"
|
||||||
assert data["status"] == "partial"
|
assert data["status"] == "partial"
|
||||||
assert "intake_questionnaire" in data
|
assert "intake_questionnaire" in data
|
||||||
assert isinstance(data["intake_questionnaire"], dict)
|
assert isinstance(data["intake_questionnaire"], dict)
|
||||||
|
|
@ -109,7 +116,7 @@ async def test_patch_org_questionnaire_partial_success(default_client: AsyncClie
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
[
|
||||||
({"organisation_id": 42}, 404),
|
({"organisation_id": 42}, 404),
|
||||||
({"organisation_id": "Org One"}, 422),
|
({"organisation_id": "Test Org"}, 422),
|
||||||
({"organisation_id": ""}, 422),
|
({"organisation_id": ""}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
(
|
(
|
||||||
|
|
@ -144,11 +151,16 @@ async def test_patch_questionnaire_partial_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_questionnaire_submit_success(default_client: AsyncClient):
|
async def test_patch_org_questionnaire_submit_success(
|
||||||
|
default_client: AsyncClient, db_session
|
||||||
|
):
|
||||||
|
org_model = db_session.get(Organisation, 1)
|
||||||
|
org_model.status = "partial"
|
||||||
|
db_session.flush()
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/questionnaire",
|
"/org/questionnaire",
|
||||||
json={
|
json={
|
||||||
"organisation_id": 3,
|
"organisation_id": 1,
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"question_one": "new answer one",
|
"question_one": "new answer one",
|
||||||
"question_two": None,
|
"question_two": None,
|
||||||
|
|
@ -160,7 +172,7 @@ async def test_patch_org_questionnaire_submit_success(default_client: AsyncClien
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
assert data["name"] == "Org Three"
|
assert data["name"] == "Test Org"
|
||||||
assert data["status"] == "submitted"
|
assert data["status"] == "submitted"
|
||||||
assert "intake_questionnaire" in data
|
assert "intake_questionnaire" in data
|
||||||
assert isinstance(data["intake_questionnaire"], dict)
|
assert isinstance(data["intake_questionnaire"], dict)
|
||||||
|
|
@ -184,7 +196,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
assert data["name"] == "Org One"
|
assert data["name"] == "Test Org"
|
||||||
assert data["status"] == status
|
assert data["status"] == status
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -192,7 +204,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
[
|
||||||
({"organisation_id": 42}, 404),
|
({"organisation_id": 42}, 404),
|
||||||
({"organisation_id": "Org One"}, 422),
|
({"organisation_id": "Test Org"}, 422),
|
||||||
({"organisation_id": ""}, 422),
|
({"organisation_id": ""}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
({"organisation_id": "1", "status": True}, 422),
|
({"organisation_id": "1", "status": True}, 422),
|
||||||
|
|
@ -217,7 +229,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
|
||||||
|
|
||||||
assert "users" in data
|
assert "users" in data
|
||||||
assert isinstance(data["users"], list)
|
assert isinstance(data["users"], list)
|
||||||
assert len(data["users"]) == 2
|
assert len(data["users"]) == 1
|
||||||
|
|
||||||
user = data["users"][0]
|
user = data["users"][0]
|
||||||
assert isinstance(user, dict)
|
assert isinstance(user, dict)
|
||||||
|
|
@ -225,7 +237,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
|
||||||
assert user["id"] == 1
|
assert user["id"] == 1
|
||||||
|
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -242,9 +254,19 @@ async def test_get_org_users_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_org_user_success(default_client: AsyncClient):
|
async def test_post_org_user_success(default_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.post(
|
resp = await default_client.post(
|
||||||
"/org/user", json={"organisation_id": 1, "user_id": 3}
|
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
@ -254,12 +276,12 @@ async def test_post_org_user_success(default_client: AsyncClient):
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert isinstance(data["organisation"], dict)
|
assert isinstance(data["organisation"], dict)
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
|
|
||||||
assert "users" in data
|
assert "users" in data
|
||||||
assert isinstance(data["users"], list)
|
assert isinstance(data["users"], list)
|
||||||
assert (
|
assert (
|
||||||
len([user for user in data["users"] if user["email"] == "root@orgtwo.com"]) == 1
|
len([user for user in data["users"] if user["email"] == "user@test.org"]) == 1
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -276,15 +298,37 @@ async def test_post_org_user_success(default_client: AsyncClient):
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_org_user_status_checks(
|
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, db_session
|
||||||
):
|
):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await 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
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_root_user_success(default_client: AsyncClient):
|
async def test_patch_org_root_user_success(default_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
|
|
@ -292,15 +336,15 @@ async def test_patch_org_root_user_success(default_client: AsyncClient):
|
||||||
|
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert data["name"] == "Org One"
|
assert data["name"] == "Test Org"
|
||||||
assert data["root_user_email"] == "user@orgone.com"
|
assert data["root_user_email"] == "user@test.org"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
[
|
||||||
({"organisation_id": 42, "user_id": 2}, 404),
|
({"organisation_id": 42, "user_id": 2}, 404),
|
||||||
({"organisation_id": "Org One", "user_id": 2}, 422),
|
({"organisation_id": "Test Org", "user_id": 2}, 422),
|
||||||
({"organisation_id": "", "user_id": 2}, 422),
|
({"organisation_id": "", "user_id": 2}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
({"user_id": 2}, 422),
|
({"user_id": 2}, 422),
|
||||||
|
|
@ -310,17 +354,39 @@ async def test_patch_org_root_user_success(default_client: AsyncClient):
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_root_user_status_checks(
|
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, db_session
|
||||||
):
|
):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
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
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_root_user_non_member(default_client: AsyncClient):
|
async def test_patch_org_root_user_non_member(default_client: AsyncClient, db_session):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
|
|
||||||
resp = await default_client.patch(
|
resp = await default_client.patch(
|
||||||
"/org/root_user", json={"organisation_id": 1, "user_id": 3}
|
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||||
)
|
)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
|
|
@ -338,14 +404,14 @@ async def test_get_org_groups_success(default_client: AsyncClient):
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert isinstance(data["organisation"], dict)
|
assert isinstance(data["organisation"], dict)
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
|
|
||||||
assert "groups" in data
|
assert "groups" in data
|
||||||
assert isinstance(data["groups"], list)
|
assert isinstance(data["groups"], list)
|
||||||
group = data["groups"][0]
|
group = data["groups"][0]
|
||||||
assert isinstance(group, dict)
|
assert isinstance(group, dict)
|
||||||
assert group["id"] == 1
|
assert group["id"] == 1
|
||||||
assert group["name"] == "Org One Group"
|
assert group["name"] == "Test Group"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -372,7 +438,7 @@ async def test_get_org_contact_success(default_client: AsyncClient, contact_type
|
||||||
|
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
|
|
||||||
attributes = [
|
attributes = [
|
||||||
"email",
|
"email",
|
||||||
|
|
@ -450,7 +516,7 @@ async def test_patch_org_contact_success(
|
||||||
|
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
|
|
||||||
attributes = [
|
attributes = [
|
||||||
"email",
|
"email",
|
||||||
|
|
@ -491,7 +557,7 @@ async def test_patch_org_contact_success(
|
||||||
({"organisation_id": 42, "contact_type": "billing"}, 404),
|
({"organisation_id": 42, "contact_type": "billing"}, 404),
|
||||||
({"organisation_id": 1, "contact_type": "security"}, 200),
|
({"organisation_id": 1, "contact_type": "security"}, 200),
|
||||||
({"organisation_id": 1, "contact_type": "owner"}, 200),
|
({"organisation_id": 1, "contact_type": "owner"}, 200),
|
||||||
({"organisation_id": "Org One", "contact_type": "billing"}, 422),
|
({"organisation_id": "Test Org", "contact_type": "billing"}, 422),
|
||||||
({"organisation_id": "", "contact_type": "billing"}, 422),
|
({"organisation_id": "", "contact_type": "billing"}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
({"organisation_id": 1, "contact_type": "not_real"}, 422),
|
({"organisation_id": 1, "contact_type": "not_real"}, 422),
|
||||||
|
|
@ -516,14 +582,26 @@ async def test_delete_org_success(default_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_delete_org_users_success(default_client: AsyncClient):
|
async def test_delete_org_users_success(db_session, default_client: AsyncClient):
|
||||||
|
db_session.add(
|
||||||
|
User(
|
||||||
|
email="user@test.org",
|
||||||
|
first_name="User",
|
||||||
|
last_name="Test",
|
||||||
|
oidc_id="abcd-efgh-ijkl-1234",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db_session.flush()
|
||||||
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
|
@pytest.mark.anyio
|
||||||
async def test_delete_preapproval_org_success(default_client: AsyncClient):
|
async def test_delete_preapproval_org_success(db_session, default_client: AsyncClient):
|
||||||
resp = await default_client.delete("/org/self?org_id=3")
|
org_model = db_session.get(Organisation, 1)
|
||||||
|
org_model.status = "partial"
|
||||||
|
db_session.flush()
|
||||||
|
resp = await default_client.delete("/org/self?org_id=1")
|
||||||
|
|
||||||
assert resp.status_code == 204
|
assert resp.status_code == 204
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,8 @@
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
from .conftest import generate_query_and_status, generate_body_and_status
|
from .conftest import generate_query_and_status
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.service_module,
|
pytest.mark.service_module,
|
||||||
|
|
@ -50,7 +51,12 @@ async def test_post_service_success(default_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status", generate_body_and_status({"name": "str"})
|
"body, expected_status",
|
||||||
|
[
|
||||||
|
({"name": "Test Service"}, 409),
|
||||||
|
({"name": 42}, 422),
|
||||||
|
({}, 422),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_service_status_checks(
|
async def test_post_service_status_checks(
|
||||||
|
|
@ -61,13 +67,6 @@ async def test_post_service_status_checks(
|
||||||
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"})
|
|
||||||
|
|
||||||
assert resp.status_code == 409
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_service_success(default_client: AsyncClient):
|
async def test_patch_service_success(default_client: AsyncClient):
|
||||||
resp = await default_client.patch("/service/key", json={"service_id": 1})
|
resp = await default_client.patch("/service/key", json={"service_id": 1})
|
||||||
|
|
@ -83,7 +82,12 @@ async def test_patch_service_success(default_client: AsyncClient):
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
generate_body_and_status({"service_id": "int"}),
|
[
|
||||||
|
({"service_id": 42}, 404),
|
||||||
|
({"service_id": "Test Service"}, 422),
|
||||||
|
({"service_id": ""}, 422),
|
||||||
|
({}, 422),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_services_status_checks(
|
async def test_patch_services_status_checks(
|
||||||
|
|
|
||||||
|
|
@ -74,7 +74,7 @@ async def test_post_user_invitation_success(default_client: AsyncClient):
|
||||||
assert "organisation" in data
|
assert "organisation" in data
|
||||||
assert isinstance(data["organisation"], dict)
|
assert isinstance(data["organisation"], dict)
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
assert data["organisation"]["name"] == "Org One"
|
assert data["organisation"]["name"] == "Test Org"
|
||||||
|
|
||||||
assert "invited_email" in data
|
assert "invited_email" in data
|
||||||
assert isinstance(data["invited_email"], str)
|
assert isinstance(data["invited_email"], str)
|
||||||
|
|
@ -135,22 +135,22 @@ async def test_get_self_orgs_success(default_client: AsyncClient):
|
||||||
|
|
||||||
org = data["organisations"][0]
|
org = data["organisations"][0]
|
||||||
assert org["organisation_id"] == 1
|
assert org["organisation_id"] == 1
|
||||||
assert org["name"] == "Org One"
|
assert org["name"] == "Test Org"
|
||||||
assert org["status"] == "approved"
|
assert org["status"] == "approved"
|
||||||
assert org["root_user_email"] == "admin@test.com"
|
assert org["root_user_email"] == "admin@test.com"
|
||||||
assert "intake_questionnaire" in org
|
assert "intake_questionnaire" in org
|
||||||
assert isinstance(org["intake_questionnaire"], dict)
|
assert isinstance(org["intake_questionnaire"], dict)
|
||||||
|
|
||||||
assert isinstance(org["billing_contact"], dict)
|
assert isinstance(org["billing_contact"], dict)
|
||||||
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
assert org["billing_contact"]["email"] == "billing@test.org"
|
||||||
assert org["billing_contact"]["id"] == 1
|
assert org["billing_contact"]["id"] == 1
|
||||||
|
|
||||||
assert isinstance(org["owner_contact"], dict)
|
assert isinstance(org["owner_contact"], dict)
|
||||||
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
assert org["owner_contact"]["email"] == "owner@test.org"
|
||||||
assert org["owner_contact"]["id"] == 2
|
assert org["owner_contact"]["id"] == 2
|
||||||
|
|
||||||
assert isinstance(org["security_contact"], dict)
|
assert isinstance(org["security_contact"], dict)
|
||||||
assert org["security_contact"]["email"] == "security@orgone.com"
|
assert org["security_contact"]["email"] == "security@test.org"
|
||||||
assert org["security_contact"]["id"] == 3
|
assert org["security_contact"]["id"] == 3
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -162,12 +162,12 @@ async def test_get_self_orgs_dynamic(default_client: AsyncClient):
|
||||||
"organisations": [
|
"organisations": [
|
||||||
{
|
{
|
||||||
"organisation_id": 1,
|
"organisation_id": 1,
|
||||||
"name": "Org One",
|
"name": "Test Org",
|
||||||
"status": "approved",
|
"status": "approved",
|
||||||
"root_user_email": "admin@test.com",
|
"root_user_email": "admin@test.com",
|
||||||
"owner_contact": {"email": "owner@orgone.com", "id": 2},
|
"owner_contact": {"email": "owner@test.org", "id": 2},
|
||||||
"security_contact": {"email": "security@orgone.com", "id": 3},
|
"security_contact": {"email": "security@test.org", "id": 3},
|
||||||
"billing_contact": {"email": "billing@orgone.com", "id": 1},
|
"billing_contact": {"email": "billing@test.org", "id": 1},
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"questions": {
|
"questions": {
|
||||||
"question_one": None,
|
"question_one": None,
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue