fitguide/tests/test_augmented_part_sync_refresh.py

"""Sync should refresh already-linked AugmentedParts' sync-owned fields when
their backing ImportedPart's data changes, add missing attachment points,
and never touch "extra" admin-owned fields or existing attachment points."""

import pytest
from sqlalchemy import select

from app.crud import attachment_point as attachment_point_crud
from app.crud import part_colour as part_colour_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 _get_ap_by_sku(db, sku):
    return (
        await db.execute(select(AugmentedPart).where(AugmentedPart.sku == sku))
    ).scalar_one()


@pytest.mark.asyncio
async def test_sync_refreshes_scalar_fields_on_already_linked_part(db_sessionmaker):
    async with db_sessionmaker() as db:
        await _seed_reference_data(db)
        db.add(
            ImportedPart(
                InventoryID="INV-REFRESH-1",
                Description="Old Scope description",
                TBXColour="Black",
            )
        )
        await db.commit()

        result = await augmented_part_sync.sync(db)
        assert result.created == 1
        assert result.updated == 0

        ap = await _get_ap_by_sku(db, "INV-REFRESH-1")
        assert ap.description == "Old Scope description"
        assert ap.uac_diameter_range_mm is None

        # Simulate the source feed changing.
        ip = await db.get(ImportedPart, "INV-REFRESH-1")
        ip.Description = "New Scope description"
        ip.TBXColour = "Tan"
        ip.UACDiameterRangemm = "03505500"
        await db.commit()

        result2 = await augmented_part_sync.sync(db)
        assert result2.created == 0
        assert result2.updated == 1

        await db.refresh(ap)
        assert ap.description == "New Scope description"
        assert ap.uac_diameter_range_mm == "03.50-55.00"
        tan = await part_colour_crud.get_by_name(db, "Tan")
        assert ap.colour_id == tan.id

        # Re-running sync with no further source changes is a no-op.
        result3 = await augmented_part_sync.sync(db)
        assert result3.updated == 0


@pytest.mark.asyncio
async def test_sync_never_touches_extra_admin_owned_fields(db_sessionmaker):
    async with db_sessionmaker() as db:
        manufacturer, colour, part_type = await _seed_reference_data(db)
        other_manufacturer = Manufacturer(name="OtherCo", url="https://other.test")
        other_part_type = PartType(name="Flip Cover")
        db.add_all([other_manufacturer, other_part_type])
        await db.flush()

        db.add(ImportedPart(InventoryID="INV-EXTRA-1", Description="A Scope widget"))
        await db.commit()

        result = await augmented_part_sync.sync(db)
        assert result.created == 1

        ap = await _get_ap_by_sku(db, "INV-EXTRA-1")

        # Simulate an admin manually re-classifying the part after creation.
        ap.part_type_id = other_part_type.id
        ap.sub_part_type_id = None
        ap.manufacturer_id = other_manufacturer.id
        ap.archived = False
        await db.commit()
        admin_part_type_id = ap.part_type_id
        admin_manufacturer_id = ap.manufacturer_id

        # Source feed also changes, to prove refresh runs at all.
        ip = await db.get(ImportedPart, "INV-EXTRA-1")
        ip.Description = "A Scope widget, renamed"
        await db.commit()

        result2 = await augmented_part_sync.sync(db)
        assert result2.updated == 1

        await db.refresh(ap)
        assert ap.description == "A Scope widget, renamed"  # sync-owned: refreshed
        assert ap.part_type_id == admin_part_type_id  # extra field: untouched
        assert ap.manufacturer_id == admin_manufacturer_id  # extra field: untouched
        assert ap.sku == "INV-EXTRA-1"  # extra field: untouched


@pytest.mark.asyncio
async def test_sync_adds_missing_attachment_point_without_touching_existing_one(
    db_sessionmaker,
):
    async with db_sessionmaker() as db:
        await _seed_reference_data(db)
        db.add(
            ImportedPart(
                InventoryID="INV-AP-1",
                Description="A Scope widget",
                # No thread/OD/UAC data yet — no attachment points created.
            )
        )
        await db.commit()

        result = await augmented_part_sync.sync(db)
        assert result.created == 1

        ap = await _get_ap_by_sku(db, "INV-AP-1")
        points = await attachment_point_crud.list_for_augmented_part(db, ap.id)
        assert points == []

        # Source feed gains an OD value -> an "Ocular" point should appear.
        ip = await db.get(ImportedPart, "INV-AP-1")
        ip.ProductODmm = "40"
        ip.ProductODin = "1.575"
        await db.commit()

        result2 = await augmented_part_sync.sync(db)
        assert result2.updated == 1

        points = await attachment_point_crud.list_for_augmented_part(db, ap.id)
        assert len(points) == 1
        ocular = points[0]
        assert ocular.attachment_point_name.name == "Ocular"
        original_static_connection_id = ocular.static_connection_id

        # Source OD changes again — the existing point must NOT be replaced.
        ip.ProductODmm = "56"
        ip.ProductODin = "2.205"
        await db.commit()

        result3 = await augmented_part_sync.sync(db)
        # description/colour/uac unchanged, and the Ocular point already
        # exists, so nothing should be reported as updated.
        assert result3.updated == 0

        points = await attachment_point_crud.list_for_augmented_part(db, ap.id)
        assert len(points) == 1
        assert points[0].id == ocular.id
        assert points[0].static_connection_id == original_static_connection_id