from datetime import datetime, timedelta from db.channel_capabilities import resolve_product_structure_for_connection from db.channel_exports import build_channel_product_payloads from db.channel_projection_cache import ( load_cached_payloads, mark_projection_rows_stale, partition_projection_cache, store_projection_payloads, ) from db.channel_projection_invalidation import ( inspect_projection_drift, invalidate_skus, ) from db.manual_taxonomy_assignments import replace_manual_taxonomy_assignments from db.models import ( ChannelVariationPolicy, ChannelProjectionRow, ChannelPublishState, MasterProduct, ManualTaxonomyAssignment, ManualTaxonomyAssignmentCollection, MasterCollectionRegistry, MasterProductOverride, MasterTaxonomyChannelLink, MasterTaxonomyNode, ) def _payload(sku: str, *, title: str, price: float = 199.0, row_hash: str = "hash-a") -> dict: return { "sku": sku, "channel_sku": f"HD-{sku}", "channel_code": "shopify", "product_role": "standalone", "canonical_fields": {"title": title, "brand": "Port & Bell"}, "fields": {"title": title, "vendor": "Port & Bell"}, "relation_fields": {}, "association_fields": {}, "taxonomy_targets": {}, "price": price, "prices": {"shopify": price}, "payload_hash": f"payload-{sku}", "source_row_hash": row_hash, "override_codes": ["title"] if sku.endswith("12") else [], } def test_channel_projection_cache_round_trip(catalog_intent_session): result = store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, payloads=[_payload("ACH-B12", title="Base Cabinet")], product_structure="flat", ) assert result["status"] == "ready" cached = load_cached_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, skus=["ACH-B12"], product_structure="flat", ) assert len(cached) == 1 assert cached[0]["sku"] == "ACH-B12" assert cached[0]["channel_sku"] == "HD-ACH-B12" assert cached[0]["fields"]["vendor"] == "Port & Bell" assert cached[0]["override_codes"] == ["title"] def test_build_channel_product_payloads_uses_projection_cache(catalog_intent_session): store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[_payload("ACH-B15", title="Base Cabinet 15", price=249.0, row_hash="h15")], ) payloads = build_channel_product_payloads( catalog_intent_session, "shopify", skus=["ACH-B15"], only_assigned=False, connection_id=1001, ) assert len(payloads) == 1 assert payloads[0]["sku"] == "ACH-B15" assert payloads[0]["channel_sku"] == "HD-ACH-B15" assert payloads[0]["price"] == 249.0 def test_build_channel_product_payloads_corrects_cached_shopping_fields_from_taxonomy_targets( catalog_intent_session, monkeypatch, ): monkeypatch.setattr("db.channel_exports._attach_channel_skus", lambda *args, **kwargs: None) monkeypatch.setattr("db.channel_exports._attach_association_fields", lambda *args, **kwargs: None) monkeypatch.setattr("db.channel_exports._attach_url_canonicals", lambda *args, **kwargs: None) monkeypatch.setattr("db.channel_exports._attach_taxonomy_targets", lambda *args, **kwargs: None) catalog_intent_session.add( MasterProduct(sku="ASG-W1230", name="ASG W1230", row_hash="stable-row", is_active=True) ) catalog_intent_session.flush() store_projection_payloads( catalog_intent_session, channel_code="magento", connection_id=4, product_structure="flat", payloads=[ { "sku": "ASG-W1230", "channel_sku": "ASG-W1230", "channel_code": "magento", "product_role": "standalone", "canonical_fields": { "title": "W1230", "category_l1": "Kitchen Cabinets", "collection": "Anna Stone Gray", "shopping_collection": "anna-stone-gray", "shopping_l1": "wall-cabinets", "shopping_l2": "wall-cabinets/single-door-glass-wall-cabinet", }, "fields": { "title": "W1230", "shopping_collection": "anna-stone-gray", "shopping_l1": "wall-cabinets", "shopping_l2": "wall-cabinets/single-door-glass-wall-cabinet", }, "relation_fields": {}, "association_fields": {}, "taxonomy_targets": { "plp_path_slugs": [ "kitchen-cabinets/anna-stone-gray", "all-products", "kitchen-cabinets/wall-cabinets", "kitchen-cabinets/wall-cabinets/basic-wall", ] }, "price": 283.25, "prices": {"base": 283.25}, "payload_hash": "stale-hash", "source_row_hash": "stable-row", "override_codes": [], } ], ) payloads = build_channel_product_payloads( catalog_intent_session, "magento", skus=["ASG-W1230"], only_assigned=False, connection_id=4, product_structure="flat", ) assert len(payloads) == 1 assert payloads[0]["fields"]["shopping_l2"] == "wall-cabinets/basic-wall" assert payloads[0]["canonical_fields"]["shopping_l2"] == "wall-cabinets/basic-wall" assert payloads[0]["payload_hash"] != "stale-hash" def test_store_projection_payloads_upserts_without_wiping_siblings(catalog_intent_session): store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[ _payload("ACH-B12", title="One"), _payload("ACH-B15", title="Two"), ], ) store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[_payload("ACH-B12", title="One Updated", row_hash="hash-b")], replace_scope=False, ) rows = list( catalog_intent_session.query(ChannelProjectionRow) .filter(ChannelProjectionRow.connection_id == 1001) .order_by(ChannelProjectionRow.master_sku) .all() ) assert [row.master_sku for row in rows] == ["ACH-B12", "ACH-B15"] assert rows[0].channel_fields_json["title"] == "One Updated" assert rows[1].channel_fields_json["title"] == "Two" def test_hash_drift_forces_partial_rebuild(catalog_intent_session): catalog_intent_session.add( MasterProduct(sku="ACH-B12", name="Base 12", row_hash="live-hash", is_active=True) ) catalog_intent_session.add( MasterProduct(sku="ACH-B15", name="Base 15", row_hash="stable-hash", is_active=True) ) catalog_intent_session.flush() store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[ _payload("ACH-B12", title="Stale", row_hash="old-hash"), _payload("ACH-B15", title="Fresh", price=249.0, row_hash="stable-hash"), ], ) partitioned = partition_projection_cache( catalog_intent_session, channel_code="shopify", connection_id=1001, skus=["ACH-B12", "ACH-B15"], product_structure="flat", ) assert partitioned["rebuild_skus"] == ["ACH-B12"] assert [p["sku"] for p in partitioned["fresh"]] == ["ACH-B15"] def test_inspect_marks_override_and_failure_drift(catalog_intent_session): now = datetime.utcnow() store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[_payload("ACH-B12", title="Base", row_hash="h1")], ) row = catalog_intent_session.query(ChannelProjectionRow).one() row.normalized_at = now - timedelta(hours=2) catalog_intent_session.flush() catalog_intent_session.add( MasterProductOverride( sku="ACH-B12", channel_code="shopify", attribute_code="title", value="Override Title", is_active=True, updated_at=now, ) ) catalog_intent_session.add( ChannelPublishState( sku="ACH-B15", channel_code="shopify", last_status="error", last_error="boom", ) ) # Ensure ACH-B15 has a projection row for failure drift store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[_payload("ACH-B15", title="Other", row_hash="h2")], replace_scope=False, ) row15 = ( catalog_intent_session.query(ChannelProjectionRow) .filter(ChannelProjectionRow.master_sku == "ACH-B15") .one() ) row15.normalized_at = now - timedelta(hours=1) catalog_intent_session.flush() report = inspect_projection_drift( catalog_intent_session, channel_code="shopify", connection_id=1001, apply=True, ) assert "ACH-B12" in report["drift_skus"] assert "override_updated" in report["reasons"]["ACH-B12"] assert "ACH-B15" in report["drift_skus"] assert "prior_publish_failure" in report["reasons"]["ACH-B15"] assert report["invalidated"] >= 2 statuses = { row.master_sku: row.projection_status for row in catalog_intent_session.query(ChannelProjectionRow).all() } assert statuses["ACH-B12"] == "stale" assert statuses["ACH-B15"] == "stale" def test_failure_invalidation_marks_rows_stale(catalog_intent_session): store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=1001, product_structure="flat", payloads=[_payload("ACH-B12", title="Base")], ) result = invalidate_skus( catalog_intent_session, channel_code="shopify", connection_id=1001, skus=["ACH-B12"], reason="shopify_push_failure:timeout", ) assert result["invalidated"] == 1 row = catalog_intent_session.query(ChannelProjectionRow).one() assert row.projection_status == "stale" partitioned = partition_projection_cache( catalog_intent_session, channel_code="shopify", connection_id=1001, skus=["ACH-B12"], product_structure="flat", ) assert partitioned["rebuild_skus"] == ["ACH-B12"] assert partitioned["fresh"] == [] def test_build_channel_product_payloads_applies_exact_manual_taxonomy_override_to_magento_targets( catalog_intent_session, monkeypatch, ): session = catalog_intent_session monkeypatch.setattr("db.channel_exports._attach_channel_skus", lambda *args, **kwargs: None) monkeypatch.setattr("db.channel_exports._attach_association_fields", lambda *args, **kwargs: None) monkeypatch.setattr("db.channel_exports._attach_url_canonicals", lambda *args, **kwargs: None) def _seed_taxonomy_targets(_session, _channel, payloads, **kwargs): for payload in payloads: payload["taxonomy_targets"] = { "plp_path_slugs": ["all-products"], "magento_category_ids": [111], "magento_category_paths": ["Default Category/Existing"], } payload["payload_hash"] = "seeded" monkeypatch.setattr("db.channel_exports._attach_taxonomy_targets", _seed_taxonomy_targets) session.add( MasterProduct( sku="ACH-W0930", name="Wall Cabinet 09x30", category_l1="Kitchen Cabinets", category_l2="Legacy Wall Cabinets", category_l3="Legacy Basic Wall", collection="Legacy Collection", row_hash="row-ach-w0930", is_active=True, ) ) session.add( MasterCollectionRegistry( code="ACH", name="Anna Caramel Harvest", path_slug="kitchen-cabinets/anna-caramel-harvest", is_active=True, aliases={"source_collection_code": "ACH", "alias_codes": ["ACH"]}, ) ) session.add_all( [ MasterTaxonomyNode( id=100, name="Anna Caramel Harvest", path_slug="kitchen-cabinets/anna-caramel-harvest", node_kind="collection", is_active=True, ), MasterTaxonomyNode( id=101, name="Microwave Shelf", path_slug="kitchen-cabinets/wall-cabinets/microwave-shelf", node_kind="category", is_active=True, ), ] ) session.add_all( [ MasterTaxonomyChannelLink( taxonomy_node_id=100, channel_code="magento", connection_id=4, taxonomy_kind="collection", remote_id="900", remote_path="Default Category/Kitchen Cabinets/Anna Caramel Harvest", is_active=True, ), MasterTaxonomyChannelLink( taxonomy_node_id=101, channel_code="magento", connection_id=4, taxonomy_kind="category", remote_id="901", remote_path="Default Category/Kitchen Cabinets/Wall Cabinets/Microwave Shelf", is_active=True, ), ] ) session.flush() replace_manual_taxonomy_assignments( session, { "templates": [ { "source_sku": "W0930", "canonical_taxonomy_path_slug": "kitchen-cabinets/base-cabinets/single-door", "canonical_taxonomy_path_label": "Kitchen Cabinets / Base Cabinets / Single Door", "shopping_l1": "base-cabinets", "shopping_l2": "base-cabinets/single-door", "l2_category": "Base Cabinets", "l3_category": "Single Door", "collections": [ { "collection_code": "ACH", "source_collection_code": "ACH", "sku_prefix": "ACH", "collection_name": "Anna Caramel Harvest", "collection_path_slug": "kitchen-cabinets/anna-caramel-harvest", "shopping_collection": "anna-caramel-harvest", "master_sku": "ACH-W0930", } ], }, { "source_sku": "ACH-W0930", "canonical_taxonomy_path_slug": "kitchen-cabinets/wall-cabinets/microwave-shelf", "canonical_taxonomy_path_label": "Kitchen Cabinets / Wall Cabinets / Microwave Shelf", "shopping_l1": "wall-cabinets", "shopping_l2": "wall-cabinets/microwave-shelf", "l2_category": "Wall Cabinets", "l3_category": "Microwave Shelf", "collections": [ { "collection_code": "ACH", "source_collection_code": "ACH", "sku_prefix": "ACH", "collection_name": "Anna Caramel Harvest", "collection_path_slug": "kitchen-cabinets/anna-caramel-harvest", "shopping_collection": "anna-caramel-harvest", "master_sku": "ACH-W0930", } ], }, ] }, ) exact_assignment = ( session.query(ManualTaxonomyAssignment) .filter(ManualTaxonomyAssignment.source_sku == "ACH-W0930") .one() ) exact_assignment.updated_at = datetime.utcnow() + timedelta(seconds=5) exact_child = ( session.query(ManualTaxonomyAssignmentCollection) .filter(ManualTaxonomyAssignmentCollection.assignment_id == exact_assignment.id) .one() ) exact_child.updated_at = exact_assignment.updated_at session.flush() payloads = build_channel_product_payloads( session, "magento", skus=["ACH-W0930"], only_assigned=False, connection_id=4, product_structure="flat", prefer_projection_cache=False, ) assert len(payloads) == 1 payload = payloads[0] assert payload["canonical_fields"]["category_l2"] == "Wall Cabinets" assert payload["canonical_fields"]["category_l3"] == "Microwave Shelf" assert payload["canonical_fields"]["collection"] == "Anna Caramel Harvest" assert payload["fields"]["shopping_collection"] == "anna-caramel-harvest" assert payload["fields"]["shopping_l1"] == "wall-cabinets" assert payload["fields"]["shopping_l2"] == "wall-cabinets/microwave-shelf" assert payload["taxonomy_targets"]["plp_path_slugs"] == [ "all-products", "kitchen-cabinets/anna-caramel-harvest", "kitchen-cabinets/wall-cabinets", "kitchen-cabinets/wall-cabinets/microwave-shelf", ] assert payload["taxonomy_targets"]["magento_category_ids"] == [111, 900, 901] assert payload["taxonomy_targets"]["magento_category_paths"] == [ "Default Category/Existing", "Default Category/Kitchen Cabinets/Anna Caramel Harvest", "Default Category/Kitchen Cabinets/Wall Cabinets/Microwave Shelf", ] assert payload["payload_hash"] != "seeded" def test_native_connection_id_normalization_for_shopify_compat(catalog_intent_session): """Shopify compat ids (1_000_000+n) must resolve to the same cache rows as native n.""" store_projection_payloads( catalog_intent_session, channel_code="shopify", connection_id=7, product_structure="flat", payloads=[_payload("ACH-B12", title="Native")], ) cached = load_cached_payloads( catalog_intent_session, channel_code="shopify", connection_id=1_000_007, skus=["ACH-B12"], product_structure="flat", ) assert len(cached) == 1 assert cached[0]["sku"] == "ACH-B12" def test_variation_policy_controls_connection_product_structure(catalog_intent_session): catalog_intent_session.add( ChannelVariationPolicy( channel_code="shopify", connection_id=7, variation_mode="single_axis", is_active=True, ) ) catalog_intent_session.flush() assert ( resolve_product_structure_for_connection( catalog_intent_session, "shopify", connection_id=7, ) == "parent_children" )