cloud-api/test/test_user.py

"""
[GET]/user/self/claims is not tested because it requires OIDC authentication.
[DELETE]/user/ is not tested because the testing client cannot attach a body to a delete request.
"""

from typing import Any

import pytest
from httpx import AsyncClient

from .conftest import generate_query_and_status, standard_test

pytestmark = [
    pytest.mark.user_module,
]


@pytest.mark.anyio
@pytest.mark.parametrize("query, expected_status", generate_query_and_status(["user_id"]))
async def test_get_user_status_checks(
    default_client: AsyncClient, query: str, expected_status: int
):
    resp = await default_client.get(f"/user?{query}")

    assert resp.status_code == expected_status


@pytest.mark.parametrize(
    "body, expected_status",
    [
        ({"organisation_id": 42, "user_email": "admin@test.com"}, 404),
        ({"organisation_id": "Test Org", "user_email": "admin@test.com"}, 422),
        ({"organisation_id": "", "user_email": "admin@test.com"}, 422),
        ({}, 422),
        ({"user_email": 42}, 422),
        ({"organisation_id": 1, "user_email": "Test User"}, 422),
    ],
)
@pytest.mark.anyio
async def test_post_user_invitation_status_checks(
    default_client: AsyncClient, body, expected_status
):
    resp = await default_client.post("/user/invitation", json=body)

    assert resp.status_code == expected_status


@pytest.mark.parametrize(
    "body, expected_status",
    [
        ({"jwt": "invalid"}, 401),
        ({"jwt": ""}, 401),
        ({"jwt": None}, 422),
        ({"jwt": 42}, 422),
    ],
)
@pytest.mark.anyio
async def test_post_user_invitation_accept_status_checks(
    default_client: AsyncClient, body, expected_status
):
    resp = await default_client.post("/user/invitation/accept", json=body)

    assert resp.status_code == expected_status

    if resp.status_code == 401:
        assert resp.json()["detail"] == "Invalid JWS"


@pytest.mark.anyio
async def test_get_self_orgs_standard(
    default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
    method = "GET"
    path = "/user/self/orgs"
    auth_level = "User"
    query = ""
    body = {}
    expected_data = {
        "organisations": [
            {
                "organisation_id": 1,
                "name": "Org One",
                "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},
                "intake_questionnaire": {
                    "questions": {
                        "question_one": None,
                        "question_three": None,
                        "question_two": "answer two",
                    },
                    "metadata": {"version": 0, "submission_date": None},
                },
            }
        ]
    }

    await standard_test(
        default_client, method, path, auth_level, query, body, expected_data, route_data
    )


@pytest.mark.anyio
async def test_get_self_db_standard(
    default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
    method = "GET"
    path = "/user/self/db"
    auth_level = "User"
    query = ""
    body = {}
    expected_data = {
        "id": 1,
        "first_name": "Admin",
        "last_name": "Test",
        "email": "admin@test.com",
        "organisations": [{"name": "Org One", "id": 1}],
        "groups": {"Org One": [{"name": "Org One Group", "id": 1}]},
    }

    await standard_test(
        default_client, method, path, auth_level, query, body, expected_data, route_data
    )


@pytest.mark.anyio
async def test_get_user_standard(
    default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
    method = "GET"
    path = "/user"
    auth_level = "Super Admin"
    query = "?user_id=1"
    body = {}
    expected_data = {
        "id": 1,
        "first_name": "Admin",
        "last_name": "Test",
        "email": "admin@test.com",
        "organisations": [{"name": "Org One", "id": 1}],
        "groups": {"Org One": [{"name": "Org One Group", "id": 1}]},
    }

    await standard_test(
        default_client, method, path, auth_level, query, body, expected_data, route_data
    )


@pytest.mark.anyio
async def test_delete_user_standard(
    default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
    method = "DELETE"
    path = "/user"
    auth_level = "Super Admin"
    query = "?user_id=1"
    body = {}
    expected_data = {}

    await standard_test(
        default_client, method, path, auth_level, query, body, expected_data, route_data
    )


@pytest.mark.anyio
async def test_post_user_invitation_standard(
    default_client: AsyncClient, route_data: dict[str, dict[str, Any]]
):
    method = "POST"
    path = "/user/invitation"
    auth_level = "Root User"
    query = ""
    body = {"user_email": "admin@test.com", "organisation_id": 1}
    expected_data = {
        "invited_email": "admin@test.com",
        "organisation": {"name": "Org One", "id": 1},
    }

    await standard_test(
        default_client, method, path, auth_level, query, body, expected_data, route_data
    )