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.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.iam.models import Group, Permission
|
||||
from src.auth.service import get_current_user, get_dev_user
|
||||
|
|
@ -91,13 +91,35 @@ def _seed(db):
|
|||
oidc_id="abcd-efgh-ijkl-mnop",
|
||||
)
|
||||
)
|
||||
db.add(Contact(org_id=1, email="billing@test.org", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=1, email="owner@test.org", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=1, email="security@test.org", phonenumber="07521539927"))
|
||||
db.add(
|
||||
User(
|
||||
email="user@orgone.com",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-qwer",
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
User(
|
||||
email="root@orgtwo.com",
|
||||
first_name="Root",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-hjkl",
|
||||
)
|
||||
)
|
||||
db.add(Contact(org_id=1, email="billing@orgone.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=1, email="owner@orgone.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=1, email="security@orgone.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=2, email="billing@orgtwo.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=2, email="owner@orgtwo.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=2, email="security@orgtwo.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=3, email="billing@orgthree.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=3, email="owner@orgthree.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=3, email="security@orgthree.com", phonenumber="07521539927"))
|
||||
db.flush()
|
||||
db.add(
|
||||
Org(
|
||||
name="Test Org",
|
||||
name="Org One",
|
||||
root_user_id=1,
|
||||
billing_contact_id=1,
|
||||
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(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()
|
||||
group_model = db.get(Group, 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
|
||||
|
||||
|
||||
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():
|
||||
routes = []
|
||||
|
||||
|
|
|
|||
|
|
@ -7,27 +7,15 @@ Delete endpoints are currently skipped because the testing system cannot use bod
|
|||
import pytest
|
||||
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 = [
|
||||
pytest.mark.auth,
|
||||
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
|
||||
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 == 200
|
||||
|
||||
|
|
@ -37,7 +25,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
|
|||
resp = await default_client.patch(
|
||||
"/org/questionnaire",
|
||||
json={
|
||||
"organisation_id": 1,
|
||||
"organisation_id": 3,
|
||||
"intake_questionnaire": {
|
||||
"question_one": "new answer one",
|
||||
"question_two": None,
|
||||
|
|
@ -53,7 +41,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
|
|||
@pytest.mark.anyio
|
||||
async def test_patch_org_status_auth_approval(default_client: AsyncClient):
|
||||
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 == 200
|
||||
|
|
@ -61,48 +49,24 @@ async def test_patch_org_status_auth_approval(default_client: AsyncClient):
|
|||
|
||||
@pytest.mark.anyio
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_post_org_user_auth_approval(default_client: AsyncClient):
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_patch_org_root_user_auth_approval(default_client: AsyncClient):
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
|
@ -110,14 +74,14 @@ async def test_patch_org_root_user_auth_approval(
|
|||
|
||||
@pytest.mark.anyio
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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 == 200
|
||||
|
||||
|
|
@ -127,7 +91,7 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
|
|||
resp = await default_client.patch(
|
||||
"/org/contact",
|
||||
json={
|
||||
"organisation_id": 1,
|
||||
"organisation_id": 3,
|
||||
"contact_type": "billing",
|
||||
"email": "user@example.com",
|
||||
},
|
||||
|
|
@ -138,21 +102,21 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
|
|||
|
||||
@pytest.mark.anyio
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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 "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
|
||||
async def test_post_iam_group_auth_approval(default_client: AsyncClient):
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
async def test_put_iam_group_permission_auth_approval(default_client: AsyncClient):
|
||||
resp = await default_client.put(
|
||||
"/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 "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
|
@ -203,7 +151,7 @@ async def test_put_iam_group_user_auth_approval(
|
|||
|
||||
@pytest.mark.anyio
|
||||
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 "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
|
||||
async def test_post_iam_permissions_search_auth_approval(default_client: AsyncClient):
|
||||
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 "has not been approved." in resp.json()["detail"]
|
||||
|
|
|
|||
|
|
@ -3,9 +3,6 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from src.organisation.models import Organisation as Org
|
||||
from src.user.models import User
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.auth,
|
||||
|
|
@ -13,34 +10,9 @@ pytestmark = [
|
|||
|
||||
|
||||
@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
|
||||
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")
|
||||
assert resp.status_code != 422
|
||||
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
|
||||
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 = [
|
||||
pytest.mark.auth,
|
||||
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
|
||||
async def test_get_org_auth_root(no_su_client: AsyncClient):
|
||||
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
|
||||
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()
|
||||
async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient):
|
||||
resp = await no_su_client.put(
|
||||
"/iam/group/permission",
|
||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 2},
|
||||
|
|
@ -158,17 +124,9 @@ async def test_put_iam_group_permission_auth_root(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_iam_group_user_auth_root(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()
|
||||
|
||||
async def test_put_iam_group_user_auth_root(
|
||||
no_su_client: AsyncClient,
|
||||
):
|
||||
resp = await no_su_client.put(
|
||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 2}
|
||||
)
|
||||
|
|
|
|||
|
|
@ -6,10 +6,6 @@ DELETE endpoints are not tested
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from src.organisation.models import OrgUsers
|
||||
from src.user.models import User
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.auth,
|
||||
pytest.mark.super_admin,
|
||||
|
|
@ -35,19 +31,7 @@ async def test_patch_org_status_auth_su(no_su_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_patch_org_root_user_auth_su(no_su_client: AsyncClient):
|
||||
resp = await no_su_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
|
@ -73,7 +57,7 @@ async def test_post_service_auth_su(no_su_client: AsyncClient):
|
|||
|
||||
|
||||
@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(
|
||||
"/iam/permission",
|
||||
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
|
||||
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()
|
||||
|
||||
async def test_post_org_user_auth_su(no_su_client: AsyncClient):
|
||||
resp = await no_su_client.post(
|
||||
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
|
|
|||
310
test/test_iam.py
310
test/test_iam.py
|
|
@ -3,12 +3,7 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
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
|
||||
|
||||
from .conftest import generate_query_and_status, generate_body_and_status
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.iam_module,
|
||||
|
|
@ -20,7 +15,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
|
|||
body = {
|
||||
"rn": {
|
||||
"service": "Test Service",
|
||||
"organisation": "Test Org",
|
||||
"organisation": "Org One",
|
||||
"resource": "test_resource",
|
||||
"instance": None,
|
||||
},
|
||||
|
|
@ -45,7 +40,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
|
|||
)
|
||||
@pytest.mark.anyio
|
||||
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 = {
|
||||
"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", "Test Org", None, "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
|
||||
async def test_act_on_resource_endpoint_status_checks(
|
||||
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 = {
|
||||
"Authorization": "Bearer not_checked_when_auth_is_disabled",
|
||||
"X-API-Key": "123456789",
|
||||
}
|
||||
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
|
||||
|
|
@ -118,16 +118,15 @@ async def test_act_on_resource_endpoint_status_checks(
|
|||
@pytest.mark.parametrize(
|
||||
"service, org, resource, action, expected_response",
|
||||
[
|
||||
("Test Service", "Test Org", "test_resource", "read", True),
|
||||
("Test Service", "Test Org", "test_resource", "create", False),
|
||||
("Test Service", "Test Org", "no_access_here", "read", False),
|
||||
("Test Service", "Test Org Two", "test_resource", "read", False),
|
||||
("Test Service", "Org One", "test_resource", "read", True),
|
||||
("Test Service", "Org One", "test_resource", "create", False),
|
||||
("Test Service", "Org One", "no_access_here", "read", False),
|
||||
("Test Service", "Org Two", "test_resource", "read", False),
|
||||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_act_on_resource_logic(
|
||||
default_client: AsyncClient,
|
||||
db_session,
|
||||
service,
|
||||
org,
|
||||
resource,
|
||||
|
|
@ -173,7 +172,7 @@ async def test_get_group_permissions_success(default_client: AsyncClient):
|
|||
)
|
||||
@pytest.mark.anyio
|
||||
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}")
|
||||
|
||||
|
|
@ -188,21 +187,7 @@ async def test_get_group_permissions_status_checks(
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
async def test_get_group_permissions_mismatch(default_client: AsyncClient, query: str):
|
||||
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -226,12 +211,12 @@ async def test_get_group_users_success(default_client: AsyncClient):
|
|||
assert "group" in data
|
||||
assert isinstance(data["group"], dict)
|
||||
assert data["group"]["id"] == 1
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
assert data["group"]["name"] == "Org One Group"
|
||||
|
||||
assert "organisation" in data
|
||||
assert isinstance(data["organisation"], dict)
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -254,21 +239,7 @@ async def test_get_group_users_status_checks(
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
async def test_get_group_users_mismatch(default_client: AsyncClient, query: str):
|
||||
resp = await default_client.get(f"/iam/group/users?{query}")
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -287,37 +258,12 @@ async def test_post_group_success(default_client: AsyncClient):
|
|||
assert "group" in data
|
||||
assert isinstance(data["group"], dict)
|
||||
assert data["group"]["name"] == "New Group"
|
||||
assert data["group"]["id"] == 2
|
||||
assert data["group"]["id"] == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"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
|
||||
],
|
||||
generate_body_and_status({"organisation_id": "int", "name": "str"}),
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_post_group_status_checks(
|
||||
|
|
@ -329,12 +275,19 @@ async def test_post_group_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
||||
db_session.add(Group(name="Test Group Two", org_id=1))
|
||||
db_session.flush()
|
||||
async def test_post_group_conflict(default_client: AsyncClient):
|
||||
resp = await default_client.post(
|
||||
"/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(
|
||||
"/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
|
||||
|
||||
|
|
@ -342,8 +295,8 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
|||
|
||||
assert "group" in data
|
||||
assert isinstance(data["group"], dict)
|
||||
assert data["group"]["name"] == "Test Group Two"
|
||||
assert data["group"]["id"] == 2
|
||||
assert data["group"]["name"] == "Org One Group Two"
|
||||
assert data["group"]["id"] == 3
|
||||
|
||||
assert "permissions" in data
|
||||
assert isinstance(data["permissions"], list)
|
||||
|
|
@ -357,67 +310,9 @@ async def test_put_group_perm_success(default_client: AsyncClient, db_session):
|
|||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
(
|
||||
{"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
|
||||
],
|
||||
generate_body_and_status(
|
||||
{"organisation_id": "int", "group_id": "int", "permission_id": "int"}
|
||||
),
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
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
|
||||
|
||||
|
||||
@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(
|
||||
"body",
|
||||
[
|
||||
|
|
@ -436,21 +341,7 @@ async def test_put_group_perm_status_checks(
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
|
||||
resp = await default_client.put("/iam/group/permission", json=body)
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -458,19 +349,7 @@ async def test_put_group_perm_mismatch(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_put_group_user_success(default_client: AsyncClient):
|
||||
resp = await default_client.put(
|
||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
|
||||
)
|
||||
|
|
@ -480,7 +359,7 @@ async def test_put_group_user_success(default_client: AsyncClient, db_session):
|
|||
|
||||
assert "group" in data
|
||||
assert isinstance(data["group"], dict)
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
assert data["group"]["name"] == "Org One Group"
|
||||
assert data["group"]["id"] == 1
|
||||
|
||||
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["first_name"] == "User"
|
||||
assert user["last_name"] == "Test"
|
||||
assert user["email"] == "user@test.org"
|
||||
assert user["email"] == "user@orgone.com"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
(
|
||||
{"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
|
||||
],
|
||||
generate_body_and_status(
|
||||
{"organisation_id": "int", "group_id": "int", "user_id": "int"}
|
||||
),
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_put_group_user_status_checks(
|
||||
|
|
@ -582,7 +416,7 @@ async def test_get_permissions_status_checks(
|
|||
|
||||
|
||||
@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(
|
||||
"/iam/permission",
|
||||
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 isinstance(data["permission"], dict)
|
||||
|
||||
assert data["permission"]["id"] == 2
|
||||
assert data["permission"]["id"] == 3
|
||||
assert data["permission"]["service_name"] == "Test Service"
|
||||
assert data["permission"]["resource"] == "test_resource"
|
||||
assert data["permission"]["action"] == "create"
|
||||
|
|
@ -671,7 +505,7 @@ async def test_post_perm_status_checks(
|
|||
],
|
||||
)
|
||||
@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)
|
||||
data = resp.json()
|
||||
|
||||
|
|
@ -679,7 +513,11 @@ async def test_post_perm_search_success(default_client: AsyncClient, db_session,
|
|||
assert "permissions" in data
|
||||
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["service_name"] == "Test Service"
|
||||
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 "group" in data
|
||||
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
|
||||
|
|
@ -816,4 +674,4 @@ async def test_delete_group_users_success(default_client: AsyncClient):
|
|||
assert len(data["users"]) == 0
|
||||
assert "group" in data
|
||||
assert data["group"]["id"] == 1
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
assert data["group"]["name"] == "Org One Group"
|
||||
|
|
|
|||
|
|
@ -5,8 +5,6 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from src.organisation.models import Organisation, OrgUsers
|
||||
from src.user.models import User
|
||||
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 org["organisation_id"] == 1
|
||||
assert org["name"] == "Test Org"
|
||||
assert org["name"] == "Org One"
|
||||
assert org["status"] == "approved"
|
||||
assert org["root_user_email"] == "admin@test.com"
|
||||
assert "intake_questionnaire" in org
|
||||
assert isinstance(org["intake_questionnaire"], dict)
|
||||
|
||||
assert org["billing_contact"]["email"] == "billing@test.org"
|
||||
assert org["owner_contact"]["email"] == "owner@test.org"
|
||||
assert org["security_contact"]["email"] == "security@test.org"
|
||||
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
||||
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
||||
assert org["security_contact"]["email"] == "security@orgone.com"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -62,7 +60,7 @@ async def test_post_org_success(default_client: AsyncClient):
|
|||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"name": "Test Org"}, 409),
|
||||
({"name": "Org One"}, 409),
|
||||
({"name": 42}, 422),
|
||||
({}, 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
|
||||
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()
|
||||
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient):
|
||||
resp = await default_client.patch(
|
||||
"/org/questionnaire",
|
||||
json={
|
||||
"organisation_id": 1,
|
||||
"organisation_id": 3,
|
||||
"intake_questionnaire": {
|
||||
"question_one": "new answer one",
|
||||
"question_two": None,
|
||||
|
|
@ -99,7 +92,7 @@ async def test_patch_org_questionnaire_partial_success(
|
|||
data = resp.json()
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["name"] == "Org Three"
|
||||
assert data["status"] == "partial"
|
||||
assert "intake_questionnaire" in data
|
||||
assert isinstance(data["intake_questionnaire"], dict)
|
||||
|
|
@ -116,7 +109,7 @@ async def test_patch_org_questionnaire_partial_success(
|
|||
"body, expected_status",
|
||||
[
|
||||
({"organisation_id": 42}, 404),
|
||||
({"organisation_id": "Test Org"}, 422),
|
||||
({"organisation_id": "Org One"}, 422),
|
||||
({"organisation_id": ""}, 422),
|
||||
({}, 422),
|
||||
(
|
||||
|
|
@ -151,16 +144,11 @@ async def test_patch_questionnaire_partial_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
async def test_patch_org_questionnaire_submit_success(default_client: AsyncClient):
|
||||
resp = await default_client.patch(
|
||||
"/org/questionnaire",
|
||||
json={
|
||||
"organisation_id": 1,
|
||||
"organisation_id": 3,
|
||||
"intake_questionnaire": {
|
||||
"question_one": "new answer one",
|
||||
"question_two": None,
|
||||
|
|
@ -172,7 +160,7 @@ async def test_patch_org_questionnaire_submit_success(
|
|||
data = resp.json()
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["name"] == "Org Three"
|
||||
assert data["status"] == "submitted"
|
||||
assert "intake_questionnaire" in data
|
||||
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()
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["name"] == "Org One"
|
||||
assert data["status"] == status
|
||||
|
||||
|
||||
|
|
@ -204,7 +192,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
|
|||
"body, expected_status",
|
||||
[
|
||||
({"organisation_id": 42}, 404),
|
||||
({"organisation_id": "Test Org"}, 422),
|
||||
({"organisation_id": "Org One"}, 422),
|
||||
({"organisation_id": ""}, 422),
|
||||
({}, 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 isinstance(data["users"], list)
|
||||
assert len(data["users"]) == 1
|
||||
assert len(data["users"]) == 2
|
||||
|
||||
user = data["users"][0]
|
||||
assert isinstance(user, dict)
|
||||
|
|
@ -237,7 +225,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
|
|||
assert user["id"] == 1
|
||||
|
||||
assert "organisation" in data
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
assert data["organisation"]["id"] == 1
|
||||
|
||||
|
||||
|
|
@ -254,19 +242,9 @@ async def test_get_org_users_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_post_org_user_success(default_client: AsyncClient):
|
||||
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
|
||||
|
|
@ -276,12 +254,12 @@ async def test_post_org_user_success(default_client: AsyncClient, db_session):
|
|||
assert "organisation" in data
|
||||
assert isinstance(data["organisation"], dict)
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
|
||||
assert "users" in data
|
||||
assert isinstance(data["users"], list)
|
||||
assert (
|
||||
len([user for user in data["users"] if user["email"] == "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
|
||||
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)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_patch_org_root_user_success(default_client: AsyncClient):
|
||||
resp = await default_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
|
@ -336,15 +292,15 @@ async def test_patch_org_root_user_success(default_client: AsyncClient, db_sessi
|
|||
|
||||
data = resp.json()
|
||||
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["root_user_email"] == "user@test.org"
|
||||
assert data["name"] == "Org One"
|
||||
assert data["root_user_email"] == "user@orgone.com"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"organisation_id": 42, "user_id": 2}, 404),
|
||||
({"organisation_id": "Test Org", "user_id": 2}, 422),
|
||||
({"organisation_id": "Org One", "user_id": 2}, 422),
|
||||
({"organisation_id": "", "user_id": 2}, 422),
|
||||
({}, 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
|
||||
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)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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()
|
||||
|
||||
async def test_patch_org_root_user_non_member(default_client: AsyncClient):
|
||||
resp = await default_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 3}
|
||||
)
|
||||
data = resp.json()
|
||||
|
||||
|
|
@ -404,14 +338,14 @@ async def test_get_org_groups_success(default_client: AsyncClient):
|
|||
assert "organisation" in data
|
||||
assert isinstance(data["organisation"], dict)
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
|
||||
assert "groups" in data
|
||||
assert isinstance(data["groups"], list)
|
||||
group = data["groups"][0]
|
||||
assert isinstance(group, dict)
|
||||
assert group["id"] == 1
|
||||
assert group["name"] == "Test Group"
|
||||
assert group["name"] == "Org One Group"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -438,7 +372,7 @@ async def test_get_org_contact_success(default_client: AsyncClient, contact_type
|
|||
|
||||
assert "organisation" in data
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
|
||||
attributes = [
|
||||
"email",
|
||||
|
|
@ -516,7 +450,7 @@ async def test_patch_org_contact_success(
|
|||
|
||||
assert "organisation" in data
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
|
||||
attributes = [
|
||||
"email",
|
||||
|
|
@ -557,7 +491,7 @@ async def test_patch_org_contact_success(
|
|||
({"organisation_id": 42, "contact_type": "billing"}, 404),
|
||||
({"organisation_id": 1, "contact_type": "security"}, 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),
|
||||
({}, 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
|
||||
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()
|
||||
async def test_delete_org_users_success(default_client: AsyncClient):
|
||||
resp = await default_client.delete("/org/user?org_id=1&user_id=2")
|
||||
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_preapproval_org_success(db_session, default_client: AsyncClient):
|
||||
org_model = db_session.get(Organisation, 1)
|
||||
org_model.status = "partial"
|
||||
db_session.flush()
|
||||
resp = await default_client.delete("/org/self?org_id=1")
|
||||
async def test_delete_preapproval_org_success(default_client: AsyncClient):
|
||||
resp = await default_client.delete("/org/self?org_id=3")
|
||||
|
||||
assert resp.status_code == 204
|
||||
|
|
|
|||
|
|
@ -5,8 +5,7 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from .conftest import generate_query_and_status
|
||||
|
||||
from .conftest import generate_query_and_status, generate_body_and_status
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.service_module,
|
||||
|
|
@ -51,12 +50,7 @@ async def test_post_service_success(default_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"name": "Test Service"}, 409),
|
||||
({"name": 42}, 422),
|
||||
({}, 422),
|
||||
],
|
||||
"body, expected_status", generate_body_and_status({"name": "str"})
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_post_service_status_checks(
|
||||
|
|
@ -67,6 +61,13 @@ async def test_post_service_status_checks(
|
|||
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
|
||||
async def test_patch_service_success(default_client: AsyncClient):
|
||||
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(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"service_id": 42}, 404),
|
||||
({"service_id": "Test Service"}, 422),
|
||||
({"service_id": ""}, 422),
|
||||
({}, 422),
|
||||
],
|
||||
generate_body_and_status({"service_id": "int"}),
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
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 isinstance(data["organisation"], dict)
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
|
||||
assert "invited_email" in data
|
||||
assert isinstance(data["invited_email"], str)
|
||||
|
|
@ -135,22 +135,22 @@ async def test_get_self_orgs_success(default_client: AsyncClient):
|
|||
|
||||
org = data["organisations"][0]
|
||||
assert org["organisation_id"] == 1
|
||||
assert org["name"] == "Test Org"
|
||||
assert org["name"] == "Org One"
|
||||
assert org["status"] == "approved"
|
||||
assert org["root_user_email"] == "admin@test.com"
|
||||
assert "intake_questionnaire" in org
|
||||
assert isinstance(org["intake_questionnaire"], dict)
|
||||
|
||||
assert isinstance(org["billing_contact"], dict)
|
||||
assert org["billing_contact"]["email"] == "billing@test.org"
|
||||
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
||||
assert org["billing_contact"]["id"] == 1
|
||||
|
||||
assert isinstance(org["owner_contact"], dict)
|
||||
assert org["owner_contact"]["email"] == "owner@test.org"
|
||||
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
||||
assert org["owner_contact"]["id"] == 2
|
||||
|
||||
assert isinstance(org["security_contact"], dict)
|
||||
assert org["security_contact"]["email"] == "security@test.org"
|
||||
assert org["security_contact"]["email"] == "security@orgone.com"
|
||||
assert org["security_contact"]["id"] == 3
|
||||
|
||||
|
||||
|
|
@ -162,12 +162,12 @@ async def test_get_self_orgs_dynamic(default_client: AsyncClient):
|
|||
"organisations": [
|
||||
{
|
||||
"organisation_id": 1,
|
||||
"name": "Test Org",
|
||||
"name": "Org One",
|
||||
"status": "approved",
|
||||
"root_user_email": "admin@test.com",
|
||||
"owner_contact": {"email": "owner@test.org", "id": 2},
|
||||
"security_contact": {"email": "security@test.org", "id": 3},
|
||||
"billing_contact": {"email": "billing@test.org", "id": 1},
|
||||
"owner_contact": {"email": "owner@orgone.com", "id": 2},
|
||||
"security_contact": {"email": "security@orgone.com", "id": 3},
|
||||
"billing_contact": {"email": "billing@orgone.com", "id": 1},
|
||||
"intake_questionnaire": {
|
||||
"questions": {
|
||||
"question_one": None,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue