cloud-api/test/test_iam.py

"""
"""
import pytest
from httpx import AsyncClient

from src.user.models import User
from src.organisation.models import Organisation as Org
from src.iam.models import Group

from .conftest import generate_query_and_status


@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 is True


@pytest.mark.parametrize(
	"service, api_key",
	[
		("Test Service", "not_the_correct_key"),
		("Test Service Two", "123456789")
	],
)
@pytest.mark.anyio
async def test_act_on_resource_wrong_key(default_client: AsyncClient, db_session, service: str, api_key: str):
	body = {
		"service": service,
		"organisation": "Test Org",
		"resource": "test_resource"
	}
	headers = {
		"Authorization": "Bearer not_checked_when_auth_is_disabled",
		"X-API-Key": api_key
	}
	resp = await default_client.post("/iam/can_act_on_resource?action=read", json=body, headers=headers)
	data = resp.json()

	assert resp.status_code == 401
	assert data["detail"] == "Invalid API key"


@pytest.mark.anyio
async def test_act_on_resource_missing_key(default_client: AsyncClient):
	body = {
		"service": "Test Service",
		"organisation": "Test Org",
		"resource": "test_resource"
	}
	headers = {
		"Authorization": "Bearer not_checked_when_auth_is_disabled"
	}
	resp = await default_client.post("/iam/can_act_on_resource?action=read", json=body, headers=headers)
	data = resp.json()

	assert resp.status_code == 401
	assert data["detail"] == "Missing API key"


@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_act_on_resource_endpoint_status_checks(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.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),
	],
)
@pytest.mark.anyio
async def test_act_on_resource_logic(default_client: AsyncClient, db_session, service, org, resource, action,
                                                     expected_response: bool):
	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)
	data = resp.json()

	assert resp.status_code == 200
	assert data == expected_response


@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 isinstance(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",
	generate_query_and_status(["group_id", "org_id"])
)
@pytest.mark.anyio
async def test_get_group_permissions_status_checks(default_client: AsyncClient, db_session, query: str, expected_status: int):
	resp = await default_client.get(f"/iam/group/permissions?{query}")

	assert resp.status_code == expected_status


@pytest.mark.parametrize(
	"query",
	[
		"org_id=1&group_id=2",
		"org_id=2&group_id=1",
	],
)
@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()
	resp = await default_client.get(f"/iam/group/permissions?{query}")

	assert resp.status_code == 401
	assert resp.json()["detail"] == "Group does not belong to this organization"


@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 isinstance(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",
	generate_query_and_status(["group_id", "org_id"])
)
@pytest.mark.anyio
async def test_get_group_users_status_checks(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.parametrize(
	"query",
	[
		"org_id=1&group_id=2",
		"org_id=2&group_id=1",
	],
)
@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()
	resp = await default_client.get(f"/iam/group/users?{query}")

	assert resp.status_code == 401
	assert resp.json()["detail"] == "Group does not belong to this organization"


@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 isinstance(data["group"], dict)
	assert data["group"]["name"] == "New Group"
	assert data["group"]["id"] == 2


@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
	],
)
@pytest.mark.anyio
async def test_post_group_status_checks(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 isinstance(data["group"], dict)
	assert data["group"]["name"] == "Test Group Two"
	assert data["group"]["id"] == 2

	assert "permissions" in data
	assert isinstance(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

		({"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(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.parametrize(
	"body",
	[
		{"organisation_id": 1, "group_id": 2, "permission_id": 1},
		{"organisation_id": 2, "group_id": 1, "permission_id": 1},
	],
)
@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()
	resp = await default_client.put("/iam/group/permission", json=body)

	assert resp.status_code == 401
	assert resp.json()["detail"] == "Group does not belong to this organization"


@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 isinstance(data["group"], dict)
	assert data["group"]["name"] == "Test Group"
	assert data["group"]["id"] == 1

	assert "users" in data
	assert isinstance(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_status_checks(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 isinstance(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",
	generate_query_and_status(["org_id"])
)
@pytest.mark.anyio
async def test_get_permissions_status_checks(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_status_checks(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 isinstance(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_status_checks(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