From 7e246be39cb2068d6a5b905cb405634b4078817d Mon Sep 17 00:00:00 2001 From: Ian Roddis <31021769+iroddis@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:33:23 -0300 Subject: [PATCH] Adding autovivify aps --- app/services/scope_import.py | 43 +++++++++++++++++++++++++------ tests/test_scope_import.py | 49 ++++++++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 8 deletions(-) diff --git a/app/services/scope_import.py b/app/services/scope_import.py index c50423b..fe998ea 100644 --- a/app/services/scope_import.py +++ b/app/services/scope_import.py @@ -25,8 +25,11 @@ and colour defaults to an "Unknown" PartColour, since the sheet carries neither. Fits are created with label POSSIBLE. The referenced part is looked up by ``product_number`` and linked to its first attachment point (chosen answer: -"sole/first AP"). Both directions are resolved in a second pass so a fit between -two scopes in the same spreadsheet resolves regardless of row order. +"sole/first AP"). If the referenced part has no attachment points at all, one +is created for it that mirrors the scope's attachment point's connection, so +the fit has something valid to attach to. Both directions are resolved in a +second pass so a fit between two scopes in the same spreadsheet resolves +regardless of row order. Bad rows (missing key, unknown fit target, unparseable thread) are skipped and reported per-row; good rows are committed as they go. @@ -186,6 +189,24 @@ async def _thread_ap(db: AsyncSession, *, augmented_part_id: int, label: str, sp ) +async def _mirrored_ap(db: AsyncSession, *, augmented_part_id: int, scope_ap): + """Create an attachment point mirroring ``scope_ap``'s connection. + + Used when the referenced part has no attachment point that already fits + the scope, so the explicit ocular_fit/objective_fit link has something + valid to attach to. + """ + return await attachment_point_crud.create( + db, + augmented_part_id=augmented_part_id, + attachment_point_name_id=scope_ap.attachment_point_name_id, + iso_thread_id=scope_ap.iso_thread_id, + unc_thread_id=scope_ap.unc_thread_id, + clamp_connection_id=scope_ap.clamp_connection_id, + static_connection_id=scope_ap.static_connection_id, + ) + + async def _link_fit( db: AsyncSession, *, @@ -195,7 +216,11 @@ async def _link_fit( scope_part_id: int, messages: list[str], ) -> None: - """Create a POSSIBLE fit from the scope's AP to the referenced part's first AP.""" + """Create a POSSIBLE fit from the scope's AP to the referenced part's first AP. + + If the referenced part has no attachment points at all, one is created + mirroring the scope's attachment point's connection. + """ if scope_ap is None: messages.append( f"{scope_role}_fit '{target_sku}' skipped: scope has no {scope_role} attachment point" @@ -209,12 +234,14 @@ async def _link_fit( messages.append(f"{scope_role}_fit '{target_sku}' skipped: self-reference") return target_aps = await attachment_point_crud.list_for_augmented_part(db, target.id) - if not target_aps: + if target_aps: + target_ap = target_aps[0] + else: + target_ap = await _mirrored_ap(db, augmented_part_id=target.id, scope_ap=scope_ap) messages.append( - f"{scope_role}_fit '{target_sku}' skipped: referenced part has no attachment points" + f"{scope_role}_fit '{target_sku}': referenced part had no attachment points, " + "created one matching the scope's attachment point" ) - return - target_ap = target_aps[0] existing = await fit_crud.get_by_attachment_points(db, scope_ap, target_ap) if existing is None: await fit_crud.create(db, ap_a=scope_ap, ap_b=target_ap, label=FitLabel.POSSIBLE) @@ -253,7 +280,7 @@ async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult if objective_lens_diameter_mm: default_product_number += f"x{objective_lens_diameter_mm}" if model_suffix: - default_product_number += " {model_suffix}" + default_product_number += f" {model_suffix}" product_number = _text(row.get("product_number")) or default_product_number result = ScopeImportRowResult( row=row_num, product_number=product_number, status="skipped", messages=[] diff --git a/tests/test_scope_import.py b/tests/test_scope_import.py index fc74f7a..67b5a0a 100644 --- a/tests/test_scope_import.py +++ b/tests/test_scope_import.py @@ -235,6 +235,55 @@ async def test_unknown_fit_reference_is_reported(db_sessionmaker): 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_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")])