tests: iam router
This commit is contained in:
parent
65a9514be6
commit
806bbfcbfc
1 changed files with 414 additions and 0 deletions
414
test/test_iam.py
Normal file
414
test/test_iam.py
Normal file
|
|
@ -0,0 +1,414 @@
|
|||
"""
|
||||
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 client, db_session
|
||||
|
||||
from src.iam.models import Group
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_act_on_resource_endpoint_success(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 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(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 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(client: AsyncClient):
|
||||
resp = await 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(client: AsyncClient, query: str, expected_status: int):
|
||||
resp = await client.get(f"/iam/group/permissions?{query}")
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_group_users_success(client: AsyncClient):
|
||||
resp = await 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(client: AsyncClient, query: str, expected_status: int):
|
||||
resp = await client.get(f"/iam/group/users?{query}")
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_group_success(client: AsyncClient):
|
||||
resp = await 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(client: AsyncClient, body: dict[str, str], expected_status: int):
|
||||
resp = await client.post("/iam/group", json=body)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_group_perm_success(client: AsyncClient, db_session):
|
||||
db_session.add(Group(name="Test Group Two", org_id=1))
|
||||
db_session.flush()
|
||||
resp = await 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(client: AsyncClient, body: dict[str, str], expected_status: int):
|
||||
resp = await client.put("/iam/group/permission", json=body)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_put_group_user_success(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 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(client: AsyncClient, body: dict[str, str], expected_status: int):
|
||||
resp = await client.put("/iam/group/user", json=body)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_permissions_success(client: AsyncClient):
|
||||
resp = await 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(client: AsyncClient, query: str, expected_status: int):
|
||||
resp = await client.get(f"/iam/permissions?{query}")
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_post_perm_success(client: AsyncClient, db_session):
|
||||
resp = await 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(client: AsyncClient, body: dict[str, str], expected_status: int):
|
||||
resp = await 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(client: AsyncClient, db_session, body):
|
||||
resp = await 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(client: AsyncClient, body: dict[str, str], expected_status: int):
|
||||
resp = await client.post("/iam/permissions/search", json=body)
|
||||
|
||||
assert resp.status_code == expected_status
|
||||
Loading…
Add table
Add a link
Reference in a new issue