Compare commits
No commits in common. "d3400f53e43e1f47b0c31a60a91884efbe376271" and "b8b6e6c7eeb4bbf7844dedbbf1a34c82890c5419" have entirely different histories.
d3400f53e4
...
b8b6e6c7ee
3 changed files with 44 additions and 407 deletions
|
|
@ -234,7 +234,7 @@ async def update_root_user(db: db_dependency, org_model: org_model_body_dependen
|
||||||
raise UnauthorizedException(message="This user does not belong to your organisation.")
|
raise UnauthorizedException(message="This user does not belong to your organisation.")
|
||||||
org_model.root_user_rel = user_model
|
org_model.root_user_rel = user_model
|
||||||
db.flush()
|
db.flush()
|
||||||
response = OrgPatchRootResponse(name=org_model.name, root_user_email=org_model.root_user_email)
|
response = OrgPatchRootResponse(**org_model.__dict__)
|
||||||
db.commit()
|
db.commit()
|
||||||
return response
|
return response
|
||||||
|
|
||||||
|
|
@ -337,8 +337,6 @@ async def update_contact(db: db_dependency, org_model: org_model_root_claim_body
|
||||||
if hasattr(contact_model, key):
|
if hasattr(contact_model, key):
|
||||||
setattr(contact_model, key, value)
|
setattr(contact_model, key, value)
|
||||||
else:
|
else:
|
||||||
if key == "contact_type" or key == "organisation_id":
|
|
||||||
continue
|
|
||||||
raise UnprocessableContentException("Invalid keys in update request")
|
raise UnprocessableContentException("Invalid keys in update request")
|
||||||
db.flush()
|
db.flush()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -20,14 +20,13 @@ SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
def db_session():
|
def db_session():
|
||||||
Base.metadata.drop_all(bind=engine)
|
|
||||||
Base.metadata.create_all(bind=engine)
|
|
||||||
db = SessionLocal()
|
db = SessionLocal()
|
||||||
try:
|
try:
|
||||||
_seed(db) # extracted seeding logic into a plain function
|
|
||||||
yield db
|
yield db
|
||||||
finally:
|
except:
|
||||||
db.rollback()
|
db.rollback()
|
||||||
|
raise
|
||||||
|
finally:
|
||||||
db.close()
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -44,14 +43,24 @@ async def client(db_session) -> AsyncGenerator[AsyncClient, None]:
|
||||||
app.dependency_overrides.clear()
|
app.dependency_overrides.clear()
|
||||||
|
|
||||||
|
|
||||||
def _seed(db):
|
@pytest.fixture(scope="session")
|
||||||
|
def setup_database():
|
||||||
|
Base.metadata.create_all(bind=engine)
|
||||||
|
yield
|
||||||
|
Base.metadata.drop_all(bind=engine)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def seed_db():
|
||||||
|
db = SessionLocal()
|
||||||
|
try:
|
||||||
db.add(User(email="admin@test.com", first_name="Admin", last_name="Test", oidc_id="abcd-efgh-ijkl-mnop"))
|
db.add(User(email="admin@test.com", first_name="Admin", last_name="Test", oidc_id="abcd-efgh-ijkl-mnop"))
|
||||||
db.add(Contact(org_id=1, email="billing@test.org", phonenumber="07521539927"))
|
db.add(Contact(org_id=1))
|
||||||
db.add(Contact(org_id=1, email="owner@test.org", phonenumber="07521539927"))
|
db.add(Contact(org_id=1))
|
||||||
db.add(Contact(org_id=1, email="security@test.org", phonenumber="07521539927"))
|
db.add(Contact(org_id=1))
|
||||||
db.flush()
|
db.flush()
|
||||||
db.add(Org(name="Test Org", root_user_id=1, billing_contact_id=1, owner_contact_id=2, security_contact_id=3,
|
db.add(Org(name="Test Org", root_user_id=1, billing_contact_id=1, owner_contact_id=1, security_contact_id=1,
|
||||||
status="approved", intake_questionnaire={"question_two": "answer two"}))
|
status="approved", intake_questionnaire="{}"))
|
||||||
db.add(Service(name="Test Service", api_key="123456789"))
|
db.add(Service(name="Test Service", api_key="123456789"))
|
||||||
db.add(Permission(service_id=1, resource="test_resource", action="read"))
|
db.add(Permission(service_id=1, resource="test_resource", action="read"))
|
||||||
db.add(Group(name="Test Group"))
|
db.add(Group(name="Test Group"))
|
||||||
|
|
@ -62,7 +71,15 @@ def _seed(db):
|
||||||
user_model = db.get(User, 1)
|
user_model = db.get(User, 1)
|
||||||
org_model = db.get(Org, 1)
|
org_model = db.get(Org, 1)
|
||||||
org_model.user_rel.append(user_model)
|
org_model.user_rel.append(user_model)
|
||||||
org_model.group_rel.append(group_model)
|
|
||||||
db.flush()
|
db.flush()
|
||||||
group_model.user_rel.append(user_model)
|
group_model.user_rel.append(user_model)
|
||||||
db.commit()
|
db.commit()
|
||||||
|
yield db
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session", autouse=True)
|
||||||
|
def seed_data(setup_database, seed_db):
|
||||||
|
yield
|
||||||
|
|
||||||
|
|
@ -1,378 +0,0 @@
|
||||||
"""
|
|
||||||
[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 client
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_get_org_success(client: AsyncClient):
|
|
||||||
resp = await client.get("/org/id?org_id=1")
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
assert data["name"] == "Test Org"
|
|
||||||
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 data["status"] == "approved"
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"query, expected_status",
|
|
||||||
[
|
|
||||||
("org_id=2", 404),
|
|
||||||
("org_id=banana", 422),
|
|
||||||
("", 422),
|
|
||||||
],
|
|
||||||
)
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_get_org_failure(client: AsyncClient, query: str, expected_status: int):
|
|
||||||
resp = await client.get(f"/org/id?{query}")
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_post_org_success(client: AsyncClient):
|
|
||||||
resp = await 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": 42}, 422),
|
|
||||||
({}, 422),
|
|
||||||
({"name": "New Test Org", "intake_questionnaire": {"question_one": 42}}, 422),
|
|
||||||
],
|
|
||||||
)
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_post_org_failure(client: AsyncClient, body: dict[str, str], expected_status: int):
|
|
||||||
resp = await client.post("/org/", json=body)
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_patch_org_questionnaire_partial_success(client: AsyncClient, db_session):
|
|
||||||
org_model = db_session.get(Organisation, 1)
|
|
||||||
org_model.status = "partial"
|
|
||||||
db_session.flush()
|
|
||||||
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})
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
assert "intake_questionnaire" in data
|
|
||||||
assert data["name"] == "Test Org"
|
|
||||||
assert data["intake_questionnaire"]["question_one"] == "new answer one"
|
|
||||||
assert data["status"] == "partial"
|
|
||||||
# assert type(data["intake_questionnaire"]["question_two"]) == str
|
|
||||||
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_failure(client: AsyncClient, body: dict[str, str], expected_status: int):
|
|
||||||
resp = await client.patch("/org/questionnaire", json=body)
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_patch_org_questionnaire_submit_success(client: AsyncClient, db_session):
|
|
||||||
org_model = db_session.get(Organisation, 1)
|
|
||||||
org_model.status = "partial"
|
|
||||||
db_session.flush()
|
|
||||||
resp = await 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 "intake_questionnaire" in data
|
|
||||||
assert data["name"] == "Test Org"
|
|
||||||
assert data["intake_questionnaire"]["question_one"] == "new answer one"
|
|
||||||
assert data["status"] == "submitted"
|
|
||||||
# assert type(data["intake_questionnaire"]["question_two"]) == str
|
|
||||||
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(client: AsyncClient, status: str):
|
|
||||||
resp = await 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_failure(client: AsyncClient, body: dict[str, str], expected_status: int):
|
|
||||||
resp = await client.patch("/org/status", json=body)
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_get_org_users_success(client: AsyncClient):
|
|
||||||
resp = await client.get("/org/users?org_id=1")
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
assert "users" in data
|
|
||||||
assert type(data["users"]) == list
|
|
||||||
assert len(data["users"]) == 1
|
|
||||||
assert data["users"][0] == "admin@test.com"
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"query, expected_status",
|
|
||||||
[
|
|
||||||
("org_id=2", 404),
|
|
||||||
("org_id=banana", 422),
|
|
||||||
("", 422),
|
|
||||||
],
|
|
||||||
)
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_get_org_users_failure(client: AsyncClient, query: str, expected_status: int):
|
|
||||||
resp = await client.get(f"/org/users?{query}")
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_post_org_user_success(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})
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
assert "users" in data
|
|
||||||
assert type(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),
|
|
||||||
],
|
|
||||||
)
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_post_org_failure(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 client.post("/org/user", json=body)
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_patch_org_root_user_success(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})
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
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_failure(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 client.patch("/org/root_user", json=body)
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_get_org_groups_success(client: AsyncClient):
|
|
||||||
resp = await client.get("/org/groups?org_id=1")
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
assert "groups" in data
|
|
||||||
assert type(data["groups"]) == list
|
|
||||||
assert "Test Group" in data["groups"]
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"query, expected_status",
|
|
||||||
[
|
|
||||||
("org_id=2", 404),
|
|
||||||
("org_id=banana", 422),
|
|
||||||
("", 422),
|
|
||||||
],
|
|
||||||
)
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_get_org_groups_failure(client: AsyncClient, query: str, expected_status: int):
|
|
||||||
resp = await 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(client: AsyncClient, contact_type: str):
|
|
||||||
resp = await client.get(f"/org/contact?org_id=1&contact_type={contact_type}")
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
|
|
||||||
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_failure(client: AsyncClient, query: str, expected_status: int):
|
|
||||||
resp = await 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(client: AsyncClient, key: str, value: str):
|
|
||||||
resp = await client.patch("/org/contact", json={"organisation_id": 1, "contact_type": "billing", key: value})
|
|
||||||
data = resp.json()
|
|
||||||
|
|
||||||
assert resp.status_code == 200
|
|
||||||
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": "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_status_failure(client: AsyncClient, body: dict[str, str], expected_status: int):
|
|
||||||
resp = await client.patch("/org/contact", json=body)
|
|
||||||
|
|
||||||
assert resp.status_code == expected_status
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue