cloud-api/test/test_auth_approval.py

"""
This test module checks relevant endpoints to ensure only approved orgs get access or, for pre-approval endpoints, that they are not blocked.
Endpoints not checked here are endpoints that do not require an org check.
Delete endpoints are currently skipped because the testing system cannot use bodies in deletes.
"""
import pytest
from httpx import AsyncClient

from .conftest import client

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


@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(client: AsyncClient):
	resp = await client.get("/org?org_id=1")
	assert resp.status_code != 422
	assert resp.status_code == 200


@pytest.mark.anyio
async def test_patch_org_questionnaire_auth_approval(client: AsyncClient):
	resp = await client.patch("/org/questionnaire", json={"organisation_id": 1,
	                                                      "intake_questionnaire": {"question_one": "new answer one",
	                                                                               "question_two": None,
	                                                                               "question_three": None},
	                                                      "partial": True})
	assert resp.status_code != 422
	assert resp.status_code == 200


@pytest.mark.anyio
async def test_patch_org_status_auth_approval(client: AsyncClient):
	resp = await client.patch("/org/status", json={"organisation_id": 1, "status": "submitted"})
	assert resp.status_code != 422
	assert resp.status_code == 200


@pytest.mark.anyio
async def test_get_org_users_auth_approval(client: AsyncClient):
	resp = await 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(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.post("/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(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 client.patch("/org/root_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_get_org_groups_auth_approval(client: AsyncClient):
	resp = await 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(client: AsyncClient):
	resp = await client.get(f"/org/contact?org_id=1&contact_type=billing")
	assert resp.status_code != 422
	assert resp.status_code == 200


@pytest.mark.anyio
async def test_patch_org_contact_auth_approval(client: AsyncClient):
	resp = await client.patch("/org/contact",
	                          json={"organisation_id": 1, "contact_type": "billing", "email": "user@example.com"})
	assert resp.status_code != 422
	assert resp.status_code == 200


@pytest.mark.anyio
async def test_get_service_auth_approval(client: AsyncClient):
	resp = await 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(client: AsyncClient):
	resp = await 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(client: AsyncClient):
	resp = await 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"]


@pytest.mark.anyio
async def test_post_iam_group_auth_approval(client: AsyncClient):
	resp = await client.post("/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(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})
	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(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})
	assert resp.status_code != 422
	assert "has not been approved." in resp.json()["detail"]


@pytest.mark.anyio
async def test_get_iam_permissions_auth_approval(client: AsyncClient):
	resp = await client.get("/iam/permissions?org_id=1")
	assert resp.status_code != 422
	assert "has not been approved." in resp.json()["detail"]


@pytest.mark.anyio
async def test_post_iam_permissions_search_auth_approval(client: AsyncClient):
	resp = await client.post("/iam/permissions/search", json={"organisation_id": 1, "action": "read"})
	assert resp.status_code != 422
	assert "has not been approved." in resp.json()["detail"]