607 lines
15 KiB
Python
607 lines
15 KiB
Python
"""
|
|
[DELETE] /org/ is not tested because the testing client cannot attach a body to a delete request.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from src.organisation.models import Organisation, OrgUsers
|
|
from src.user.models import User
|
|
from .conftest import generate_query_and_status
|
|
|
|
|
|
pytestmark = [
|
|
pytest.mark.org_module,
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_get_org_success(default_client: AsyncClient):
|
|
resp = await default_client.get("/org?org_id=1")
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 200
|
|
|
|
org = data["organisations"][0]
|
|
|
|
assert isinstance(org, dict)
|
|
assert org["organisation_id"] == 1
|
|
assert org["name"] == "Test Org"
|
|
assert org["status"] == "approved"
|
|
assert org["root_user_email"] == "admin@test.com"
|
|
assert "intake_questionnaire" in org
|
|
assert isinstance(org["intake_questionnaire"], dict)
|
|
|
|
assert org["billing_contact"]["email"] == "billing@test.org"
|
|
assert org["owner_contact"]["email"] == "owner@test.org"
|
|
assert org["security_contact"]["email"] == "security@test.org"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"query, expected_status", generate_query_and_status(["org_id"])
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_get_org_status_checks(
|
|
default_client: AsyncClient, query: str, expected_status: int
|
|
):
|
|
resp = await default_client.get(f"/org?{query}")
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_post_org_success(default_client: AsyncClient):
|
|
resp = await default_client.post("/org", json={"name": "New Test Org"})
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 201
|
|
assert data["name"] == "New Test Org"
|
|
assert data["status"] == "partial"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, expected_status",
|
|
[
|
|
({"name": "Test Org"}, 409),
|
|
({"name": 42}, 422),
|
|
({}, 422),
|
|
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_post_org_status_checks(
|
|
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
|
):
|
|
resp = await default_client.post("/org", json=body)
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_questionnaire_partial_success(
|
|
default_client: AsyncClient, db_session
|
|
):
|
|
org_model = db_session.get(Organisation, 1)
|
|
org_model.status = "partial"
|
|
db_session.flush()
|
|
resp = await default_client.patch(
|
|
"/org/questionnaire",
|
|
json={
|
|
"organisation_id": 1,
|
|
"intake_questionnaire": {
|
|
"question_one": "new answer one",
|
|
"question_two": None,
|
|
"question_three": None,
|
|
},
|
|
"partial": True,
|
|
},
|
|
)
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 200
|
|
assert data["name"] == "Test Org"
|
|
assert data["status"] == "partial"
|
|
assert "intake_questionnaire" in data
|
|
assert isinstance(data["intake_questionnaire"], dict)
|
|
metadata = data["intake_questionnaire"]["metadata"]
|
|
assert metadata["version"] == 0
|
|
assert metadata["submission_date"] is None
|
|
questions = data["intake_questionnaire"]["questions"]
|
|
assert questions["question_one"] == "new answer one"
|
|
assert questions["question_two"] == "answer two"
|
|
assert questions["question_three"] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, expected_status",
|
|
[
|
|
({"organisation_id": 42}, 404),
|
|
({"organisation_id": "Test Org"}, 422),
|
|
({"organisation_id": ""}, 422),
|
|
({}, 422),
|
|
(
|
|
{
|
|
"organisation_id": "1",
|
|
"intake_questionnaire": {"question_one": 42},
|
|
"partial": True,
|
|
},
|
|
422,
|
|
),
|
|
(
|
|
{"organisation_id": "1", "intake_questionnaire": {"question_one": "valid"}},
|
|
422,
|
|
),
|
|
(
|
|
{
|
|
"organisation_id": "1",
|
|
"intake_questionnaire": {"question_one": "valid"},
|
|
"partial": 42,
|
|
},
|
|
422,
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_patch_questionnaire_partial_status_checks(
|
|
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
|
):
|
|
resp = await default_client.patch("/org/questionnaire", json=body)
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_questionnaire_submit_success(
|
|
default_client: AsyncClient, db_session
|
|
):
|
|
org_model = db_session.get(Organisation, 1)
|
|
org_model.status = "partial"
|
|
db_session.flush()
|
|
resp = await default_client.patch(
|
|
"/org/questionnaire",
|
|
json={
|
|
"organisation_id": 1,
|
|
"intake_questionnaire": {
|
|
"question_one": "new answer one",
|
|
"question_two": None,
|
|
"question_three": None,
|
|
},
|
|
"partial": False,
|
|
},
|
|
)
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 200
|
|
assert data["name"] == "Test Org"
|
|
assert data["status"] == "submitted"
|
|
assert "intake_questionnaire" in data
|
|
assert isinstance(data["intake_questionnaire"], dict)
|
|
metadata = data["intake_questionnaire"]["metadata"]
|
|
assert metadata["version"] == 0
|
|
assert metadata["submission_date"] is not None
|
|
questions = data["intake_questionnaire"]["questions"]
|
|
assert questions["question_one"] == "new answer one"
|
|
assert questions["question_two"] == "answer two"
|
|
assert questions["question_three"] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status", ["partial", "submitted", "remediation", "approved", "rejected", "removed"]
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_status_success(default_client: AsyncClient, status: str):
|
|
resp = await default_client.patch(
|
|
"/org/status", json={"organisation_id": 1, "status": status}
|
|
)
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 200
|
|
assert data["name"] == "Test Org"
|
|
assert data["status"] == status
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, expected_status",
|
|
[
|
|
({"organisation_id": 42}, 404),
|
|
({"organisation_id": "Test Org"}, 422),
|
|
({"organisation_id": ""}, 422),
|
|
({}, 422),
|
|
({"organisation_id": "1", "status": True}, 422),
|
|
({"organisation_id": "1", "status": 42}, 422),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_status_status_checks(
|
|
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
|
):
|
|
resp = await default_client.patch("/org/status", json=body)
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_get_org_users_success(default_client: AsyncClient):
|
|
resp = await default_client.get("/org/users?org_id=1")
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 200
|
|
|
|
assert "users" in data
|
|
assert isinstance(data["users"], list)
|
|
assert len(data["users"]) == 1
|
|
|
|
user = data["users"][0]
|
|
assert isinstance(user, dict)
|
|
assert user["email"] == "admin@test.com"
|
|
assert user["id"] == 1
|
|
|
|
assert "organisation" in data
|
|
assert data["organisation"]["name"] == "Test Org"
|
|
assert data["organisation"]["id"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"query, expected_status", generate_query_and_status(["org_id"])
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_get_org_users_status_checks(
|
|
default_client: AsyncClient, query: str, expected_status: int
|
|
):
|
|
resp = await default_client.get(f"/org/users?{query}")
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_post_org_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.post(
|
|
"/org/user", json={"organisation_id": 1, "user_id": 2}
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
|
|
assert "organisation" in data
|
|
assert isinstance(data["organisation"], dict)
|
|
assert data["organisation"]["id"] == 1
|
|
assert data["organisation"]["name"] == "Test Org"
|
|
|
|
assert "users" in data
|
|
assert isinstance(data["users"], list)
|
|
assert (
|
|
len([user for user in data["users"] if user["email"] == "user@test.org"]) == 1
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, expected_status",
|
|
[
|
|
({"organisation_id": 42}, 404),
|
|
({}, 422),
|
|
({"organisation_id": 1, "user_id": "id"}, 422),
|
|
({"user_id": 2}, 422),
|
|
({"organisation_id": 1, "user_id": 42}, 404),
|
|
({"organisation_id": 1, "user_id": 1}, 409),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_post_org_user_status_checks(
|
|
default_client: AsyncClient, body: dict[str, str], expected_status: int, 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.post("/org/user", json=body)
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_root_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()
|
|
db_session.add(OrgUsers(org_id=1, user_id=2))
|
|
db_session.flush()
|
|
|
|
resp = await default_client.patch(
|
|
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
|
|
assert data["name"] == "Test Org"
|
|
assert data["root_user_email"] == "user@test.org"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, expected_status",
|
|
[
|
|
({"organisation_id": 42, "user_id": 2}, 404),
|
|
({"organisation_id": "Test Org", "user_id": 2}, 422),
|
|
({"organisation_id": "", "user_id": 2}, 422),
|
|
({}, 422),
|
|
({"user_id": 2}, 422),
|
|
({"user_id": 42}, 404),
|
|
({"organisation_id": 1, "user_id": "Test User"}, 422),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_patch_root_user_status_checks(
|
|
default_client: AsyncClient, body: dict[str, str], expected_status: int, 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 default_client.patch("/org/root_user", json=body)
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_root_user_non_member(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.patch(
|
|
"/org/root_user", json={"organisation_id": 1, "user_id": 2}
|
|
)
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 422
|
|
assert data["detail"] == "This user does not belong to your organisation."
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_get_org_groups_success(default_client: AsyncClient):
|
|
resp = await default_client.get("/org/groups?org_id=1")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
|
|
assert "organisation" in data
|
|
assert isinstance(data["organisation"], dict)
|
|
assert data["organisation"]["id"] == 1
|
|
assert data["organisation"]["name"] == "Test Org"
|
|
|
|
assert "groups" in data
|
|
assert isinstance(data["groups"], list)
|
|
group = data["groups"][0]
|
|
assert isinstance(group, dict)
|
|
assert group["id"] == 1
|
|
assert group["name"] == "Test Group"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"query, expected_status", generate_query_and_status(["org_id"])
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_get_org_groups_status_checks(
|
|
default_client: AsyncClient, query: str, expected_status: int
|
|
):
|
|
resp = await default_client.get(f"/org/groups?{query}")
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.parametrize("contact_type", ["billing", "security", "owner"])
|
|
@pytest.mark.anyio
|
|
async def test_get_org_contact_success(default_client: AsyncClient, contact_type: str):
|
|
resp = await default_client.get(
|
|
f"/org/contact?org_id=1&contact_type={contact_type}"
|
|
)
|
|
data = resp.json()
|
|
|
|
assert resp.status_code == 200
|
|
|
|
assert "organisation" in data
|
|
assert data["organisation"]["id"] == 1
|
|
assert data["organisation"]["name"] == "Test Org"
|
|
|
|
attributes = [
|
|
"email",
|
|
"first_name",
|
|
"last_name",
|
|
"phonenumber",
|
|
"vat_number",
|
|
"address",
|
|
]
|
|
|
|
for attribute in attributes:
|
|
assert attribute in data["contact"]
|
|
|
|
address_attributes = [
|
|
"post_office_box_number",
|
|
"street_address",
|
|
"street_address_line_2",
|
|
"locality",
|
|
"address_region",
|
|
"country_code",
|
|
"postal_code",
|
|
]
|
|
|
|
for attribute in address_attributes:
|
|
assert attribute in data["contact"]["address"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"query, expected_status",
|
|
[
|
|
("org_id=42&contact_type=billing", 404),
|
|
("org_id=banana&contact_type=billing", 422),
|
|
("", 422),
|
|
("org_id=1&contact_type=contact", 422),
|
|
("contact_type=billing", 422),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_get_org_contact_status_checks(
|
|
default_client: AsyncClient, query: str, expected_status: int
|
|
):
|
|
resp = await default_client.get(f"/org/contact?{query}")
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"key, value",
|
|
[
|
|
("email", "user@example.com"),
|
|
("first_name", "John"),
|
|
("last_name", "Doe"),
|
|
("phonenumber", "+441234567890"),
|
|
("vat_number", "GB123456789"),
|
|
("post_office_box_number", "PO Box 123"),
|
|
("street_address", "123 Example Street"),
|
|
("street_address_line_2", "Suite 4B"),
|
|
("locality", "Glasgow"),
|
|
("address_region", "Glasgow City"),
|
|
("country_code", "GB"),
|
|
("postal_code", "G1 1AA"),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_contact_success(
|
|
default_client: AsyncClient, key: str, value: str
|
|
):
|
|
resp = await default_client.patch(
|
|
"/org/contact",
|
|
json={"organisation_id": 1, "contact_type": "billing", key: value},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
|
|
assert "organisation" in data
|
|
assert data["organisation"]["id"] == 1
|
|
assert data["organisation"]["name"] == "Test Org"
|
|
|
|
attributes = [
|
|
"email",
|
|
"first_name",
|
|
"last_name",
|
|
"phonenumber",
|
|
"vat_number",
|
|
"address",
|
|
]
|
|
|
|
for attribute in attributes:
|
|
assert attribute in data["contact"]
|
|
|
|
address_attributes = [
|
|
"post_office_box_number",
|
|
"street_address",
|
|
"street_address_line_2",
|
|
"locality",
|
|
"address_region",
|
|
"country_code",
|
|
"postal_code",
|
|
]
|
|
|
|
for attribute in address_attributes:
|
|
assert attribute in data["contact"]["address"]
|
|
|
|
if key in data["contact"]:
|
|
assert data["contact"][key] == value
|
|
elif key in data["contact"]["address"]:
|
|
assert data["contact"]["address"][key] == value
|
|
else:
|
|
pytest.fail(f"Invalid contact key: {key}")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, expected_status",
|
|
[
|
|
({"organisation_id": 42, "contact_type": "billing"}, 404),
|
|
({"organisation_id": 1, "contact_type": "security"}, 200),
|
|
({"organisation_id": 1, "contact_type": "owner"}, 200),
|
|
({"organisation_id": "Test Org", "contact_type": "billing"}, 422),
|
|
({"organisation_id": "", "contact_type": "billing"}, 422),
|
|
({}, 422),
|
|
({"organisation_id": 1, "contact_type": "not_real"}, 422),
|
|
({"organisation_id": 1, "contact_type": 42}, 422),
|
|
({"organisation_id": 1, "contact_type": ""}, 422),
|
|
],
|
|
)
|
|
@pytest.mark.anyio
|
|
async def test_patch_org_contact_status_checks(
|
|
default_client: AsyncClient, body: dict[str, str], expected_status: int
|
|
):
|
|
resp = await default_client.patch("/org/contact", json=body)
|
|
|
|
assert resp.status_code == expected_status
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_delete_org_success(default_client: AsyncClient):
|
|
resp = await default_client.delete("/org?org_id=1")
|
|
|
|
assert resp.status_code == 204
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_delete_org_users_success(db_session, default_client: AsyncClient):
|
|
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.delete("/org/user?org_id=1&user_id=2")
|
|
|
|
assert resp.status_code == 204
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_delete_preapproval_org_success(db_session, default_client: AsyncClient):
|
|
org_model = db_session.get(Organisation, 1)
|
|
org_model.status = "partial"
|
|
db_session.flush()
|
|
resp = await default_client.delete("/org/self?org_id=1")
|
|
|
|
assert resp.status_code == 204
|