258 lines
8.8 KiB
Python
258 lines
8.8 KiB
Python
import pytest
|
|
|
|
from app.crud import augmented_part as augmented_part_crud
|
|
from app.models.augmented_part import AugmentedPart
|
|
from app.models.imported_part import ImportedPart
|
|
from app.models.manufacturer import Manufacturer
|
|
from app.models.part_colour import PartColour
|
|
from app.models.part_type import PartType
|
|
from app.services import augmented_part_sync
|
|
|
|
|
|
async def _seed_reference_data(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()
|
|
return manufacturer, colour, part_type
|
|
|
|
|
|
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"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_and_count_exclude_archived_by_default(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
manufacturer, colour, part_type = await _seed_reference_data(db)
|
|
active = AugmentedPart(
|
|
description="Active part",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
)
|
|
archived = AugmentedPart(
|
|
description="Archived part",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
archived=True,
|
|
)
|
|
db.add_all([active, archived])
|
|
await db.commit()
|
|
|
|
assert [p.description for p in await augmented_part_crud.list_all(db)] == [
|
|
"Active part"
|
|
]
|
|
assert await augmented_part_crud.count_filtered(db) == 1
|
|
|
|
all_descs = {
|
|
p.description
|
|
for p in await augmented_part_crud.list_all(db, include_archived=True)
|
|
}
|
|
assert all_descs == {"Active part", "Archived part"}
|
|
assert await augmented_part_crud.count_filtered(db, include_archived=True) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_archived_toggles_flag(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
manufacturer, colour, part_type = await _seed_reference_data(db)
|
|
ap = AugmentedPart(
|
|
description="Widget",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
)
|
|
db.add(ap)
|
|
await db.commit()
|
|
|
|
ap = await augmented_part_crud.set_archived(db, ap, True)
|
|
assert ap.archived is True
|
|
|
|
ap = await augmented_part_crud.set_archived(db, ap, False)
|
|
assert ap.archived is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_archives_part_only_when_all_backing_imports_are_gone(
|
|
db_sessionmaker,
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
manufacturer, colour, part_type = await _seed_reference_data(db)
|
|
ip_x = ImportedPart(InventoryID="INV-X", Description="Widget X")
|
|
ip_y = ImportedPart(InventoryID="INV-Y", Description="Widget Y")
|
|
db.add_all([ip_x, ip_y])
|
|
await db.flush()
|
|
|
|
# One AugmentedPart backed by two ImportedParts.
|
|
merged = AugmentedPart(
|
|
description="Merged widget",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
imported_parts=[ip_x, ip_y],
|
|
)
|
|
db.add(merged)
|
|
await db.commit()
|
|
merged_id = merged.id
|
|
|
|
# Delete one backing ImportedPart directly (simulating an external feed
|
|
# change) and sync: the AugmentedPart still has INV-Y, so it survives.
|
|
await db.delete(ip_x)
|
|
await db.commit()
|
|
|
|
result = await augmented_part_sync.sync(db)
|
|
assert result.archived == 0
|
|
|
|
refreshed = await augmented_part_crud.get(db, merged_id)
|
|
assert refreshed.archived is False
|
|
|
|
# Now delete the last remaining backing ImportedPart.
|
|
await db.delete(ip_y)
|
|
await db.commit()
|
|
|
|
result = await augmented_part_sync.sync(db)
|
|
assert result.archived == 1
|
|
assert result.created == 0
|
|
|
|
refreshed = await augmented_part_crud.get(db, merged_id)
|
|
assert refreshed.archived is True
|
|
|
|
# Re-running sync is a no-op — it must not double count or explode.
|
|
result = await augmented_part_sync.sync(db)
|
|
assert result.archived == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_imported_part_via_api_archives_orphaned_augmented_part(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
manufacturer, colour, part_type = await _seed_reference_data(db)
|
|
ip = ImportedPart(InventoryID="INV-SOLE", Description="Solo widget")
|
|
db.add(ip)
|
|
await db.flush()
|
|
ap = AugmentedPart(
|
|
description="Solo widget",
|
|
sku="INV-SOLE",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
imported_parts=[ip],
|
|
)
|
|
db.add(ap)
|
|
await db.commit()
|
|
ap_id = ap.id
|
|
|
|
await make_user("root@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root@example.com')}"}
|
|
|
|
resp = await client.delete("/api/imported-parts/INV-SOLE", headers=headers)
|
|
assert resp.status_code == 204
|
|
|
|
# ImportedPart is really gone.
|
|
assert (
|
|
await client.get("/api/imported-parts/INV-SOLE", headers=headers)
|
|
).status_code == 404
|
|
|
|
# AugmentedPart is archived, not deleted — direct fetch still finds it.
|
|
get_resp = await client.get(f"/api/augmented-parts/{ap_id}", headers=headers)
|
|
assert get_resp.status_code == 200
|
|
assert get_resp.json()["archived"] is True
|
|
|
|
# Hidden from the default list...
|
|
default_list = await client.get("/api/augmented-parts", headers=headers)
|
|
assert ap_id not in [p["id"] for p in default_list.json()]
|
|
|
|
# ...but present when archived parts are explicitly requested.
|
|
full_list = await client.get(
|
|
"/api/augmented-parts?include_archived=true", headers=headers
|
|
)
|
|
assert ap_id in [p["id"] for p in full_list.json()]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_can_unarchive_an_augmented_part(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
manufacturer, colour, part_type = await _seed_reference_data(db)
|
|
ap = AugmentedPart(
|
|
description="Previously archived",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
archived=True,
|
|
)
|
|
db.add(ap)
|
|
await db.commit()
|
|
ap_id = ap.id
|
|
|
|
await make_user("root2@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root2@example.com')}"}
|
|
|
|
default_list = await client.get("/api/augmented-parts", headers=headers)
|
|
assert ap_id not in [p["id"] for p in default_list.json()]
|
|
|
|
resp = await client.patch(
|
|
f"/api/augmented-parts/{ap_id}", json={"archived": False}, headers=headers
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["archived"] is False
|
|
|
|
default_list = await client.get("/api/augmented-parts", headers=headers)
|
|
assert ap_id in [p["id"] for p in default_list.json()]
|
|
|
|
|
|
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_admin_ui_hides_archived_unless_requested_and_can_toggle(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
manufacturer, colour, part_type = await _seed_reference_data(db)
|
|
ap = AugmentedPart(
|
|
description="Hidden gem",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
archived=True,
|
|
)
|
|
db.add(ap)
|
|
await db.commit()
|
|
ap_id = ap.id
|
|
|
|
await make_user("admin@example.com", role_ids=[seeded["roles"]["admin"].id])
|
|
await _web_login(client, "admin@example.com")
|
|
|
|
resp = await client.get("/admin/augmented-parts")
|
|
assert resp.status_code == 200
|
|
assert "Hidden gem" not in resp.text
|
|
|
|
resp = await client.get("/admin/augmented-parts/filter?include_archived=true")
|
|
assert resp.status_code == 200
|
|
assert "Hidden gem" in resp.text
|
|
|
|
resp = await client.post(f"/admin/augmented-parts/{ap_id}/unarchive")
|
|
assert resp.status_code == 200
|
|
assert "Archived" not in resp.text
|
|
|
|
resp = await client.get("/admin/augmented-parts")
|
|
assert "Hidden gem" in resp.text
|
|
|
|
resp = await client.post(f"/admin/augmented-parts/{ap_id}/archive")
|
|
assert resp.status_code == 200
|
|
|
|
resp = await client.get("/admin/augmented-parts")
|
|
assert "Hidden gem" not in resp.text
|