luxferre
76e889d836
Some checks failed
ci / lint_and_test (push) Failing after 8s
New assertions added for new data being delivered (eg IDs on endpoints not previously serving IDs)
569 lines
14 KiB
Python
569 lines
14 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
|
|
assert data["id"] == 1
|
|
assert data["name"] == "Test Org"
|
|
assert data["status"] == "approved"
|
|
assert data["root_user"] == "admin@test.com"
|
|
assert data["billing_contact"] == "billing@test.org"
|
|
assert data["owner_contact"] == "owner@test.org"
|
|
assert data["security_contact"] == "security@test.org"
|
|
assert "intake_questionnaire" in data
|
|
assert isinstance(data["intake_questionnaire"], dict)
|
|
|
|
|
|
@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)
|
|
assert data["intake_questionnaire"]["question_one"] == "new answer one"
|
|
assert data["intake_questionnaire"]["question_two"] == "answer two"
|
|
assert data["intake_questionnaire"]["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)
|
|
assert data["intake_questionnaire"]["question_one"] == "new answer one"
|
|
assert data["intake_questionnaire"]["question_two"] == "answer two"
|
|
assert data["intake_questionnaire"]["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 "users" in data
|
|
assert isinstance(data["users"], list)
|
|
assert "user@test.org" in data["users"]
|
|
|
|
|
|
@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 "groups" in data
|
|
assert isinstance(data["groups"], list)
|
|
assert "Test Group" in data["groups"]
|
|
|
|
|
|
@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
|