""" Act on resource tests only check for pass/fail on input validation. Logic is not tested. """ import pytest from httpx import AsyncClient from src.user.models import User from .conftest import default_client, db_session from src.iam.models import Group @pytest.mark.anyio async def test_post_act_on_resource_endpoint_success(default_client: AsyncClient): body = { "service": "Test Service", "organisation": "Test Org", "resource": "test_resource" } headers = { "Authorization": "Bearer not_checked_when_auth_is_disabled", "X-API-Key": "123456789" } resp = await default_client.post("/iam/can_act_on_resource?action=read", json=body, headers=headers) data = resp.json() assert resp.status_code == 200 assert data == True @pytest.mark.anyio @pytest.mark.parametrize( "service, org, resource, action, expected_status", [ (None, "Test Org", "test_resource", "read", 422), (42, "Test Org", "test_resource", "read", 422), ("Test Service", None, "test_resource", "read", 422), ("Test Service", 42, "test_resource", "read", 422), ("Test Service", "Test Org", None, "read", 422), ("Test Service", "Test Org", 42, "read", 422), ], ) @pytest.mark.anyio async def test_post_act_on_resource_endpoint_failure(default_client: AsyncClient, service, org, resource, action, expected_status: int): 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(f"/iam/can_act_on_resource?action={action}", json=body, headers=headers) assert resp.status_code == expected_status @pytest.mark.anyio async def test_get_group_permissions_success(default_client: AsyncClient): resp = await default_client.get("/iam/group/permissions?org_id=1&group_id=1") data = resp.json() assert resp.status_code == 200 assert "permissions" in data assert type(data["permissions"]) == list assert data["permissions"][0]["service_name"] == "Test Service" assert data["permissions"][0]["resource"] == "test_resource" assert data["permissions"][0]["action"] == "read" @pytest.mark.parametrize( "query, expected_status", [ ("org_id=2&group_id=1", 404), # Non-exists org, valid group ("org_id=banana&group_id=1", 422), # Invalid org, valid group ("org_id=&group_id=1", 422), # Blank org, valid group ("org_id=-1&group_id=1", 422), # Negative org, valid group ("group_id=1", 422), # Only group ("", 422), # Blank query ("org_id=&group_id=", 422), # Both blank ("org_id=1&group_id=2", 404), # Valid org, non-exists group ("org_id=1&group_id=banana", 422), # Valid org, invalid group ("org_id=1&group_id=", 422), # Valid org, blank group ("org_id=1&group_id=-1", 422), # Valid org, negative group ("org_id=1", 422), # Only org ], ) @pytest.mark.anyio async def test_get_group_permissions_failure(default_client: AsyncClient, query: str, expected_status: int): resp = await default_client.get(f"/iam/group/permissions?{query}") assert resp.status_code == expected_status @pytest.mark.anyio async def test_get_group_users_success(default_client: AsyncClient): resp = await default_client.get("/iam/group/users?org_id=1&group_id=1") data = resp.json() assert resp.status_code == 200 assert "users" in data assert type(data["users"]) == list assert data["users"][0]["id"] == 1 assert data["users"][0]["first_name"] == "Admin" assert data["users"][0]["last_name"] == "Test" assert data["users"][0]["email"] == "admin@test.com" @pytest.mark.parametrize( "query, expected_status", [ ("org_id=2&group_id=1", 404), # Non-exists org, valid group ("org_id=banana&group_id=1", 422), # Invalid org, valid group ("org_id=&group_id=1", 422), # Blank org, valid group ("org_id=-1&group_id=1", 422), # Negative org, valid group ("group_id=1", 422), # Only group ("", 422), # Blank query ("org_id=&group_id=", 422), # Both blank ("org_id=1&group_id=2", 404), # Valid org, non-exists group ("org_id=1&group_id=banana", 422), # Valid org, invalid group ("org_id=1&group_id=", 422), # Valid org, blank group ("org_id=1&group_id=-1", 422), # Valid org, negative group ("org_id=1", 422), # Only org ], ) @pytest.mark.anyio async def test_get_group_users_failure(default_client: AsyncClient, query: str, expected_status: int): resp = await default_client.get(f"/iam/group/users?{query}") assert resp.status_code == expected_status @pytest.mark.anyio async def test_post_group_success(default_client: AsyncClient): resp = await default_client.post("/iam/group", json={"name": "New Group", "organisation_id": 1}) data = resp.json() assert resp.status_code == 200 assert "group" in data assert type(data["group"]) == dict assert data["group"]["name"] == "New Group" assert data["group"]["id"] == 2 @pytest.mark.parametrize( "body, expected_status", [ ({"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_failure(default_client: AsyncClient, body: dict[str, str], expected_status: int): resp = await default_client.post("/iam/group", json=body) assert resp.status_code == expected_status @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() resp = await default_client.put("/iam/group/permission", json={"permission_id": 1, "group_id": 2, "organisation_id": 1}) data = resp.json() assert resp.status_code == 200 assert "group" in data assert type(data["group"]) == dict assert data["group"]["name"] == "Test Group Two" assert data["group"]["id"] == 2 assert "permissions" in data assert type(data["permissions"]) == list assert data["permissions"][0]["service_name"] == "Test Service" assert data["permissions"][0]["resource"] == "test_resource" assert data["permissions"][0]["action"] == "read" @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 ], ) @pytest.mark.anyio async def test_put_group_perm_failure(default_client: AsyncClient, body: dict[str, str], expected_status: int): resp = await default_client.put("/iam/group/permission", json=body) assert resp.status_code == expected_status @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() resp = await default_client.put("/iam/group/user", json={"user_id": 2, "group_id": 1, "organisation_id": 1}) data = resp.json() assert resp.status_code == 200 assert "group" in data assert type(data["group"]) == dict assert data["group"]["name"] == "Test Group" assert data["group"]["id"] == 1 assert "users" in data assert type(data["users"]) == list assert data["users"][1]["id"] == 2 assert data["users"][1]["first_name"] == "User" assert data["users"][1]["last_name"] == "Test" assert data["users"][1]["email"] == "user@test.org" @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 ], ) @pytest.mark.anyio async def test_put_group_user_failure(default_client: AsyncClient, body: dict[str, str], expected_status: int): resp = await default_client.put("/iam/group/user", json=body) assert resp.status_code == expected_status @pytest.mark.anyio async def test_get_permissions_success(default_client: AsyncClient): resp = await default_client.get("/iam/permissions?org_id=1") data = resp.json() assert resp.status_code == 200 assert "permissions" in data assert type(data["permissions"]) == list assert data["permissions"][0]["service_name"] == "Test Service" assert data["permissions"][0]["resource"] == "test_resource" assert data["permissions"][0]["action"] == "read" @pytest.mark.parametrize( "query, expected_status", [ ("org_id=42", 404), # Non-exists org ("org_id=banana", 422), # Invalid org ("org_id=", 422), # Blank org ("org_id=-1", 422), # Negative org ("", 422), # Blank query ], ) @pytest.mark.anyio async def test_get_permissions_failure(default_client: AsyncClient, query: str, expected_status: int): resp = await default_client.get(f"/iam/permissions?{query}") assert resp.status_code == expected_status @pytest.mark.anyio 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"}) data = resp.json() assert resp.status_code == 200 assert "permission" in data assert data["permission"]["service_name"] == "Test Service" assert data["permission"]["resource"] == "test_resource" assert data["permission"]["action"] == "create" @pytest.mark.parametrize( "body, expected_status", [ # service_id tests ({"service_id": 42, "resource": "test_resource", "action": "read"}, 404), # Non-existent service ({"service_id": "banana", "resource": "test_resource", "action": "read"}, 422), # Invalid service ID ({"service_id": "", "resource": "test_resource", "action": "read"}, 422), # Blank service ID ({"service_id": -1, "resource": "test_resource", "action": "read"}, 422), # Negative service ID # resource tests ({"service_id": 1, "resource": 42, "action": "read"}, 422), # Invalid resource type # action tests ({"service_id": 1, "resource": "test_resource", "action": 42}, 422), # Invalid action type # missing/partial body tests ({}, 422), # Blank body ({"resource": "test_resource"}, 422), # Only resource ({"action": "read"}, 422), # Only action ({"service_id": 1}, 422), # Only service ({"service_id": 1, "action": "read"}, 422), # Missing resource ({"service_id": 1, "resource": "test_resource"}, 422), # Missing action ({"resource": "test_resource", "action": "read"}, 422), # Missing service ], ) @pytest.mark.anyio async def test_post_perm_failure(default_client: AsyncClient, body: dict[str, str], expected_status: int): resp = await default_client.post("/iam/permission", json=body) assert resp.status_code == expected_status @pytest.mark.parametrize( "body", [ {"organisation_id": 1, "service_id": 1, "resource": "test_resource", "action": "read"}, {"organisation_id": 1, "service_id": 1}, {"organisation_id": 1, "resource": "test_resource"}, {"organisation_id": 1, "action": "read"}, {"organisation_id": 1, "service_id": 1, "action": "read"}, {"organisation_id": 1, "service_id": 1, "resource": "test_resource"}, {"organisation_id": 1, "resource": "test_resource", "action": "read"}, ], ) @pytest.mark.anyio 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() assert resp.status_code == 200 assert "permissions" in data assert type(data["permissions"]) == list assert data["permissions"][0]["service_name"] == "Test Service" assert data["permissions"][0]["resource"] == "test_resource" assert data["permissions"][0]["action"] == "read" @pytest.mark.parametrize( "body, expected_status", [ # organisation_id tests ({"organisation_id": 42, "service_id": 1, "resource": "test_resource", "action": "read"}, 404), # Non-existent organisation ({"organisation_id": "banana", "service_id": 1, "resource": "test_resource", "action": "read"}, 422), # Invalid organisation ID ({"organisation_id": "", "service_id": 1, "resource": "test_resource", "action": "read"}, 422), # Blank organisation ID ({"organisation_id": -1, "service_id": 1, "resource": "test_resource", "action": "read"}, 422), # Negative organisation ID # service_id tests ({"organisation_id": 1, "service_id": "banana", "resource": "test_resource", "action": "read"}, 422), # Invalid service ID ({"organisation_id": 1, "service_id": "", "resource": "test_resource", "action": "read"}, 422), # Blank service ID ({"organisation_id": 1, "service_id": -1, "resource": "test_resource", "action": "read"}, 422), # Negative service ID # resource tests ({"organisation_id": 1, "service_id": 1, "resource": 42, "action": "read"}, 422), # Invalid resource type # action tests ({"organisation_id": 1, "service_id": 1, "resource": "test_resource", "action": 42}, 422), # Invalid action type # missing/partial body tests ({}, 422), # Blank body ], ) @pytest.mark.anyio async def test_post_perm_search_failure(default_client: AsyncClient, body: dict[str, str], expected_status: int): resp = await default_client.post("/iam/permissions/search", json=body) assert resp.status_code == expected_status