Compare commits
5 commits
8b89595531
...
a655eaf543
| Author | SHA1 | Date | |
|---|---|---|---|
| a655eaf543 | |||
| fe2171df44 | |||
| c0b9763669 | |||
| fc9d7f8536 | |||
| 778f1dbece |
9 changed files with 268 additions and 547 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
|
from src.organisation.models import Organisation as Org, OrgUsers
|
||||||
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,13 +91,35 @@ def _seed(db):
|
||||||
oidc_id="abcd-efgh-ijkl-mnop",
|
oidc_id="abcd-efgh-ijkl-mnop",
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
db.add(Contact(org_id=1, email="billing@test.org", phonenumber="07521539927"))
|
db.add(
|
||||||
db.add(Contact(org_id=1, email="owner@test.org", phonenumber="07521539927"))
|
User(
|
||||||
db.add(Contact(org_id=1, email="security@test.org", phonenumber="07521539927"))
|
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.flush()
|
||||||
db.add(
|
db.add(
|
||||||
Org(
|
Org(
|
||||||
name="Test Org",
|
name="Org One",
|
||||||
root_user_id=1,
|
root_user_id=1,
|
||||||
billing_contact_id=1,
|
billing_contact_id=1,
|
||||||
owner_contact_id=2,
|
owner_contact_id=2,
|
||||||
|
|
@ -109,9 +131,41 @@ 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(Group(name="Test Group", org_id=1))
|
db.add(Permission(service_id=1, resource="test_resource", action="move"))
|
||||||
|
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)
|
||||||
|
|
@ -159,6 +213,45 @@ 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,27 +7,15 @@ 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=1")
|
resp = await default_client.get("/org?org_id=3")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
|
|
@ -37,7 +25,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": 1,
|
"organisation_id": 3,
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"question_one": "new answer one",
|
"question_one": "new answer one",
|
||||||
"question_two": None,
|
"question_two": None,
|
||||||
|
|
@ -53,7 +41,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": 1, "status": "submitted"}
|
"/org/status", json={"organisation_id": 3, "status": "submitted"}
|
||||||
)
|
)
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
@ -61,48 +49,24 @@ 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=1")
|
resp = await default_client.get("/org/users?org_id=3")
|
||||||
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, db_session):
|
async def test_post_org_user_auth_approval(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.post(
|
resp = await default_client.post(
|
||||||
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
"/org/user", json={"organisation_id": 3, "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(
|
async def test_patch_org_root_user_auth_approval(default_client: AsyncClient):
|
||||||
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": 3, "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"]
|
||||||
|
|
@ -110,14 +74,14 @@ async def test_patch_org_root_user_auth_approval(
|
||||||
|
|
||||||
@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=1")
|
resp = await default_client.get("/org/groups?org_id=3")
|
||||||
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=1&contact_type=billing")
|
resp = await default_client.get("/org/contact?org_id=3&contact_type=billing")
|
||||||
assert resp.status_code != 422
|
assert resp.status_code != 422
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
|
|
@ -127,7 +91,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": 1,
|
"organisation_id": 3,
|
||||||
"contact_type": "billing",
|
"contact_type": "billing",
|
||||||
"email": "user@example.com",
|
"email": "user@example.com",
|
||||||
},
|
},
|
||||||
|
|
@ -138,21 +102,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=1")
|
resp = await default_client.get("/service?org_id=3")
|
||||||
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=1&group_id=1")
|
resp = await default_client.get("/iam/group/permissions?org_id=3&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=1&group_id=1")
|
resp = await default_client.get("/iam/group/users?org_id=3&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"]
|
||||||
|
|
||||||
|
|
@ -160,42 +124,26 @@ 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": 1}
|
"/iam/group", json={"name": "New Group", "organisation_id": 3}
|
||||||
)
|
)
|
||||||
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(
|
async def test_put_iam_group_permission_auth_approval(default_client: AsyncClient):
|
||||||
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": 1},
|
json={"permission_id": 1, "group_id": 2, "organisation_id": 3},
|
||||||
)
|
)
|
||||||
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(
|
async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
|
||||||
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": 1}
|
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 3}
|
||||||
)
|
)
|
||||||
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"]
|
||||||
|
|
@ -203,7 +151,7 @@ async def test_put_iam_group_user_auth_approval(
|
||||||
|
|
||||||
@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=1")
|
resp = await default_client.get("/iam/permissions?org_id=3")
|
||||||
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"]
|
||||||
|
|
||||||
|
|
@ -211,7 +159,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": 1, "action": "read"}
|
"/iam/permissions/search", json={"organisation_id": 3, "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,9 +3,6 @@
|
||||||
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,
|
||||||
|
|
@ -13,34 +10,9 @@ pytestmark = [
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_org_auth_root_su(default_client: AsyncClient, db_session):
|
async def test_get_org_auth_root_su(default_client: AsyncClient):
|
||||||
# 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"] == "Test Org Two"
|
assert resp.json()["organisations"][0]["name"] == "Org Two"
|
||||||
|
|
|
||||||
|
|
@ -6,42 +6,12 @@ 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")
|
||||||
|
|
@ -143,11 +113,7 @@ 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(
|
async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient):
|
||||||
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},
|
||||||
|
|
@ -158,17 +124,9 @@ async def test_put_iam_group_permission_auth_root(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_iam_group_user_auth_root(no_su_client: AsyncClient, db_session):
|
async def test_put_iam_group_user_auth_root(
|
||||||
db_session.add(
|
no_su_client: AsyncClient,
|
||||||
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,10 +6,6 @@ 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,
|
||||||
|
|
@ -35,19 +31,7 @@ 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, db_session):
|
async def test_patch_org_root_user_auth_su(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()
|
|
||||||
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}
|
||||||
)
|
)
|
||||||
|
|
@ -73,7 +57,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, db_session):
|
async def test_post_perm_auth_su(no_su_client: AsyncClient):
|
||||||
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"},
|
||||||
|
|
@ -84,17 +68,7 @@ async def test_post_perm_auth_su(no_su_client: AsyncClient, db_session):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_org_user_auth_su(no_su_client: AsyncClient, db_session):
|
async def test_post_org_user_auth_su(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.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,12 +3,7 @@
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
from src.user.models import User
|
from .conftest import generate_query_and_status, generate_body_and_status
|
||||||
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,
|
||||||
|
|
@ -20,7 +15,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
|
||||||
body = {
|
body = {
|
||||||
"rn": {
|
"rn": {
|
||||||
"service": "Test Service",
|
"service": "Test Service",
|
||||||
"organisation": "Test Org",
|
"organisation": "Org One",
|
||||||
"resource": "test_resource",
|
"resource": "test_resource",
|
||||||
"instance": None,
|
"instance": None,
|
||||||
},
|
},
|
||||||
|
|
@ -45,7 +40,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, db_session, service: str, api_key: str
|
default_client: AsyncClient, service: str, api_key: str
|
||||||
):
|
):
|
||||||
body = {
|
body = {
|
||||||
"rn": {
|
"rn": {
|
||||||
|
|
@ -97,19 +92,24 @@ 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 = {"service": service, "organisation": org, "resource": resource}
|
body = {
|
||||||
|
"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(
|
||||||
f"/iam/can_act_on_resource?action={action}", json=body, headers=headers
|
"/iam/can_act_on_resource", json=body, headers=headers
|
||||||
)
|
)
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
assert resp.status_code == expected_status
|
||||||
|
|
@ -118,16 +118,15 @@ 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", "Test Org", "test_resource", "read", True),
|
("Test Service", "Org One", "test_resource", "read", True),
|
||||||
("Test Service", "Test Org", "test_resource", "create", False),
|
("Test Service", "Org One", "test_resource", "create", False),
|
||||||
("Test Service", "Test Org", "no_access_here", "read", False),
|
("Test Service", "Org One", "no_access_here", "read", False),
|
||||||
("Test Service", "Test Org Two", "test_resource", "read", False),
|
("Test Service", "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,
|
||||||
|
|
@ -173,7 +172,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, db_session, query: str, expected_status: int
|
default_client: AsyncClient, 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}")
|
||||||
|
|
||||||
|
|
@ -188,21 +187,7 @@ async def test_get_group_permissions_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_group_permissions_mismatch(
|
async def test_get_group_permissions_mismatch(default_client: AsyncClient, query: str):
|
||||||
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
|
||||||
|
|
@ -226,12 +211,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"] == "Test Group"
|
assert data["group"]["name"] == "Org One 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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -254,21 +239,7 @@ async def test_get_group_users_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_get_group_users_mismatch(
|
async def test_get_group_users_mismatch(default_client: AsyncClient, query: str):
|
||||||
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
|
||||||
|
|
@ -287,37 +258,12 @@ 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"] == 2
|
assert data["group"]["id"] == 4
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
generate_body_and_status({"organisation_id": "int", "name": "str"}),
|
||||||
({"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(
|
||||||
|
|
@ -329,12 +275,19 @@ async def test_post_group_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
async def test_post_group_conflict(default_client: AsyncClient):
|
||||||
db_session.add(Group(name="Test Group Two", org_id=1))
|
resp = await default_client.post(
|
||||||
db_session.flush()
|
"/iam/group", json={"organisation_id": 1, "name": "Org One Group"}
|
||||||
|
)
|
||||||
|
|
||||||
|
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": 2, "organisation_id": 1},
|
json={"permission_id": 1, "group_id": 3, "organisation_id": 1},
|
||||||
)
|
)
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
|
|
@ -342,8 +295,8 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
||||||
|
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert isinstance(data["group"], dict)
|
assert isinstance(data["group"], dict)
|
||||||
assert data["group"]["name"] == "Test Group Two"
|
assert data["group"]["name"] == "Org One Group Two"
|
||||||
assert data["group"]["id"] == 2
|
assert data["group"]["id"] == 3
|
||||||
|
|
||||||
assert "permissions" in data
|
assert "permissions" in data
|
||||||
assert isinstance(data["permissions"], list)
|
assert isinstance(data["permissions"], list)
|
||||||
|
|
@ -357,67 +310,9 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
||||||
|
|
||||||
@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(
|
||||||
|
|
@ -428,6 +323,16 @@ 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",
|
||||||
[
|
[
|
||||||
|
|
@ -436,21 +341,7 @@ async def test_put_group_perm_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_perm_mismatch(
|
async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
|
||||||
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
|
||||||
|
|
@ -458,19 +349,7 @@ async def test_put_group_perm_mismatch(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_put_group_user_success(default_client: AsyncClient, db_session):
|
async def test_put_group_user_success(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()
|
|
||||||
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}
|
||||||
)
|
)
|
||||||
|
|
@ -480,7 +359,7 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
|
||||||
|
|
||||||
assert "group" in data
|
assert "group" in data
|
||||||
assert isinstance(data["group"], dict)
|
assert isinstance(data["group"], dict)
|
||||||
assert data["group"]["name"] == "Test Group"
|
assert data["group"]["name"] == "Org One Group"
|
||||||
assert data["group"]["id"] == 1
|
assert data["group"]["id"] == 1
|
||||||
|
|
||||||
assert "users" in data
|
assert "users" in data
|
||||||
|
|
@ -490,59 +369,14 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
|
||||||
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@test.org"
|
assert user["email"] == "user@orgone.com"
|
||||||
|
|
||||||
|
|
||||||
@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(
|
||||||
|
|
@ -582,7 +416,7 @@ async def test_get_permissions_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_perm_success(default_client: AsyncClient, db_session):
|
async def test_post_perm_success(default_client: AsyncClient):
|
||||||
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"},
|
||||||
|
|
@ -594,7 +428,7 @@ async def test_post_perm_success(default_client: AsyncClient, db_session):
|
||||||
assert "permission" in data
|
assert "permission" in data
|
||||||
assert isinstance(data["permission"], dict)
|
assert isinstance(data["permission"], dict)
|
||||||
|
|
||||||
assert data["permission"]["id"] == 2
|
assert data["permission"]["id"] == 3
|
||||||
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"
|
||||||
|
|
@ -671,7 +505,7 @@ async def test_post_perm_status_checks(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_perm_search_success(default_client: AsyncClient, db_session, body):
|
async def test_post_perm_search_success(default_client: AsyncClient, 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()
|
||||||
|
|
||||||
|
|
@ -679,7 +513,11 @@ async def test_post_perm_search_success(default_client: AsyncClient, db_session,
|
||||||
assert "permissions" in data
|
assert "permissions" in data
|
||||||
assert isinstance(data["permissions"], list)
|
assert isinstance(data["permissions"], list)
|
||||||
|
|
||||||
permission = data["permissions"][0]
|
permissions_filtered = [
|
||||||
|
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"
|
||||||
|
|
@ -795,7 +633,27 @@ 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"] == "Test Group"
|
assert data["group"]["name"] == "Org One 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
|
||||||
|
|
@ -816,4 +674,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"] == "Test Group"
|
assert data["group"]["name"] == "Org One Group"
|
||||||
|
|
|
||||||
|
|
@ -5,8 +5,6 @@
|
||||||
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
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -26,15 +24,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"] == "Test Org"
|
assert org["name"] == "Org One"
|
||||||
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@test.org"
|
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
||||||
assert org["owner_contact"]["email"] == "owner@test.org"
|
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
||||||
assert org["security_contact"]["email"] == "security@test.org"
|
assert org["security_contact"]["email"] == "security@orgone.com"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -62,7 +60,7 @@ async def test_post_org_success(default_client: AsyncClient):
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
[
|
||||||
({"name": "Test Org"}, 409),
|
({"name": "Org One"}, 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),
|
||||||
|
|
@ -78,16 +76,11 @@ async def test_post_org_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_questionnaire_partial_success(
|
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient):
|
||||||
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": 1,
|
"organisation_id": 3,
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"question_one": "new answer one",
|
"question_one": "new answer one",
|
||||||
"question_two": None,
|
"question_two": None,
|
||||||
|
|
@ -99,7 +92,7 @@ async def test_patch_org_questionnaire_partial_success(
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
assert data["name"] == "Test Org"
|
assert data["name"] == "Org Three"
|
||||||
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)
|
||||||
|
|
@ -116,7 +109,7 @@ async def test_patch_org_questionnaire_partial_success(
|
||||||
"body, expected_status",
|
"body, expected_status",
|
||||||
[
|
[
|
||||||
({"organisation_id": 42}, 404),
|
({"organisation_id": 42}, 404),
|
||||||
({"organisation_id": "Test Org"}, 422),
|
({"organisation_id": "Org One"}, 422),
|
||||||
({"organisation_id": ""}, 422),
|
({"organisation_id": ""}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
(
|
(
|
||||||
|
|
@ -151,16 +144,11 @@ async def test_patch_questionnaire_partial_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_patch_org_questionnaire_submit_success(
|
async def test_patch_org_questionnaire_submit_success(default_client: AsyncClient):
|
||||||
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": 1,
|
"organisation_id": 3,
|
||||||
"intake_questionnaire": {
|
"intake_questionnaire": {
|
||||||
"question_one": "new answer one",
|
"question_one": "new answer one",
|
||||||
"question_two": None,
|
"question_two": None,
|
||||||
|
|
@ -172,7 +160,7 @@ async def test_patch_org_questionnaire_submit_success(
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
assert data["name"] == "Test Org"
|
assert data["name"] == "Org Three"
|
||||||
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)
|
||||||
|
|
@ -196,7 +184,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"] == "Test Org"
|
assert data["name"] == "Org One"
|
||||||
assert data["status"] == status
|
assert data["status"] == status
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -204,7 +192,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": "Test Org"}, 422),
|
({"organisation_id": "Org One"}, 422),
|
||||||
({"organisation_id": ""}, 422),
|
({"organisation_id": ""}, 422),
|
||||||
({}, 422),
|
({}, 422),
|
||||||
({"organisation_id": "1", "status": True}, 422),
|
({"organisation_id": "1", "status": True}, 422),
|
||||||
|
|
@ -229,7 +217,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"]) == 1
|
assert len(data["users"]) == 2
|
||||||
|
|
||||||
user = data["users"][0]
|
user = data["users"][0]
|
||||||
assert isinstance(user, dict)
|
assert isinstance(user, dict)
|
||||||
|
|
@ -237,7 +225,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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
assert data["organisation"]["id"] == 1
|
assert data["organisation"]["id"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -254,19 +242,9 @@ async def test_get_org_users_status_checks(
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_post_org_user_success(default_client: AsyncClient, db_session):
|
async def test_post_org_user_success(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.post(
|
resp = await default_client.post(
|
||||||
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
"/org/user", json={"organisation_id": 1, "user_id": 3}
|
||||||
)
|
)
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
@ -276,12 +254,12 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
|
||||||
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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
|
|
||||||
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"] == "user@test.org"]) == 1
|
len([user for user in data["users"] if user["email"] == "root@orgtwo.com"]) == 1
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -298,37 +276,15 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
|
||||||
)
|
)
|
||||||
@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, db_session
|
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
||||||
):
|
):
|
||||||
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, db_session):
|
async def test_patch_org_root_user_success(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()
|
|
||||||
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}
|
||||||
)
|
)
|
||||||
|
|
@ -336,15 +292,15 @@ async def test_patch_org_root_user_success(default_client: AsyncClient, db_sessi
|
||||||
|
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
assert data["name"] == "Test Org"
|
assert data["name"] == "Org One"
|
||||||
assert data["root_user_email"] == "user@test.org"
|
assert data["root_user_email"] == "user@orgone.com"
|
||||||
|
|
||||||
|
|
||||||
@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": "Test Org", "user_id": 2}, 422),
|
({"organisation_id": "Org One", "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),
|
||||||
|
|
@ -354,39 +310,17 @@ async def test_patch_org_root_user_success(default_client: AsyncClient, db_sessi
|
||||||
)
|
)
|
||||||
@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, db_session
|
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
||||||
):
|
):
|
||||||
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, db_session):
|
async def test_patch_org_root_user_non_member(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.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": 3}
|
||||||
)
|
)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
|
||||||
|
|
@ -404,14 +338,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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
|
|
||||||
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"] == "Test Group"
|
assert group["name"] == "Org One Group"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
@ -438,7 +372,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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
|
|
||||||
attributes = [
|
attributes = [
|
||||||
"email",
|
"email",
|
||||||
|
|
@ -516,7 +450,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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
|
|
||||||
attributes = [
|
attributes = [
|
||||||
"email",
|
"email",
|
||||||
|
|
@ -557,7 +491,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": "Test Org", "contact_type": "billing"}, 422),
|
({"organisation_id": "Org One", "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),
|
||||||
|
|
@ -582,26 +516,14 @@ async def test_delete_org_success(default_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_delete_org_users_success(db_session, default_client: AsyncClient):
|
async def test_delete_org_users_success(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(db_session, default_client: AsyncClient):
|
async def test_delete_preapproval_org_success(default_client: AsyncClient):
|
||||||
org_model = db_session.get(Organisation, 1)
|
resp = await default_client.delete("/org/self?org_id=3")
|
||||||
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,8 +5,7 @@
|
||||||
import pytest
|
import pytest
|
||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
|
|
||||||
from .conftest import generate_query_and_status
|
from .conftest import generate_query_and_status, generate_body_and_status
|
||||||
|
|
||||||
|
|
||||||
pytestmark = [
|
pytestmark = [
|
||||||
pytest.mark.service_module,
|
pytest.mark.service_module,
|
||||||
|
|
@ -51,12 +50,7 @@ async def test_post_service_success(default_client: AsyncClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
"body, expected_status",
|
"body, expected_status", generate_body_and_status({"name": "str"})
|
||||||
[
|
|
||||||
({"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(
|
||||||
|
|
@ -67,6 +61,13 @@ 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})
|
||||||
|
|
@ -82,12 +83,7 @@ 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"] == "Test Org"
|
assert data["organisation"]["name"] == "Org One"
|
||||||
|
|
||||||
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"] == "Test Org"
|
assert org["name"] == "Org One"
|
||||||
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@test.org"
|
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
||||||
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@test.org"
|
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
||||||
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@test.org"
|
assert org["security_contact"]["email"] == "security@orgone.com"
|
||||||
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": "Test Org",
|
"name": "Org One",
|
||||||
"status": "approved",
|
"status": "approved",
|
||||||
"root_user_email": "admin@test.com",
|
"root_user_email": "admin@test.com",
|
||||||
"owner_contact": {"email": "owner@test.org", "id": 2},
|
"owner_contact": {"email": "owner@orgone.com", "id": 2},
|
||||||
"security_contact": {"email": "security@test.org", "id": 3},
|
"security_contact": {"email": "security@orgone.com", "id": 3},
|
||||||
"billing_contact": {"email": "billing@test.org", "id": 1},
|
"billing_contact": {"email": "billing@orgone.com", "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