Compare commits
No commits in common. "a655eaf543d56dc13a6210e6aa360f3f2d7e8b26" and "8b89595531e5c6bc4d8b88dbb33cb50642a138fa" have entirely different histories.
a655eaf543
...
8b89595531
9 changed files with 547 additions and 268 deletions
105
test/conftest.py
105
test/conftest.py
|
|
@ -8,7 +8,7 @@ from sqlalchemy.orm import sessionmaker
|
|||
|
||||
from src.user.models import User
|
||||
from src.service.models import Service
|
||||
from src.organisation.models import Organisation as Org, OrgUsers
|
||||
from src.organisation.models import Organisation as Org
|
||||
from src.contact.models import Contact
|
||||
from src.iam.models import Group, Permission
|
||||
from src.auth.service import get_current_user, get_dev_user
|
||||
|
|
@ -91,35 +91,13 @@ def _seed(db):
|
|||
oidc_id="abcd-efgh-ijkl-mnop",
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
User(
|
||||
email="user@orgone.com",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-qwer",
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
User(
|
||||
email="root@orgtwo.com",
|
||||
first_name="Root",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-hjkl",
|
||||
)
|
||||
)
|
||||
db.add(Contact(org_id=1, email="billing@orgone.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=1, email="owner@orgone.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=1, email="security@orgone.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=2, email="billing@orgtwo.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=2, email="owner@orgtwo.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=2, email="security@orgtwo.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=3, email="billing@orgthree.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=3, email="owner@orgthree.com", phonenumber="07521539927"))
|
||||
db.add(Contact(org_id=3, email="security@orgthree.com", phonenumber="07521539927"))
|
||||
db.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.flush()
|
||||
db.add(
|
||||
Org(
|
||||
name="Org One",
|
||||
name="Test Org",
|
||||
root_user_id=1,
|
||||
billing_contact_id=1,
|
||||
owner_contact_id=2,
|
||||
|
|
@ -131,41 +109,9 @@ def _seed(db):
|
|||
},
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
Org(
|
||||
name="Org Two",
|
||||
root_user_id=3,
|
||||
billing_contact_id=4,
|
||||
owner_contact_id=5,
|
||||
security_contact_id=6,
|
||||
status="approved",
|
||||
intake_questionnaire={
|
||||
"metadata": {"version": 0, "submission_date": None},
|
||||
"questions": {"question_two": "answer two"},
|
||||
},
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
Org(
|
||||
name="Org Three",
|
||||
root_user_id=1,
|
||||
billing_contact_id=7,
|
||||
owner_contact_id=8,
|
||||
security_contact_id=9,
|
||||
status="partial",
|
||||
intake_questionnaire={
|
||||
"metadata": {"version": 0, "submission_date": None},
|
||||
"questions": {"question_two": "answer two"},
|
||||
},
|
||||
)
|
||||
)
|
||||
db.add(OrgUsers(org_id=1, user_id=2))
|
||||
db.add(Service(name="Test Service", api_key="123456789"))
|
||||
db.add(Permission(service_id=1, resource="test_resource", action="read"))
|
||||
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.add(Group(name="Test Group", org_id=1))
|
||||
db.flush()
|
||||
group_model = db.get(Group, 1)
|
||||
perm_model = db.get(Permission, 1)
|
||||
|
|
@ -213,45 +159,6 @@ def generate_query_and_status(params) -> list[tuple[str, int]]:
|
|||
return query_and_status
|
||||
|
||||
|
||||
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,15 +7,27 @@ 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=3")
|
||||
resp = await default_client.get("/org?org_id=1")
|
||||
assert resp.status_code != 422
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
|
@ -25,7 +37,7 @@ async def test_patch_org_questionnaire_auth_approval(default_client: AsyncClient
|
|||
resp = await default_client.patch(
|
||||
"/org/questionnaire",
|
||||
json={
|
||||
"organisation_id": 3,
|
||||
"organisation_id": 1,
|
||||
"intake_questionnaire": {
|
||||
"question_one": "new answer one",
|
||||
"question_two": None,
|
||||
|
|
@ -41,7 +53,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": 3, "status": "submitted"}
|
||||
"/org/status", json={"organisation_id": 1, "status": "submitted"}
|
||||
)
|
||||
assert resp.status_code != 422
|
||||
assert resp.status_code == 200
|
||||
|
|
@ -49,24 +61,48 @@ 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=3")
|
||||
resp = await default_client.get("/org/users?org_id=1")
|
||||
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):
|
||||
async def test_post_org_user_auth_approval(default_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.post(
|
||||
"/org/user", json={"organisation_id": 3, "user_id": 2}
|
||||
"/org/user", json={"organisation_id": 1, "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):
|
||||
async def test_patch_org_root_user_auth_approval(
|
||||
default_client: AsyncClient, db_session
|
||||
):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 3, "user_id": 2}
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
|
@ -74,14 +110,14 @@ async def test_patch_org_root_user_auth_approval(default_client: AsyncClient):
|
|||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_org_groups_auth_approval(default_client: AsyncClient):
|
||||
resp = await default_client.get("/org/groups?org_id=3")
|
||||
resp = await default_client.get("/org/groups?org_id=1")
|
||||
assert resp.status_code != 422
|
||||
assert "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=3&contact_type=billing")
|
||||
resp = await default_client.get("/org/contact?org_id=1&contact_type=billing")
|
||||
assert resp.status_code != 422
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
|
@ -91,7 +127,7 @@ async def test_patch_org_contact_auth_approval(default_client: AsyncClient):
|
|||
resp = await default_client.patch(
|
||||
"/org/contact",
|
||||
json={
|
||||
"organisation_id": 3,
|
||||
"organisation_id": 1,
|
||||
"contact_type": "billing",
|
||||
"email": "user@example.com",
|
||||
},
|
||||
|
|
@ -102,21 +138,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=3")
|
||||
resp = await default_client.get("/service?org_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_permissions_auth_approval(default_client: AsyncClient):
|
||||
resp = await default_client.get("/iam/group/permissions?org_id=3&group_id=1")
|
||||
resp = await default_client.get("/iam/group/permissions?org_id=1&group_id=1")
|
||||
assert resp.status_code != 422
|
||||
assert "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=3&group_id=1")
|
||||
resp = await default_client.get("/iam/group/users?org_id=1&group_id=1")
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
|
@ -124,26 +160,42 @@ async def test_get_iam_group_users_auth_approval(default_client: AsyncClient):
|
|||
@pytest.mark.anyio
|
||||
async def test_post_iam_group_auth_approval(default_client: AsyncClient):
|
||||
resp = await default_client.post(
|
||||
"/iam/group", json={"name": "New Group", "organisation_id": 3}
|
||||
"/iam/group", json={"name": "New Group", "organisation_id": 1}
|
||||
)
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_iam_group_permission_auth_approval(default_client: AsyncClient):
|
||||
async def test_put_iam_group_permission_auth_approval(
|
||||
default_client: AsyncClient, db_session
|
||||
):
|
||||
db_session.add(Group(name="Test Group Two", org_id=1))
|
||||
db_session.flush()
|
||||
resp = await default_client.put(
|
||||
"/iam/group/permission",
|
||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 3},
|
||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
|
||||
)
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
|
||||
async def test_put_iam_group_user_auth_approval(
|
||||
default_client: AsyncClient, db_session
|
||||
):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.put(
|
||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 3}
|
||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
|
||||
)
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
|
@ -151,7 +203,7 @@ async def test_put_iam_group_user_auth_approval(default_client: AsyncClient):
|
|||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
|
||||
resp = await default_client.get("/iam/permissions?org_id=3")
|
||||
resp = await default_client.get("/iam/permissions?org_id=1")
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
||||
|
|
@ -159,7 +211,7 @@ async def test_get_iam_permissions_auth_approval(default_client: AsyncClient):
|
|||
@pytest.mark.anyio
|
||||
async def test_post_iam_permissions_search_auth_approval(default_client: AsyncClient):
|
||||
resp = await default_client.post(
|
||||
"/iam/permissions/search", json={"organisation_id": 3, "action": "read"}
|
||||
"/iam/permissions/search", json={"organisation_id": 1, "action": "read"}
|
||||
)
|
||||
assert resp.status_code != 422
|
||||
assert "has not been approved." in resp.json()["detail"]
|
||||
|
|
|
|||
|
|
@ -3,6 +3,9 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from src.organisation.models import Organisation as Org
|
||||
from src.user.models import User
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.auth,
|
||||
|
|
@ -10,9 +13,34 @@ pytestmark = [
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_org_auth_root_su(default_client: AsyncClient):
|
||||
async def test_get_org_auth_root_su(default_client: AsyncClient, db_session):
|
||||
# If a super admin can access a resource when not the root user
|
||||
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"] == "Org Two"
|
||||
assert resp.json()["organisations"][0]["name"] == "Test Org Two"
|
||||
|
|
|
|||
|
|
@ -6,12 +6,42 @@ 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")
|
||||
|
|
@ -113,7 +143,11 @@ 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):
|
||||
async def test_put_iam_group_permission_auth_root(
|
||||
no_su_client: AsyncClient, db_session
|
||||
):
|
||||
db_session.add(Group(name="Test Group Two", org_id=2))
|
||||
db_session.flush()
|
||||
resp = await no_su_client.put(
|
||||
"/iam/group/permission",
|
||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 2},
|
||||
|
|
@ -124,9 +158,17 @@ async def test_put_iam_group_permission_auth_root(no_su_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_iam_group_user_auth_root(
|
||||
no_su_client: AsyncClient,
|
||||
):
|
||||
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()
|
||||
|
||||
resp = await no_su_client.put(
|
||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 2}
|
||||
)
|
||||
|
|
|
|||
|
|
@ -6,6 +6,10 @@ 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,
|
||||
|
|
@ -31,7 +35,19 @@ 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):
|
||||
async def test_patch_org_root_user_auth_su(no_su_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||
db_session.flush()
|
||||
|
||||
resp = await no_su_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
|
@ -57,7 +73,7 @@ async def test_post_service_auth_su(no_su_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_perm_auth_su(no_su_client: AsyncClient):
|
||||
async def test_post_perm_auth_su(no_su_client: AsyncClient, db_session):
|
||||
resp = await no_su_client.post(
|
||||
"/iam/permission",
|
||||
json={"service_id": 1, "resource": "test_resource", "action": "create"},
|
||||
|
|
@ -68,7 +84,17 @@ async def test_post_perm_auth_su(no_su_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_org_user_auth_su(no_su_client: AsyncClient):
|
||||
async def test_post_org_user_auth_su(no_su_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
|
||||
resp = await no_su_client.post(
|
||||
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
|
|
|||
310
test/test_iam.py
310
test/test_iam.py
|
|
@ -3,7 +3,12 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from .conftest import generate_query_and_status, generate_body_and_status
|
||||
from src.user.models import User
|
||||
from src.organisation.models import Organisation as Org, OrgUsers
|
||||
from src.iam.models import Group
|
||||
|
||||
from .conftest import generate_query_and_status
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.iam_module,
|
||||
|
|
@ -15,7 +20,7 @@ async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient
|
|||
body = {
|
||||
"rn": {
|
||||
"service": "Test Service",
|
||||
"organisation": "Org One",
|
||||
"organisation": "Test Org",
|
||||
"resource": "test_resource",
|
||||
"instance": None,
|
||||
},
|
||||
|
|
@ -40,7 +45,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, service: str, api_key: str
|
||||
default_client: AsyncClient, db_session, service: str, api_key: str
|
||||
):
|
||||
body = {
|
||||
"rn": {
|
||||
|
|
@ -92,24 +97,19 @@ async def test_act_on_resource_missing_key(default_client: AsyncClient):
|
|||
("Test Service", 42, "test_resource", "read", 422),
|
||||
("Test Service", "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 = {
|
||||
"rn": {"service": service, "organisation": org, "resource": resource},
|
||||
"action": action,
|
||||
}
|
||||
body = {"service": service, "organisation": org, "resource": resource}
|
||||
headers = {
|
||||
"Authorization": "Bearer not_checked_when_auth_is_disabled",
|
||||
"X-API-Key": "123456789",
|
||||
}
|
||||
resp = await default_client.post(
|
||||
"/iam/can_act_on_resource", json=body, headers=headers
|
||||
f"/iam/can_act_on_resource?action={action}", json=body, headers=headers
|
||||
)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
|
@ -118,15 +118,16 @@ async def test_act_on_resource_endpoint_status_checks(
|
|||
@pytest.mark.parametrize(
|
||||
"service, org, resource, action, expected_response",
|
||||
[
|
||||
("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),
|
||||
("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),
|
||||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_act_on_resource_logic(
|
||||
default_client: AsyncClient,
|
||||
db_session,
|
||||
service,
|
||||
org,
|
||||
resource,
|
||||
|
|
@ -172,7 +173,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, query: str, expected_status: int
|
||||
default_client: AsyncClient, db_session, query: str, expected_status: int
|
||||
):
|
||||
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
||||
|
||||
|
|
@ -187,7 +188,21 @@ async def test_get_group_permissions_status_checks(
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_get_group_permissions_mismatch(default_client: AsyncClient, query: str):
|
||||
async def test_get_group_permissions_mismatch(
|
||||
default_client: AsyncClient, db_session, query: str
|
||||
):
|
||||
db_session.add(
|
||||
Org(
|
||||
name="Another Test Org",
|
||||
root_user_id=1,
|
||||
billing_contact_id=1,
|
||||
owner_contact_id=2,
|
||||
security_contact_id=3,
|
||||
status="approved",
|
||||
)
|
||||
)
|
||||
db_session.add(Group(name="Another Test Group", org_id=2))
|
||||
db_session.flush()
|
||||
resp = await default_client.get(f"/iam/group/permissions?{query}")
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -211,12 +226,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"] == "Org One Group"
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
|
||||
assert "organisation" in data
|
||||
assert isinstance(data["organisation"], dict)
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -239,7 +254,21 @@ async def test_get_group_users_status_checks(
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_get_group_users_mismatch(default_client: AsyncClient, query: str):
|
||||
async def test_get_group_users_mismatch(
|
||||
default_client: AsyncClient, db_session, query: str
|
||||
):
|
||||
db_session.add(
|
||||
Org(
|
||||
name="Another Test Org",
|
||||
root_user_id=1,
|
||||
billing_contact_id=1,
|
||||
owner_contact_id=2,
|
||||
security_contact_id=3,
|
||||
status="approved",
|
||||
)
|
||||
)
|
||||
db_session.add(Group(name="Another Test Group", org_id=2))
|
||||
db_session.flush()
|
||||
resp = await default_client.get(f"/iam/group/users?{query}")
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -258,12 +287,37 @@ 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"] == 4
|
||||
assert data["group"]["id"] == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"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
|
||||
async def test_post_group_status_checks(
|
||||
|
|
@ -275,19 +329,12 @@ async def test_post_group_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
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):
|
||||
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()
|
||||
resp = await default_client.put(
|
||||
"/iam/group/permission",
|
||||
json={"permission_id": 1, "group_id": 3, "organisation_id": 1},
|
||||
json={"permission_id": 1, "group_id": 2, "organisation_id": 1},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
|
@ -295,8 +342,8 @@ async def test_put_group_perm_success(default_client: AsyncClient):
|
|||
|
||||
assert "group" in data
|
||||
assert isinstance(data["group"], dict)
|
||||
assert data["group"]["name"] == "Org One Group Two"
|
||||
assert data["group"]["id"] == 3
|
||||
assert data["group"]["name"] == "Test Group Two"
|
||||
assert data["group"]["id"] == 2
|
||||
|
||||
assert "permissions" in data
|
||||
assert isinstance(data["permissions"], list)
|
||||
|
|
@ -310,9 +357,67 @@ async def test_put_group_perm_success(default_client: AsyncClient):
|
|||
|
||||
@pytest.mark.parametrize(
|
||||
"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
|
||||
async def test_put_group_perm_status_checks(
|
||||
|
|
@ -323,16 +428,6 @@ async def test_put_group_perm_status_checks(
|
|||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@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",
|
||||
[
|
||||
|
|
@ -341,7 +436,21 @@ async def test_put_group_perm_conflict(default_client: AsyncClient):
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
|
||||
async def test_put_group_perm_mismatch(
|
||||
default_client: AsyncClient, db_session, body: dict
|
||||
):
|
||||
db_session.add(
|
||||
Org(
|
||||
name="Another Test Org",
|
||||
root_user_id=1,
|
||||
billing_contact_id=1,
|
||||
owner_contact_id=2,
|
||||
security_contact_id=3,
|
||||
status="approved",
|
||||
)
|
||||
)
|
||||
db_session.add(Group(name="Another Test Group", org_id=2))
|
||||
db_session.flush()
|
||||
resp = await default_client.put("/iam/group/permission", json=body)
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -349,7 +458,19 @@ async def test_put_group_perm_mismatch(default_client: AsyncClient, body: dict):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_group_user_success(default_client: AsyncClient):
|
||||
async def test_put_group_user_success(default_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
db_session.add(OrgUsers(user_id=2, org_id=1))
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.put(
|
||||
"/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}
|
||||
)
|
||||
|
|
@ -359,7 +480,7 @@ async def test_put_group_user_success(default_client: AsyncClient):
|
|||
|
||||
assert "group" in data
|
||||
assert isinstance(data["group"], dict)
|
||||
assert data["group"]["name"] == "Org One Group"
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
assert data["group"]["id"] == 1
|
||||
|
||||
assert "users" in data
|
||||
|
|
@ -369,14 +490,59 @@ async def test_put_group_user_success(default_client: AsyncClient):
|
|||
assert user["id"] == 2
|
||||
assert user["first_name"] == "User"
|
||||
assert user["last_name"] == "Test"
|
||||
assert user["email"] == "user@orgone.com"
|
||||
assert user["email"] == "user@test.org"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"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
|
||||
async def test_put_group_user_status_checks(
|
||||
|
|
@ -416,7 +582,7 @@ async def test_get_permissions_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_perm_success(default_client: AsyncClient):
|
||||
async def test_post_perm_success(default_client: AsyncClient, db_session):
|
||||
resp = await default_client.post(
|
||||
"/iam/permission",
|
||||
json={"service_id": 1, "resource": "test_resource", "action": "create"},
|
||||
|
|
@ -428,7 +594,7 @@ async def test_post_perm_success(default_client: AsyncClient):
|
|||
assert "permission" in data
|
||||
assert isinstance(data["permission"], dict)
|
||||
|
||||
assert data["permission"]["id"] == 3
|
||||
assert data["permission"]["id"] == 2
|
||||
assert data["permission"]["service_name"] == "Test Service"
|
||||
assert data["permission"]["resource"] == "test_resource"
|
||||
assert data["permission"]["action"] == "create"
|
||||
|
|
@ -505,7 +671,7 @@ async def test_post_perm_status_checks(
|
|||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_post_perm_search_success(default_client: AsyncClient, body):
|
||||
async def test_post_perm_search_success(default_client: AsyncClient, db_session, body):
|
||||
resp = await default_client.post("/iam/permissions/search", json=body)
|
||||
data = resp.json()
|
||||
|
||||
|
|
@ -513,11 +679,7 @@ async def test_post_perm_search_success(default_client: AsyncClient, body):
|
|||
assert "permissions" in data
|
||||
assert isinstance(data["permissions"], list)
|
||||
|
||||
permissions_filtered = [
|
||||
permission for permission in data["permissions"] if permission["id"] == 1
|
||||
]
|
||||
assert len(permissions_filtered) == 1
|
||||
permission = permissions_filtered[0]
|
||||
permission = data["permissions"][0]
|
||||
assert permission["id"] == 1
|
||||
assert permission["service_name"] == "Test Service"
|
||||
assert permission["resource"] == "test_resource"
|
||||
|
|
@ -633,27 +795,7 @@ async def test_delete_group_permissions_success(default_client: AsyncClient):
|
|||
assert len(data["permissions"]) == 0
|
||||
assert "group" in data
|
||||
assert data["group"]["id"] == 1
|
||||
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
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
|
|
@ -674,4 +816,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"] == "Org One Group"
|
||||
assert data["group"]["name"] == "Test Group"
|
||||
|
|
|
|||
|
|
@ -5,6 +5,8 @@
|
|||
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
|
||||
|
||||
|
||||
|
|
@ -24,15 +26,15 @@ async def test_get_org_success(default_client: AsyncClient):
|
|||
|
||||
assert isinstance(org, dict)
|
||||
assert org["organisation_id"] == 1
|
||||
assert org["name"] == "Org One"
|
||||
assert org["name"] == "Test Org"
|
||||
assert org["status"] == "approved"
|
||||
assert org["root_user_email"] == "admin@test.com"
|
||||
assert "intake_questionnaire" in org
|
||||
assert isinstance(org["intake_questionnaire"], dict)
|
||||
|
||||
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
||||
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
||||
assert org["security_contact"]["email"] == "security@orgone.com"
|
||||
assert org["billing_contact"]["email"] == "billing@test.org"
|
||||
assert org["owner_contact"]["email"] == "owner@test.org"
|
||||
assert org["security_contact"]["email"] == "security@test.org"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -60,7 +62,7 @@ async def test_post_org_success(default_client: AsyncClient):
|
|||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"name": "Org One"}, 409),
|
||||
({"name": "Test Org"}, 409),
|
||||
({"name": 42}, 422),
|
||||
({}, 422),
|
||||
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
|
||||
|
|
@ -76,11 +78,16 @@ async def test_post_org_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_patch_org_questionnaire_partial_success(default_client: AsyncClient):
|
||||
async def test_patch_org_questionnaire_partial_success(
|
||||
default_client: AsyncClient, db_session
|
||||
):
|
||||
org_model = db_session.get(Organisation, 1)
|
||||
org_model.status = "partial"
|
||||
db_session.flush()
|
||||
resp = await default_client.patch(
|
||||
"/org/questionnaire",
|
||||
json={
|
||||
"organisation_id": 3,
|
||||
"organisation_id": 1,
|
||||
"intake_questionnaire": {
|
||||
"question_one": "new answer one",
|
||||
"question_two": None,
|
||||
|
|
@ -92,7 +99,7 @@ async def test_patch_org_questionnaire_partial_success(default_client: AsyncClie
|
|||
data = resp.json()
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert data["name"] == "Org Three"
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["status"] == "partial"
|
||||
assert "intake_questionnaire" in data
|
||||
assert isinstance(data["intake_questionnaire"], dict)
|
||||
|
|
@ -109,7 +116,7 @@ async def test_patch_org_questionnaire_partial_success(default_client: AsyncClie
|
|||
"body, expected_status",
|
||||
[
|
||||
({"organisation_id": 42}, 404),
|
||||
({"organisation_id": "Org One"}, 422),
|
||||
({"organisation_id": "Test Org"}, 422),
|
||||
({"organisation_id": ""}, 422),
|
||||
({}, 422),
|
||||
(
|
||||
|
|
@ -144,11 +151,16 @@ async def test_patch_questionnaire_partial_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_patch_org_questionnaire_submit_success(default_client: AsyncClient):
|
||||
async def test_patch_org_questionnaire_submit_success(
|
||||
default_client: AsyncClient, db_session
|
||||
):
|
||||
org_model = db_session.get(Organisation, 1)
|
||||
org_model.status = "partial"
|
||||
db_session.flush()
|
||||
resp = await default_client.patch(
|
||||
"/org/questionnaire",
|
||||
json={
|
||||
"organisation_id": 3,
|
||||
"organisation_id": 1,
|
||||
"intake_questionnaire": {
|
||||
"question_one": "new answer one",
|
||||
"question_two": None,
|
||||
|
|
@ -160,7 +172,7 @@ async def test_patch_org_questionnaire_submit_success(default_client: AsyncClien
|
|||
data = resp.json()
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert data["name"] == "Org Three"
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["status"] == "submitted"
|
||||
assert "intake_questionnaire" in data
|
||||
assert isinstance(data["intake_questionnaire"], dict)
|
||||
|
|
@ -184,7 +196,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
|
|||
data = resp.json()
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert data["name"] == "Org One"
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["status"] == status
|
||||
|
||||
|
||||
|
|
@ -192,7 +204,7 @@ async def test_patch_org_status_success(default_client: AsyncClient, status: str
|
|||
"body, expected_status",
|
||||
[
|
||||
({"organisation_id": 42}, 404),
|
||||
({"organisation_id": "Org One"}, 422),
|
||||
({"organisation_id": "Test Org"}, 422),
|
||||
({"organisation_id": ""}, 422),
|
||||
({}, 422),
|
||||
({"organisation_id": "1", "status": True}, 422),
|
||||
|
|
@ -217,7 +229,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
|
|||
|
||||
assert "users" in data
|
||||
assert isinstance(data["users"], list)
|
||||
assert len(data["users"]) == 2
|
||||
assert len(data["users"]) == 1
|
||||
|
||||
user = data["users"][0]
|
||||
assert isinstance(user, dict)
|
||||
|
|
@ -225,7 +237,7 @@ async def test_get_org_users_success(default_client: AsyncClient):
|
|||
assert user["id"] == 1
|
||||
|
||||
assert "organisation" in data
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
assert data["organisation"]["id"] == 1
|
||||
|
||||
|
||||
|
|
@ -242,9 +254,19 @@ async def test_get_org_users_status_checks(
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_org_user_success(default_client: AsyncClient):
|
||||
async def test_post_org_user_success(default_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.post(
|
||||
"/org/user", json={"organisation_id": 1, "user_id": 3}
|
||||
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
|
|
@ -254,12 +276,12 @@ async def test_post_org_user_success(default_client: AsyncClient):
|
|||
assert "organisation" in data
|
||||
assert isinstance(data["organisation"], dict)
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
|
||||
assert "users" in data
|
||||
assert isinstance(data["users"], list)
|
||||
assert (
|
||||
len([user for user in data["users"] if user["email"] == "root@orgtwo.com"]) == 1
|
||||
len([user for user in data["users"] if user["email"] == "user@test.org"]) == 1
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -276,15 +298,37 @@ async def test_post_org_user_success(default_client: AsyncClient):
|
|||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_post_org_user_status_checks(
|
||||
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
||||
default_client: AsyncClient, body: dict[str, str], expected_status: int, db_session
|
||||
):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.post("/org/user", json=body)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_patch_org_root_user_success(default_client: AsyncClient):
|
||||
async def test_patch_org_root_user_success(default_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
|
|
@ -292,15 +336,15 @@ async def test_patch_org_root_user_success(default_client: AsyncClient):
|
|||
|
||||
data = resp.json()
|
||||
|
||||
assert data["name"] == "Org One"
|
||||
assert data["root_user_email"] == "user@orgone.com"
|
||||
assert data["name"] == "Test Org"
|
||||
assert data["root_user_email"] == "user@test.org"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status",
|
||||
[
|
||||
({"organisation_id": 42, "user_id": 2}, 404),
|
||||
({"organisation_id": "Org One", "user_id": 2}, 422),
|
||||
({"organisation_id": "Test Org", "user_id": 2}, 422),
|
||||
({"organisation_id": "", "user_id": 2}, 422),
|
||||
({}, 422),
|
||||
({"user_id": 2}, 422),
|
||||
|
|
@ -310,17 +354,39 @@ async def test_patch_org_root_user_success(default_client: AsyncClient):
|
|||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_patch_root_user_status_checks(
|
||||
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
||||
default_client: AsyncClient, body: dict[str, str], expected_status: int, db_session
|
||||
):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
db_session.add(OrgUsers(org_id=1, user_id=2))
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.patch("/org/root_user", json=body)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_patch_org_root_user_non_member(default_client: AsyncClient):
|
||||
async def test_patch_org_root_user_non_member(default_client: AsyncClient, db_session):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
|
||||
resp = await default_client.patch(
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 3}
|
||||
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
||||
)
|
||||
data = resp.json()
|
||||
|
||||
|
|
@ -338,14 +404,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"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
|
||||
assert "groups" in data
|
||||
assert isinstance(data["groups"], list)
|
||||
group = data["groups"][0]
|
||||
assert isinstance(group, dict)
|
||||
assert group["id"] == 1
|
||||
assert group["name"] == "Org One Group"
|
||||
assert group["name"] == "Test Group"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -372,7 +438,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"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
|
||||
attributes = [
|
||||
"email",
|
||||
|
|
@ -450,7 +516,7 @@ async def test_patch_org_contact_success(
|
|||
|
||||
assert "organisation" in data
|
||||
assert data["organisation"]["id"] == 1
|
||||
assert data["organisation"]["name"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
|
||||
attributes = [
|
||||
"email",
|
||||
|
|
@ -491,7 +557,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": "Org One", "contact_type": "billing"}, 422),
|
||||
({"organisation_id": "Test Org", "contact_type": "billing"}, 422),
|
||||
({"organisation_id": "", "contact_type": "billing"}, 422),
|
||||
({}, 422),
|
||||
({"organisation_id": 1, "contact_type": "not_real"}, 422),
|
||||
|
|
@ -516,14 +582,26 @@ async def test_delete_org_success(default_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_org_users_success(default_client: AsyncClient):
|
||||
async def test_delete_org_users_success(db_session, default_client: AsyncClient):
|
||||
db_session.add(
|
||||
User(
|
||||
email="user@test.org",
|
||||
first_name="User",
|
||||
last_name="Test",
|
||||
oidc_id="abcd-efgh-ijkl-1234",
|
||||
)
|
||||
)
|
||||
db_session.flush()
|
||||
resp = await default_client.delete("/org/user?org_id=1&user_id=2")
|
||||
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_delete_preapproval_org_success(default_client: AsyncClient):
|
||||
resp = await default_client.delete("/org/self?org_id=3")
|
||||
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")
|
||||
|
||||
assert resp.status_code == 204
|
||||
|
|
|
|||
|
|
@ -5,7 +5,8 @@
|
|||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from .conftest import generate_query_and_status, generate_body_and_status
|
||||
from .conftest import generate_query_and_status
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.service_module,
|
||||
|
|
@ -50,7 +51,12 @@ async def test_post_service_success(default_client: AsyncClient):
|
|||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, expected_status", generate_body_and_status({"name": "str"})
|
||||
"body, expected_status",
|
||||
[
|
||||
({"name": "Test Service"}, 409),
|
||||
({"name": 42}, 422),
|
||||
({}, 422),
|
||||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_post_service_status_checks(
|
||||
|
|
@ -61,13 +67,6 @@ 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})
|
||||
|
|
@ -83,7 +82,12 @@ async def test_patch_service_success(default_client: AsyncClient):
|
|||
|
||||
@pytest.mark.parametrize(
|
||||
"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
|
||||
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"] == "Org One"
|
||||
assert data["organisation"]["name"] == "Test Org"
|
||||
|
||||
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"] == "Org One"
|
||||
assert org["name"] == "Test Org"
|
||||
assert org["status"] == "approved"
|
||||
assert org["root_user_email"] == "admin@test.com"
|
||||
assert "intake_questionnaire" in org
|
||||
assert isinstance(org["intake_questionnaire"], dict)
|
||||
|
||||
assert isinstance(org["billing_contact"], dict)
|
||||
assert org["billing_contact"]["email"] == "billing@orgone.com"
|
||||
assert org["billing_contact"]["email"] == "billing@test.org"
|
||||
assert org["billing_contact"]["id"] == 1
|
||||
|
||||
assert isinstance(org["owner_contact"], dict)
|
||||
assert org["owner_contact"]["email"] == "owner@orgone.com"
|
||||
assert org["owner_contact"]["email"] == "owner@test.org"
|
||||
assert org["owner_contact"]["id"] == 2
|
||||
|
||||
assert isinstance(org["security_contact"], dict)
|
||||
assert org["security_contact"]["email"] == "security@orgone.com"
|
||||
assert org["security_contact"]["email"] == "security@test.org"
|
||||
assert org["security_contact"]["id"] == 3
|
||||
|
||||
|
||||
|
|
@ -162,12 +162,12 @@ async def test_get_self_orgs_dynamic(default_client: AsyncClient):
|
|||
"organisations": [
|
||||
{
|
||||
"organisation_id": 1,
|
||||
"name": "Org One",
|
||||
"name": "Test Org",
|
||||
"status": "approved",
|
||||
"root_user_email": "admin@test.com",
|
||||
"owner_contact": {"email": "owner@orgone.com", "id": 2},
|
||||
"security_contact": {"email": "security@orgone.com", "id": 3},
|
||||
"billing_contact": {"email": "billing@orgone.com", "id": 1},
|
||||
"owner_contact": {"email": "owner@test.org", "id": 2},
|
||||
"security_contact": {"email": "security@test.org", "id": 3},
|
||||
"billing_contact": {"email": "billing@test.org", "id": 1},
|
||||
"intake_questionnaire": {
|
||||
"questions": {
|
||||
"question_one": None,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue