398 lines
15 KiB
Python
398 lines
15 KiB
Python
import io
|
|
|
|
import pytest
|
|
from openpyxl import Workbook
|
|
|
|
from app.crud import attachment_point as attachment_point_crud
|
|
from app.crud import attachment_point_name as apn_crud
|
|
from app.crud import augmented_part as augmented_part_crud
|
|
from app.crud import fit as fit_crud
|
|
from app.crud import manufacturer as manufacturer_crud
|
|
from app.crud import part_colour as part_colour_crud
|
|
from app.crud import part_type as part_type_crud
|
|
from app.crud import scope_info as scope_info_crud
|
|
from app.crud import static_connection as static_connection_crud
|
|
from app.models.fit import FitLabel
|
|
from app.services import scope_import as scope_import_service
|
|
|
|
HEADERS = [
|
|
"manufacturer",
|
|
"model",
|
|
"magnification_min",
|
|
"magnification_max",
|
|
"objective_lens_diameter_mm",
|
|
"model_suffix",
|
|
"sub_part_type",
|
|
"product_number",
|
|
"url",
|
|
"ocular_outer_diameter_mm",
|
|
"ocular_fit",
|
|
"objective_thread",
|
|
"objective_fit",
|
|
]
|
|
|
|
|
|
def _workbook(rows, headers=HEADERS):
|
|
# ``headers`` are the labels written to the header row (may be re-cased to
|
|
# exercise case-insensitive matching); values are always pulled by the
|
|
# canonical lower-case keys the row dicts use.
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.append(headers)
|
|
for r in rows:
|
|
ws.append([r.get(key) for key in HEADERS])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def _scope_row(**overrides):
|
|
row = {
|
|
"manufacturer": "Vortex",
|
|
"model": "Viper",
|
|
"magnification_min": 5,
|
|
"magnification_max": 25,
|
|
"objective_lens_diameter_mm": 56,
|
|
"model_suffix": "PST",
|
|
"sub_part_type": "Riflescope",
|
|
"product_number": "SCOPE-A",
|
|
"url": "https://example.com/viper",
|
|
"ocular_outer_diameter_mm": 40,
|
|
"ocular_fit": None,
|
|
"objective_thread": "M40x0.75",
|
|
"objective_fit": None,
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
async def _make_accessory(db, sku, od_mm):
|
|
"""A minimal non-scope part with a single static attachment point."""
|
|
colour = await part_colour_crud.get_by_name(db, "Black")
|
|
if colour is None:
|
|
colour = await part_colour_crud.create(db, name="Black")
|
|
manu = await manufacturer_crud.get_by_name(db, "Armament")
|
|
if manu is None:
|
|
manu = await manufacturer_crud.create(db, name="Armament", url="")
|
|
pt = await part_type_crud.get_by_name(db, "Flip Cover")
|
|
if pt is None:
|
|
pt = await part_type_crud.create(db, name="Flip Cover")
|
|
part = await augmented_part_crud.create(
|
|
db,
|
|
sku=sku,
|
|
description=f"Accessory {sku}",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manu.id,
|
|
scope_model_id=None,
|
|
part_type_id=pt.id,
|
|
sub_part_type_id=None,
|
|
)
|
|
sc = await static_connection_crud.create(db, od_mm=od_mm, od_in=od_mm / 25.4)
|
|
apn = await apn_crud.get_by_name(db, "Cover")
|
|
if apn is None:
|
|
apn = await apn_crud.create(db, name="Cover")
|
|
ap = await attachment_point_crud.create(
|
|
db,
|
|
augmented_part_id=part.id,
|
|
attachment_point_name_id=apn.id,
|
|
static_connection_id=sc.id,
|
|
)
|
|
return part, ap
|
|
|
|
|
|
# ---- parsing ---------------------------------------------------------------
|
|
|
|
|
|
def test_parse_workbook_is_case_insensitive_and_skips_blank_rows():
|
|
headers = [h.upper() for h in HEADERS]
|
|
data = _workbook([_scope_row(), _scope_row(product_number=None)], headers=headers)
|
|
# add a fully blank row is implicit via None product; both rows are non-blank here
|
|
rows = scope_import_service.parse_workbook(data)
|
|
assert len(rows) == 2
|
|
assert rows[0]["manufacturer"] == "Vortex"
|
|
assert rows[0]["product_number"] == "SCOPE-A"
|
|
|
|
|
|
# ---- import ----------------------------------------------------------------
|
|
|
|
|
|
async def test_import_creates_scope_with_info_and_attachment_points(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
data = _workbook([_scope_row()])
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
|
|
assert result.created == 1
|
|
assert result.updated == 0
|
|
assert result.total == 1
|
|
|
|
part = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
assert part is not None
|
|
assert part.description == "Vortex Viper PST"
|
|
assert part.part_type.name == "Scope"
|
|
assert part.colour.name == "Unknown"
|
|
assert part.manufacturer.name == "Vortex"
|
|
assert part.scope_model is not None
|
|
assert part.scope_model.name == "Viper"
|
|
assert part.sub_part_type is not None
|
|
assert part.sub_part_type.name == "Riflescope"
|
|
|
|
info = await scope_info_crud.get_by_augmented_part(db, part.id)
|
|
assert info.magnification_min == 5
|
|
assert info.magnification_max == 25
|
|
assert info.model_suffix == "PST"
|
|
assert info.objective_lens_diameter_mm == 56
|
|
|
|
aps = await attachment_point_crud.list_for_augmented_part(db, part.id)
|
|
by_label = {}
|
|
for ap in aps:
|
|
by_label.setdefault(ap.attachment_point_name.name, []).append(ap)
|
|
|
|
assert len(by_label["Ocular"]) == 1
|
|
assert by_label["Ocular"][0].static_connection.od_mm == 40
|
|
|
|
# The lens diameter is descriptive only: the sole Objective point is the
|
|
# thread.
|
|
objectives = by_label["Objective"]
|
|
assert len(objectives) == 1
|
|
assert objectives[0].static_connection is None
|
|
assert objectives[0].iso_thread.diameter_mm == 40
|
|
assert objectives[0].iso_thread.pitch_mm == 0.75
|
|
|
|
|
|
async def test_import_resolves_ocular_and_objective_fits(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
_, acc1_ap = await _make_accessory(db, "ACC-1", od_mm=40)
|
|
_, acc2_ap = await _make_accessory(db, "ACC-2", od_mm=56)
|
|
|
|
data = _workbook([_scope_row(ocular_fit="ACC-1", objective_fit="ACC-2")])
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
assert result.created == 1
|
|
|
|
part = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
fits = await fit_crud.list_for_part(db, part.id)
|
|
assert len(fits) == 2
|
|
for f in fits:
|
|
assert f.label == FitLabel.POSSIBLE
|
|
linked_ap_ids = {f.attachment_point_a_id for f in fits} | {
|
|
f.attachment_point_b_id for f in fits
|
|
}
|
|
assert acc1_ap.id in linked_ap_ids
|
|
assert acc2_ap.id in linked_ap_ids
|
|
|
|
|
|
async def test_import_upserts_on_product_number(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
await scope_import_service.import_from_bytes(db, _workbook([_scope_row()]))
|
|
# Re-import same product number with a changed model.
|
|
result = await scope_import_service.import_from_bytes(
|
|
db, _workbook([_scope_row(model="Razor")])
|
|
)
|
|
assert result.updated == 1
|
|
assert result.created == 0
|
|
|
|
# Only one part with that sku, and its model was updated.
|
|
parts = await augmented_part_crud.list_all(db)
|
|
matching = [p for p in parts if p.sku == "SCOPE-A"]
|
|
assert len(matching) == 1
|
|
assert matching[0].scope_model.name == "Razor"
|
|
|
|
# Attachment points were rebuilt, not duplicated.
|
|
aps = await attachment_point_crud.list_for_augmented_part(db, matching[0].id)
|
|
assert len(aps) == 2
|
|
|
|
|
|
@pytest.mark.skip(
|
|
"Product numbers are now autovivified from model and other attributes"
|
|
)
|
|
async def test_row_without_product_number_is_skipped(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
data = _workbook([_scope_row(product_number=None)])
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
assert result.created == 0
|
|
assert result.skipped == 1
|
|
assert result.rows[0].status == "skipped"
|
|
assert "missing product_number" in result.rows[0].messages
|
|
|
|
|
|
async def test_unparseable_objective_thread_is_reported(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
data = _workbook([_scope_row(objective_thread="not-a-thread")])
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
assert result.created == 1
|
|
msgs = " ".join(result.rows[0].messages)
|
|
assert "could not parse objective_thread" in msgs
|
|
|
|
|
|
async def test_unknown_fit_reference_is_reported(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
data = _workbook([_scope_row(ocular_fit="DOES-NOT-EXIST")])
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
assert result.created == 1
|
|
msgs = " ".join(result.rows[0].messages)
|
|
assert "unknown product 'DOES-NOT-EXIST'" in msgs
|
|
|
|
part = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
assert await fit_crud.list_for_part(db, part.id) == []
|
|
|
|
|
|
async def _make_bare_part(db, sku):
|
|
"""A non-scope part with no attachment points at all."""
|
|
colour = await part_colour_crud.get_by_name(db, "Black")
|
|
if colour is None:
|
|
colour = await part_colour_crud.create(db, name="Black")
|
|
manu = await manufacturer_crud.get_by_name(db, "Armament")
|
|
if manu is None:
|
|
manu = await manufacturer_crud.create(db, name="Armament", url="")
|
|
pt = await part_type_crud.get_by_name(db, "Flip Cover")
|
|
if pt is None:
|
|
pt = await part_type_crud.create(db, name="Flip Cover")
|
|
return await augmented_part_crud.create(
|
|
db,
|
|
sku=sku,
|
|
description=f"Accessory {sku}",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manu.id,
|
|
scope_model_id=None,
|
|
part_type_id=pt.id,
|
|
sub_part_type_id=None,
|
|
)
|
|
|
|
|
|
async def test_fit_reference_without_attachment_points_creates_matching_one(
|
|
db_sessionmaker,
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
await _make_bare_part(db, "ACC-BARE")
|
|
|
|
data = _workbook([_scope_row(ocular_fit="ACC-BARE")])
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
assert result.created == 1
|
|
msgs = " ".join(result.rows[0].messages)
|
|
assert "created one matching the scope's attachment point" in msgs
|
|
|
|
acc = await augmented_part_crud.get_by_sku(db, "ACC-BARE")
|
|
acc_aps = await attachment_point_crud.list_for_augmented_part(db, acc.id)
|
|
assert len(acc_aps) == 1
|
|
assert acc_aps[0].static_connection is not None
|
|
assert acc_aps[0].static_connection.od_mm == 40
|
|
assert acc_aps[0].attachment_point_name.name == "Ocular"
|
|
|
|
part = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
fits = await fit_crud.list_for_part(db, part.id)
|
|
assert len(fits) == 1
|
|
linked_ap_ids = {fits[0].attachment_point_a_id, fits[0].attachment_point_b_id}
|
|
assert acc_aps[0].id in linked_ap_ids
|
|
|
|
|
|
async def test_scope_fit_without_scope_attachment_point_creates_manual_one(
|
|
db_sessionmaker,
|
|
):
|
|
async with db_sessionmaker() as db:
|
|
_, acc_ap = await _make_accessory(db, "ACC-1", od_mm=40)
|
|
|
|
# No ocular_outer_diameter_mm, so the scope itself has no Ocular AP,
|
|
# but an ocular_fit reference is still given.
|
|
data = _workbook(
|
|
[_scope_row(ocular_outer_diameter_mm=None, ocular_fit="ACC-1")]
|
|
)
|
|
result = await scope_import_service.import_from_bytes(db, data)
|
|
assert result.created == 1
|
|
msgs = " ".join(result.rows[0].messages)
|
|
assert "created a manual one to record the fit" in msgs
|
|
|
|
part = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
scope_aps = await attachment_point_crud.list_for_augmented_part(db, part.id)
|
|
by_label = {ap.attachment_point_name.name: ap for ap in scope_aps}
|
|
assert "Ocular" in by_label
|
|
ocular_ap = by_label["Ocular"]
|
|
assert ocular_ap.iso_thread is None
|
|
assert ocular_ap.unc_thread is None
|
|
assert ocular_ap.clamp_connection is None
|
|
assert ocular_ap.static_connection is None
|
|
# Objective still gets its normal thread-based AP.
|
|
assert "Objective" in by_label
|
|
|
|
fits = await fit_crud.list_for_part(db, part.id)
|
|
assert len(fits) == 1
|
|
linked_ap_ids = {fits[0].attachment_point_a_id, fits[0].attachment_point_b_id}
|
|
assert ocular_ap.id in linked_ap_ids
|
|
assert acc_ap.id in linked_ap_ids
|
|
|
|
|
|
async def test_unc_objective_thread_creates_unc_attachment_point(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
data = _workbook([_scope_row(objective_thread="1.750-32 UNS")])
|
|
await scope_import_service.import_from_bytes(db, data)
|
|
part = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
aps = await attachment_point_crud.list_for_augmented_part(db, part.id)
|
|
unc_aps = [a for a in aps if a.unc_thread is not None]
|
|
assert len(unc_aps) == 1
|
|
assert unc_aps[0].unc_thread.diameter_fraction == "1.750"
|
|
assert unc_aps[0].unc_thread.tpi == 32
|
|
|
|
|
|
# ---- API endpoint ----------------------------------------------------------
|
|
|
|
|
|
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 test_api_upload_requires_permission(client, seeded, make_user):
|
|
# "user" role only grants reads, not augmented_parts:write.
|
|
await make_user("reader@example.com", role_ids=[seeded["roles"]["user"].id])
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'reader@example.com')}"}
|
|
files = {
|
|
"file": (
|
|
"scopes.xlsx",
|
|
_workbook([_scope_row()]),
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
)
|
|
}
|
|
resp = await client.post("/api/scope-import", files=files, headers=headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_api_upload_imports_scopes(client, seeded, make_user):
|
|
await make_user("root@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root@example.com')}"}
|
|
files = {
|
|
"file": (
|
|
"scopes.xlsx",
|
|
_workbook([_scope_row()]),
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
)
|
|
}
|
|
resp = await client.post("/api/scope-import", files=files, headers=headers)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["created"] == 1
|
|
assert body["total"] == 1
|
|
assert body["rows"][0]["product_number"] == "SCOPE-A"
|
|
|
|
|
|
async def test_api_upload_rejects_non_xlsx(client, seeded, make_user):
|
|
await make_user("root2@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root2@example.com')}"}
|
|
files = {"file": ("scopes.csv", b"not,a,spreadsheet", "text/csv")}
|
|
resp = await client.post("/api/scope-import", files=files, headers=headers)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
async def test_fits_between_two_scopes_in_same_sheet_resolve(db_sessionmaker):
|
|
async with db_sessionmaker() as db:
|
|
rows = [
|
|
_scope_row(product_number="SCOPE-A", ocular_fit="SCOPE-B"),
|
|
_scope_row(product_number="SCOPE-B", model="Diamondback"),
|
|
]
|
|
result = await scope_import_service.import_from_bytes(db, _workbook(rows))
|
|
assert result.created == 2
|
|
|
|
a = await augmented_part_crud.get_by_sku(db, "SCOPE-A")
|
|
fits = await fit_crud.list_for_part(db, a.id)
|
|
assert len(fits) == 1
|