492 lines
17 KiB
Python
492 lines
17 KiB
Python
import pytest
|
||
|
||
from app.models.attachment_point import AttachmentPoint
|
||
from app.models.attachment_point_name import AttachmentPointName
|
||
from app.models.augmented_part import AugmentedPart
|
||
from app.models.fit import Fit, FitLabel
|
||
from app.models.iso_thread import IsoThread
|
||
from app.models.part_type import PartType
|
||
from app.models.scope_info import ScopeInfo
|
||
from app.models.sub_part_type import SubPartType
|
||
from app.services import scope_guide
|
||
|
||
|
||
async def _seed_guide_data(db):
|
||
scope_t = PartType(name="Scope")
|
||
cover_t = PartType(name="Flip Cover")
|
||
ard_t = PartType(name="ARD")
|
||
db.add_all([scope_t, cover_t, ard_t])
|
||
await db.flush()
|
||
|
||
rifle = SubPartType(name="Riflescope", part_type_id=scope_t.id)
|
||
reddot = SubPartType(name="Red Dot", part_type_id=scope_t.id)
|
||
db.add_all([rifle, reddot])
|
||
await db.flush()
|
||
|
||
def part(desc, type_id, manufacturer_id=1, sub_part_type_id=None):
|
||
p = AugmentedPart(
|
||
description=desc,
|
||
colour_id=1,
|
||
manufacturer_id=manufacturer_id,
|
||
part_type_id=type_id,
|
||
sub_part_type_id=sub_part_type_id,
|
||
)
|
||
db.add(p)
|
||
return p
|
||
|
||
s1 = part(
|
||
"Alpha 3-18x50", scope_t.id, manufacturer_id=10, sub_part_type_id=rifle.id
|
||
)
|
||
s2 = part("Zeta 1-6x24", scope_t.id, manufacturer_id=20, sub_part_type_id=reddot.id)
|
||
fc1 = part("Flip cover 50mm", cover_t.id)
|
||
a1 = part("ARD 50mm", ard_t.id)
|
||
a2 = part("ARD 56mm", ard_t.id)
|
||
await db.flush()
|
||
|
||
db.add(
|
||
ScopeInfo(
|
||
augmented_part_id=s1.id,
|
||
magnification_min=3,
|
||
magnification_max=18,
|
||
objective_lens_diameter_mm=50,
|
||
objective_thread_pitch="M52x0.75",
|
||
)
|
||
)
|
||
|
||
def fit(ap_a, ap_b, a, b, label):
|
||
db.add(
|
||
Fit(
|
||
attachment_point_a_id=ap_a,
|
||
attachment_point_b_id=ap_b,
|
||
augmented_part_a_id=a.id,
|
||
augmented_part_b_id=b.id,
|
||
label=label,
|
||
)
|
||
)
|
||
|
||
# S1 fits FC1 (exact) via two attachment points -> should dedup to one exact.
|
||
fit(1, 2, s1, fc1, FitLabel.POSSIBLE)
|
||
fit(3, 4, s1, fc1, FitLabel.EXACT)
|
||
# S1 fits A1 (possible); an explicit NONE to A2 must be excluded.
|
||
fit(5, 6, s1, a1, FitLabel.POSSIBLE)
|
||
fit(7, 8, s1, a2, FitLabel.NONE)
|
||
# S2 fits FC1 (possible); order reversed to exercise the b-side branch.
|
||
fit(9, 10, fc1, s2, FitLabel.POSSIBLE)
|
||
await db.commit()
|
||
|
||
return {
|
||
"scope": scope_t,
|
||
"cover": cover_t,
|
||
"ard": ard_t,
|
||
"s1": s1,
|
||
"s2": s2,
|
||
"rifle": rifle,
|
||
"reddot": reddot,
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_columns_exclude_scope_type(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
guide = await scope_guide.build(db)
|
||
|
||
names = [ct.name for ct in guide["column_types"]]
|
||
assert names == ["ARD", "Flip Cover"] # sorted, Scope excluded
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_rows_group_fits_by_type_and_dedup(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
guide = await scope_guide.build(db)
|
||
|
||
cover_id, ard_id = ids["cover"].id, ids["ard"].id
|
||
rows = {r["scope"].description: r for r in guide["rows"]}
|
||
assert list(rows) == ["Alpha 3-18x50", "Zeta 1-6x24"] # ordered by description
|
||
|
||
s1 = rows["Alpha 3-18x50"]
|
||
# Two fits to the same cover collapse to one entry, keeping the exact label.
|
||
cover_cell = s1["cells"][cover_id]
|
||
assert [(e["part"].description, e["label"]) for e in cover_cell] == [
|
||
("Flip cover 50mm", "exact")
|
||
]
|
||
# Possible ARD present; the NONE-labelled ARD is excluded entirely.
|
||
ard_cell = s1["cells"][ard_id]
|
||
assert [(e["part"].description, e["label"]) for e in ard_cell] == [
|
||
("ARD 50mm", "possible")
|
||
]
|
||
assert s1["info"] is not None
|
||
assert s1["info"].magnification_max == 18
|
||
|
||
s2 = rows["Zeta 1-6x24"]
|
||
assert [e["part"].description for e in s2["cells"][cover_id]] == ["Flip cover 50mm"]
|
||
assert ard_id not in s2["cells"] # no ARD fits for S2
|
||
assert s2["info"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_no_scope_type_returns_empty_rows(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
db.add(PartType(name="Flip Cover"))
|
||
await db.commit()
|
||
guide = await scope_guide.build(db)
|
||
assert guide["rows"] == []
|
||
assert [ct.name for ct in guide["column_types"]] == ["Flip Cover"]
|
||
|
||
|
||
def _descriptions(guide):
|
||
return [r["scope"].description for r in guide["rows"]]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_filters_by_description(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
guide = await scope_guide.build(db, description="alpha")
|
||
assert _descriptions(guide) == ["Alpha 3-18x50"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_filters_by_manufacturer(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
guide = await scope_guide.build(db, manufacturer_id=20)
|
||
assert _descriptions(guide) == ["Zeta 1-6x24"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_filters_by_sub_part_type(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
guide = await scope_guide.build(db, sub_part_type_id=ids["rifle"].id)
|
||
assert _descriptions(guide) == ["Alpha 3-18x50"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_filters_by_required_fit_type(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
# Only Alpha has an ARD fit; both scopes fit a flip cover.
|
||
ard_only = await scope_guide.build(db, fits_part_type_id=ids["ard"].id)
|
||
cover = await scope_guide.build(db, fits_part_type_id=ids["cover"].id)
|
||
assert _descriptions(ard_only) == ["Alpha 3-18x50"]
|
||
assert _descriptions(cover) == ["Alpha 3-18x50", "Zeta 1-6x24"]
|
||
|
||
|
||
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_scopes_page_renders_for_reader(client, seeded, make_user):
|
||
await make_user("reader@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
assert "Scopes" in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_renders_rows_and_fits(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
await make_user("reader2@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader2@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
body = resp.text
|
||
assert "Alpha 3-18x50" in body # scope row
|
||
assert "Flip cover 50mm" in body # a matching part in a fit column
|
||
assert "fit-badge-exact" in body # the exact fit badge
|
||
assert "M52x0.75" in body # scope info: objective thread
|
||
assert "3-18" in body # magnification range, integral values un-padded
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_includes_objective_lens_diameter(db_sessionmaker):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
guide = await scope_guide.build(db)
|
||
|
||
rows = {r["scope"].description: r for r in guide["rows"]}
|
||
assert rows["Alpha 3-18x50"]["info"].objective_lens_diameter_mm == 50
|
||
assert rows["Zeta 1-6x24"]["info"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_prefers_objective_attachment_point_over_scope_info(
|
||
db_sessionmaker,
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
apn = AttachmentPointName(name="Objective")
|
||
db.add(apn)
|
||
await db.flush()
|
||
iso = IsoThread(diameter_mm=44, pitch_mm=0.5)
|
||
db.add(iso)
|
||
await db.flush()
|
||
db.add(
|
||
AttachmentPoint(
|
||
augmented_part_id=ids["s1"].id,
|
||
attachment_point_name_id=apn.id,
|
||
iso_thread_id=iso.id,
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
guide = await scope_guide.build(db)
|
||
|
||
rows = {r["scope"].description: r for r in guide["rows"]}
|
||
s1_ap = rows["Alpha 3-18x50"]["objective_attachment_point"]
|
||
assert s1_ap is not None
|
||
assert s1_ap.iso_thread.diameter_mm == 44
|
||
# No "Objective"-named attachment point for s2 -> falls back to None.
|
||
assert rows["Zeta 1-6x24"]["objective_attachment_point"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_renders_objective_attachment_point_thread(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
apn = AttachmentPointName(name="Objective")
|
||
db.add(apn)
|
||
await db.flush()
|
||
iso = IsoThread(diameter_mm=44, pitch_mm=0.5)
|
||
db.add(iso)
|
||
await db.flush()
|
||
db.add(
|
||
AttachmentPoint(
|
||
augmented_part_id=ids["s1"].id,
|
||
attachment_point_name_id=apn.id,
|
||
iso_thread_id=iso.id,
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
await make_user("reader5@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader5@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
body = resp.text
|
||
# The attachment point's thread spec wins over ScopeInfo.objective_thread_pitch.
|
||
assert "M44.0×0.5" in body
|
||
assert "M52x0.75" not in body
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_renders_objective_lens_diameter_in_magnification(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
await make_user("reader4@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader4@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
assert "3-18x50" in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_filter_endpoint_narrows_rows(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
await make_user("reader3@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader3@example.com")
|
||
resp = await client.get("/scopes/filter", params={"description": "Alpha"})
|
||
assert resp.status_code == 200
|
||
assert "Alpha 3-18x50" in resp.text
|
||
assert "Zeta 1-6x24" not in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_filter_endpoint_forbidden_without_permission(
|
||
client, seeded, make_user
|
||
):
|
||
await make_user("plain2@example.com") # no roles, not a superuser
|
||
await _web_login(client, "plain2@example.com")
|
||
resp = await client.get("/scopes/filter")
|
||
assert resp.status_code == 403
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_forbidden_without_permission(client, seeded, make_user):
|
||
await make_user("plain@example.com") # no roles, not a superuser
|
||
await _web_login(client, "plain@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 403
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_redirects_when_anonymous(client):
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 302
|
||
# Deep link is preserved so the user lands back here after logging in.
|
||
assert resp.headers["location"] == "/login?next=%2Fscopes"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_hides_checkboxes_for_read_only_user(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db)
|
||
await make_user("reader6@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader6@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
assert 'name="scope_ids"' not in resp.text
|
||
assert "delete-selected-scopes-btn" not in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_shows_checkboxes_for_write_user(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
await make_user("writer@example.com", is_superuser=True)
|
||
await _web_login(client, "writer@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
assert f'name="scope_ids" value="{ids["s1"].id}"' in resp.text
|
||
assert "delete-selected-scopes-btn" in resp.text
|
||
assert 'hx-post="/scopes/bulk-delete"' in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bulk_delete_scopes_removes_selected_augmented_parts(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
await make_user("writer2@example.com", is_superuser=True)
|
||
await _web_login(client, "writer2@example.com")
|
||
|
||
resp = await client.post(
|
||
"/scopes/bulk-delete",
|
||
data={"scope_ids": [str(ids["s1"].id), str(ids["s2"].id)]},
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
async with db_sessionmaker() as db:
|
||
from app.crud import augmented_part as augmented_part_crud
|
||
|
||
assert await augmented_part_crud.get(db, ids["s1"].id) is None
|
||
assert await augmented_part_crud.get(db, ids["s2"].id) is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bulk_delete_scopes_leaves_unselected_rows(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
await make_user("writer3@example.com", is_superuser=True)
|
||
await _web_login(client, "writer3@example.com")
|
||
|
||
resp = await client.post(
|
||
"/scopes/bulk-delete", data={"scope_ids": [str(ids["s1"].id)]}
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
async with db_sessionmaker() as db:
|
||
from app.crud import augmented_part as augmented_part_crud
|
||
|
||
assert await augmented_part_crud.get(db, ids["s1"].id) is None
|
||
assert await augmented_part_crud.get(db, ids["s2"].id) is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_page_shows_single_page_summary(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db) # 2 scopes, well under the default page size
|
||
await make_user("pager1@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "pager1@example.com")
|
||
resp = await client.get("/scopes")
|
||
assert resp.status_code == 200
|
||
assert "Showing 2 records." in resp.text
|
||
assert "Next" not in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_filter_paginates_results(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
scope_t = PartType(name="Scope")
|
||
db.add(scope_t)
|
||
await db.flush()
|
||
for i in range(5):
|
||
db.add(
|
||
AugmentedPart(
|
||
description=f"Scope {i:02d}",
|
||
colour_id=1,
|
||
manufacturer_id=1,
|
||
part_type_id=scope_t.id,
|
||
)
|
||
)
|
||
await db.commit()
|
||
await make_user("pager2@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "pager2@example.com")
|
||
|
||
resp = await client.get("/scopes/filter", params={"page_size": 2})
|
||
assert resp.status_code == 200
|
||
body = resp.text
|
||
assert "Scope 00" in body
|
||
assert "Scope 01" in body
|
||
assert "Scope 02" not in body
|
||
assert "Showing 1–2 of 5." in body
|
||
|
||
resp2 = await client.get("/scopes/filter", params={"page_size": 2, "page": 3})
|
||
assert resp2.status_code == 200
|
||
body2 = resp2.text
|
||
assert "Scope 04" in body2
|
||
assert "Scope 00" not in body2
|
||
assert "Showing 5–5 of 5." in body2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_scopes_filter_clamps_out_of_range_page(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
await _seed_guide_data(db) # 2 scopes
|
||
await make_user("pager3@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "pager3@example.com")
|
||
|
||
resp = await client.get("/scopes/filter", params={"page_size": 1, "page": 99})
|
||
assert resp.status_code == 200
|
||
# Clamped to the last real page (2), not an empty page 99.
|
||
assert "Zeta 1-6x24" in resp.text
|
||
assert "Alpha 3-18x50" not in resp.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bulk_delete_scopes_forbidden_without_write_permission(
|
||
client, seeded, make_user, db_sessionmaker
|
||
):
|
||
async with db_sessionmaker() as db:
|
||
ids = await _seed_guide_data(db)
|
||
await make_user("reader7@example.com", role_ids=[seeded["roles"]["user"].id])
|
||
await _web_login(client, "reader7@example.com")
|
||
|
||
resp = await client.post(
|
||
"/scopes/bulk-delete", data={"scope_ids": [str(ids["s1"].id)]}
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
async with db_sessionmaker() as db:
|
||
from app.crud import augmented_part as augmented_part_crud
|
||
|
||
assert await augmented_part_crud.get(db, ids["s1"].id) is not None
|