322 lines
9.9 KiB
Python
322 lines
9.9 KiB
Python
"""Duplicate-value pre-checks (409) for every unique-constrained field, across
|
|
both the JSON API (app/api/*.py) and the HTMX admin UI (app/web/admin.py)."""
|
|
|
|
import pytest
|
|
|
|
|
|
async def _token(client, email, password="secret123"):
|
|
resp = await client.post(
|
|
"/api/auth/login", data={"username": email, "password": password}
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
async def _superuser_headers(client, make_user, email):
|
|
await make_user(email, is_superuser=True)
|
|
return {"Authorization": f"Bearer {await _token(client, email)}"}
|
|
|
|
|
|
async def _web_login(client, email, password="secret123"):
|
|
resp = await client.post("/login", data={"email": email, "password": password})
|
|
assert resp.status_code == 303
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_part_type_duplicate_name_409_on_create_and_update(
|
|
client, seeded, make_user
|
|
):
|
|
headers = await _superuser_headers(client, make_user, "pt-api@example.com")
|
|
|
|
resp = await client.post("/api/part-types", json={"name": "Scope"}, headers=headers)
|
|
assert resp.status_code == 201
|
|
other = (
|
|
await client.post("/api/part-types", json={"name": "Cover"}, headers=headers)
|
|
).json()
|
|
|
|
resp = await client.post("/api/part-types", json={"name": "Scope"}, headers=headers)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/part-types/{other['id']}", json={"name": "Scope"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
# Renaming to the resource's own current name is not a false-positive collision.
|
|
resp = await client.patch(
|
|
f"/api/part-types/{other['id']}", json={"name": "Cover"}, headers=headers
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_part_colour_duplicate_name_409_on_create_and_update(
|
|
client, seeded, make_user
|
|
):
|
|
headers = await _superuser_headers(client, make_user, "pc-api@example.com")
|
|
|
|
assert (
|
|
await client.post("/api/part-colours", json={"name": "Black"}, headers=headers)
|
|
).status_code == 201
|
|
other = (
|
|
await client.post("/api/part-colours", json={"name": "Tan"}, headers=headers)
|
|
).json()
|
|
|
|
resp = await client.post(
|
|
"/api/part-colours", json={"name": "Black"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/part-colours/{other['id']}", json={"name": "Black"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_attachment_point_name_duplicate_409_on_create_and_update(
|
|
client, seeded, make_user
|
|
):
|
|
headers = await _superuser_headers(client, make_user, "apn-api@example.com")
|
|
|
|
assert (
|
|
await client.post(
|
|
"/api/attachment-point-names", json={"name": "Ocular"}, headers=headers
|
|
)
|
|
).status_code == 201
|
|
other = (
|
|
await client.post(
|
|
"/api/attachment-point-names", json={"name": "Objective"}, headers=headers
|
|
)
|
|
).json()
|
|
|
|
resp = await client.post(
|
|
"/api/attachment-point-names", json={"name": "Ocular"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/attachment-point-names/{other['id']}",
|
|
json={"name": "Ocular"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_static_connection_duplicate_od_mm_409_on_create_and_update(
|
|
client, seeded, make_user
|
|
):
|
|
headers = await _superuser_headers(client, make_user, "sc-api@example.com")
|
|
|
|
assert (
|
|
await client.post(
|
|
"/api/static-connections",
|
|
json={"od_mm": 40.0, "od_in": 1.575},
|
|
headers=headers,
|
|
)
|
|
).status_code == 201
|
|
other = (
|
|
await client.post(
|
|
"/api/static-connections",
|
|
json={"od_mm": 56.0, "od_in": 2.205},
|
|
headers=headers,
|
|
)
|
|
).json()
|
|
|
|
resp = await client.post(
|
|
"/api/static-connections", json={"od_mm": 40.0, "od_in": 1.6}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/static-connections/{other['id']}",
|
|
json={"od_mm": 40.0},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sub_part_type_duplicate_composite_409_on_create_and_update(
|
|
client, seeded, make_user
|
|
):
|
|
headers = await _superuser_headers(client, make_user, "spt-api@example.com")
|
|
|
|
part_type = (
|
|
await client.post("/api/part-types", json={"name": "Scope"}, headers=headers)
|
|
).json()
|
|
|
|
assert (
|
|
await client.post(
|
|
"/api/sub-part-types",
|
|
json={"name": "FFP", "part_type_id": part_type["id"]},
|
|
headers=headers,
|
|
)
|
|
).status_code == 201
|
|
other = (
|
|
await client.post(
|
|
"/api/sub-part-types",
|
|
json={"name": "SFP", "part_type_id": part_type["id"]},
|
|
headers=headers,
|
|
)
|
|
).json()
|
|
|
|
resp = await client.post(
|
|
"/api/sub-part-types",
|
|
json={"name": "FFP", "part_type_id": part_type["id"]},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/sub-part-types/{other['id']}", json={"name": "FFP"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
# Same name, but a *different* part_type_id, is not a collision.
|
|
other_part_type = (
|
|
await client.post("/api/part-types", json={"name": "Cover"}, headers=headers)
|
|
).json()
|
|
resp = await client.post(
|
|
"/api/sub-part-types",
|
|
json={"name": "FFP", "part_type_id": other_part_type["id"]},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_role_duplicate_name_409_on_create_and_update(client, seeded, make_user):
|
|
headers = await _superuser_headers(client, make_user, "role-api@example.com")
|
|
|
|
assert (
|
|
await client.post("/api/roles", json={"name": "editors"}, headers=headers)
|
|
).status_code == 201
|
|
other = (
|
|
await client.post("/api/roles", json={"name": "viewers"}, headers=headers)
|
|
).json()
|
|
|
|
resp = await client.post("/api/roles", json={"name": "editors"}, headers=headers)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/roles/{other['id']}", json={"name": "editors"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_duplicate_email_409_on_create_and_update(client, seeded, make_user):
|
|
headers = await _superuser_headers(client, make_user, "user-api@example.com")
|
|
|
|
assert (
|
|
await client.post(
|
|
"/api/users",
|
|
json={"email": "dup@example.com", "password": "secret123"},
|
|
headers=headers,
|
|
)
|
|
).status_code == 201
|
|
other = (
|
|
await client.post(
|
|
"/api/users",
|
|
json={"email": "other@example.com", "password": "secret123"},
|
|
headers=headers,
|
|
)
|
|
).json()
|
|
|
|
resp = await client.post(
|
|
"/api/users",
|
|
json={"email": "dup@example.com", "password": "secret123"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.patch(
|
|
f"/api/users/{other['id']}", json={"email": "dup@example.com"}, headers=headers
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_ui_update_part_type_duplicate_name_409(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
from app.crud import part_type as part_type_crud
|
|
|
|
async with db_sessionmaker() as db:
|
|
await part_type_crud.create(db, name="Scope")
|
|
cover = await part_type_crud.create(db, name="Cover")
|
|
cover_id = cover.id
|
|
|
|
await make_user("pt-web@example.com", role_ids=[seeded["roles"]["admin"].id])
|
|
await _web_login(client, "pt-web@example.com")
|
|
|
|
resp = await client.post(f"/admin/part-types/{cover_id}", data={"name": "Scope"})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_ui_augmented_part_duplicate_sku_409_on_create_and_update(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
from app.models.augmented_part import AugmentedPart
|
|
from app.models.manufacturer import Manufacturer
|
|
from app.models.part_colour import PartColour
|
|
from app.models.part_type import PartType
|
|
|
|
async with db_sessionmaker() as db:
|
|
manufacturer = Manufacturer(name="Acme", url="https://acme.test")
|
|
colour = PartColour(name="Black")
|
|
part_type = PartType(name="Scope")
|
|
db.add_all([manufacturer, colour, part_type])
|
|
await db.flush()
|
|
existing = AugmentedPart(
|
|
description="Taken sku part",
|
|
sku="DUP-1",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
)
|
|
other = AugmentedPart(
|
|
description="Other part",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
)
|
|
db.add_all([existing, other])
|
|
await db.commit()
|
|
manufacturer_id, colour_id, part_type_id = (
|
|
manufacturer.id,
|
|
colour.id,
|
|
part_type.id,
|
|
)
|
|
other_id = other.id
|
|
|
|
await make_user("ap-web@example.com", role_ids=[seeded["roles"]["admin"].id])
|
|
await _web_login(client, "ap-web@example.com")
|
|
|
|
resp = await client.post(
|
|
"/admin/augmented-parts",
|
|
data={
|
|
"sku": "DUP-1",
|
|
"description": "New part",
|
|
"colour_id": colour_id,
|
|
"manufacturer_id": manufacturer_id,
|
|
"part_type_id": part_type_id,
|
|
},
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
resp = await client.post(
|
|
f"/admin/augmented-parts/{other_id}",
|
|
data={
|
|
"sku": "DUP-1",
|
|
"description": "Other part",
|
|
"colour_id": colour_id,
|
|
"manufacturer_id": manufacturer_id,
|
|
"part_type_id": part_type_id,
|
|
},
|
|
)
|
|
assert resp.status_code == 409
|