from __future__ import annotations from datetime import datetime, timezone from sqlalchemy import select, text from sqlalchemy.orm import Session from db.models import ( AttributeDef, AttributeOption, ChannelAttributeAlias, MagentoAttributeOptionRegistry, MagentoAttributeRegistry, MagentoConnection, ManualAttributeVocabulary, ManualAttributeVocabularyValue, MasterAttributeDefinition, MasterCollectionRegistry, MasterProduct, MasterProductAttributeValue, MasterProductFamily, MasterTaxonomyNode, ) from db.models_v2 import ( V2Attribute, V2AttributeSeedCandidate, V2AttributeSeedMappingCandidate, V2AttributeValue, V2AttributeValueSeedCandidate, V2ChannelAttributeMapping, V2ChannelAttributeValueMapping, ) from db.v2_attribute_review import ( apply_override_payload, approve_candidate, bulk_merge_candidates, bulk_merge_values, candidate_detail, create_and_merge_values, create_value, exclude_candidate, merge_candidate, unmapped_product_value_report, update_candidate, upsert_candidate_mapping, ) from db.v2_attribute_seed import apply_approved_attribute_seed, canonicalize_collection_family_label, seed_v2_attributes def _create_v2_attribute_tables(session: Session) -> None: for stmt in ( """ CREATE TABLE IF NOT EXISTS v2_attribute ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, label TEXT NOT NULL, data_type TEXT NOT NULL, ui_control TEXT NOT NULL DEFAULT 'text_field', is_variant_axis BOOLEAN NOT NULL DEFAULT 0, is_channel_required BOOLEAN NOT NULL DEFAULT 1, lifecycle_status TEXT NOT NULL DEFAULT 'active', aliases JSON NOT NULL DEFAULT '[]', applies_to_product_types JSON NOT NULL DEFAULT '[]', owner_scope TEXT NOT NULL DEFAULT 'product', created_at TEXT DEFAULT CURRENT_TIMESTAMP ) """, """ CREATE TABLE IF NOT EXISTS v2_attribute_value ( id INTEGER PRIMARY KEY AUTOINCREMENT, attribute_id INTEGER NOT NULL, value_code TEXT NOT NULL, display_label TEXT NOT NULL, sort_order INTEGER NOT NULL DEFAULT 0, lifecycle_status TEXT NOT NULL DEFAULT 'active', aliases JSON NOT NULL DEFAULT '[]', created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(attribute_id, value_code), FOREIGN KEY(attribute_id) REFERENCES v2_attribute(id) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS v2_channel_attribute_mapping ( id INTEGER PRIMARY KEY AUTOINCREMENT, channel TEXT NOT NULL, canonical_attribute_id INTEGER NOT NULL, channel_attribute_code TEXT, channel_attribute_id TEXT, sync_mode TEXT, is_optional BOOLEAN NOT NULL DEFAULT 0, representation JSON NOT NULL DEFAULT '{}', compatibility_status TEXT NOT NULL DEFAULT 'unresolved', created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(channel, canonical_attribute_id), FOREIGN KEY(canonical_attribute_id) REFERENCES v2_attribute(id) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS v2_channel_attribute_value_mapping ( id INTEGER PRIMARY KEY AUTOINCREMENT, channel TEXT NOT NULL, canonical_attribute_value_id INTEGER NOT NULL, channel_option_id TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(channel, canonical_attribute_value_id), FOREIGN KEY(canonical_attribute_value_id) REFERENCES v2_attribute_value(id) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS v2_attribute_seed_candidate ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_code TEXT NOT NULL UNIQUE, proposed_code TEXT NOT NULL, proposed_label TEXT NOT NULL, discovered_data_type TEXT NOT NULL, proposed_data_type TEXT NOT NULL, proposed_ui_control TEXT NOT NULL DEFAULT 'text_field', is_variant_axis BOOLEAN NOT NULL DEFAULT 0, is_channel_required BOOLEAN NOT NULL DEFAULT 0, lifecycle_status TEXT NOT NULL DEFAULT 'active', owner_scope TEXT NOT NULL DEFAULT 'product', review_status TEXT NOT NULL DEFAULT 'pending_review', merge_into_code TEXT, sources JSON NOT NULL DEFAULT '[]', aliases JSON NOT NULL DEFAULT '[]', product_usage_count INTEGER NOT NULL DEFAULT 0, value_count INTEGER NOT NULL DEFAULT 0, channel_map_count INTEGER NOT NULL DEFAULT 0, magento_frontend_input TEXT, magento_attribute_code TEXT, shopify_representation JSON NOT NULL DEFAULT '{}', conflict_reasons JSON NOT NULL DEFAULT '[]', proposed_action TEXT NOT NULL DEFAULT 'keep', duplicate_group TEXT, notes TEXT, applied_at TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ) """, """ CREATE TABLE IF NOT EXISTS v2_attribute_value_seed_candidate ( id INTEGER PRIMARY KEY AUTOINCREMENT, attribute_source_code TEXT NOT NULL, source_value_code TEXT NOT NULL, proposed_value_code TEXT NOT NULL, proposed_label TEXT NOT NULL, sort_order INTEGER NOT NULL DEFAULT 0, review_status TEXT NOT NULL DEFAULT 'pending_review', merge_into_value_code TEXT, aliases JSON NOT NULL DEFAULT '[]', product_usage_count INTEGER NOT NULL DEFAULT 0, lifecycle_status TEXT NOT NULL DEFAULT 'active', created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(attribute_source_code, source_value_code), FOREIGN KEY(attribute_source_code) REFERENCES v2_attribute_seed_candidate(source_code) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS v2_attribute_seed_mapping_candidate ( id INTEGER PRIMARY KEY AUTOINCREMENT, attribute_source_code TEXT NOT NULL, channel TEXT NOT NULL, channel_attribute_code TEXT, channel_attribute_id TEXT, representation JSON NOT NULL DEFAULT '{}', compatibility_status TEXT NOT NULL DEFAULT 'unresolved', is_manual BOOLEAN NOT NULL DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(channel, attribute_source_code), FOREIGN KEY(attribute_source_code) REFERENCES v2_attribute_seed_candidate(source_code) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS attribute_def ( attribute_code TEXT PRIMARY KEY, frontend_label TEXT, type TEXT, is_required BOOLEAN DEFAULT 0, allows_free_text BOOLEAN DEFAULT 0, options_json JSON, updated_at TEXT ) """, """ CREATE TABLE IF NOT EXISTS attribute_option ( id INTEGER PRIMARY KEY AUTOINCREMENT, attribute_code TEXT NOT NULL, option_value TEXT NOT NULL, created_at TEXT, UNIQUE(attribute_code, option_value) ) """, ): session.execute(text(stmt)) session.commit() def _add_definition(session: Session, code: str, **kwargs) -> MasterAttributeDefinition: row = MasterAttributeDefinition( attribute_code=code, label=kwargs.get("label") or code.replace("_", " ").title(), purpose=kwargs.get("purpose") or "presentation", data_type=kwargs.get("data_type") or "text", is_required=kwargs.get("is_required", False), is_active=kwargs.get("is_active", True), ) session.add(row) return row def _approve_all(session: Session) -> None: from db.v2_attribute_seed import candidate_ownership for row in session.scalars(select(V2AttributeSeedCandidate)).all(): if candidate_ownership(row.source_code, row.proposed_label, row.proposed_code) != "attribute": continue if row.review_status in {"excluded", "merged"}: continue approve_candidate(session, int(row.id)) def _discover_apply(session: Session) -> dict: seed_v2_attributes(session) session.flush() _approve_all(session) return apply_approved_attribute_seed(session) def test_seed_discovers_candidates_without_creating_live_attributes(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "door_style", label="Door Style", data_type="text") _add_definition(session, "variation_width_in", label="Width", data_type="number") _add_definition(session, "brand", label="Brand", data_type="text") vocab = ManualAttributeVocabulary( attribute_code="door_style", label="Door Style", is_locked=True, unknown_policy="drop", covered_codes_json=["door_style"], is_active=True, ) session.add(vocab) session.flush() session.add_all( [ ManualAttributeVocabularyValue(vocabulary_id=int(vocab.id), value="Shaker", is_active=True), ManualAttributeVocabularyValue(vocabulary_id=int(vocab.id), value="Slab", is_active=True), ] ) session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="door_style", channel_attribute_code="cabinet_door_style", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) session.commit() first = seed_v2_attributes(session) session.commit() codes = {row.source_code for row in session.scalars(select(V2AttributeSeedCandidate)).all()} door = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "door_style")).scalar_one() width = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "variation_width_in") ).scalar_one() values = session.execute( select(V2AttributeValueSeedCandidate).where(V2AttributeValueSeedCandidate.attribute_source_code == "door_style") ).scalars().all() assert {"door_style", "variation_width_in", "title", "description", "short_description", "features", "specifications"}.issubset(codes) assert "brand" not in codes assert first["candidates_created"] >= 7 assert door.proposed_data_type == "select" assert door.proposed_ui_control == "dropdown" assert door.review_status == "pending_review" assert width.is_variant_axis is True assert sorted(row.proposed_label for row in values) == ["Shaker", "Slab"] assert session.execute(select(V2Attribute)).scalars().all() == [] assert "candidates" in first["report"] second = seed_v2_attributes(session) session.flush() assert second["candidates_created"] == 0 assert {row.source_code for row in session.scalars(select(V2AttributeSeedCandidate)).all()} >= {"door_style", "title"} def test_approve_then_apply_creates_live_registry(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "door_style", label="Door Style", data_type="text") vocab = ManualAttributeVocabulary(attribute_code="door_style", label="Door Style", is_active=True) session.add(vocab) session.flush() session.add(ManualAttributeVocabularyValue(vocabulary_id=int(vocab.id), value="Shaker", is_active=True)) session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="door_style", channel_attribute_code="cabinet_door_style", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) session.commit() applied = _discover_apply(session) session.flush() door = session.execute(select(V2Attribute).where(V2Attribute.code == "door_style")).scalar_one() values = session.execute(select(V2AttributeValue).where(V2AttributeValue.attribute_id == door.id)).scalars().all() mapping = session.execute( select(V2ChannelAttributeMapping).where(V2ChannelAttributeMapping.canonical_attribute_id == door.id) ).scalar_one() assert applied["attributes_created"] == 1 assert door.data_type == "select" assert door.ui_control == "dropdown" assert [row.display_label for row in values] == ["Shaker"] assert mapping.channel_attribute_code == "cabinet_door_style" again = apply_approved_attribute_seed(session) assert again["attributes_created"] == 0 assert again["attributes_skipped_existing"] == 1 def test_exclude_and_merge_are_not_applied(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "door_style", label="Door Style", data_type="select") _add_definition(session, "doorstyle", label="Door Style", data_type="select") _add_definition(session, "legacy_notes", label="Legacy Notes", data_type="text") session.commit() seed_v2_attributes(session) session.flush() door = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "door_style")).scalar_one() duplicate = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "doorstyle")).scalar_one() notes = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "legacy_notes")).scalar_one() approve_candidate(session, int(door.id)) merge_candidate(session, int(duplicate.id), "door_style") exclude_candidate(session, int(notes.id)) apply_approved_attribute_seed(session) session.flush() codes = {row.code for row in session.scalars(select(V2Attribute)).all()} survivor = session.execute(select(V2Attribute).where(V2Attribute.code == "door_style")).scalar_one() assert codes == {"door_style"} assert "doorstyle" in (survivor.aliases or []) def test_bulk_merge_folds_values_and_mappings(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_or_rta", label="Assembled or RTA", data_type="select") _add_definition(session, "assembly_type", label="Assembly Type", data_type="select") session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="assembly_type", channel_attribute_code="assembly_type", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) session.commit() seed_v2_attributes(session) session.flush() survivor = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembled_or_rta") ).scalar_one() duplicate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembly_type") ).scalar_one() result = bulk_merge_candidates(session, [int(survivor.id), int(duplicate.id)], int(survivor.id)) assert result["survivor_code"] == "assembled_or_rta" session.flush() duplicate = session.get(V2AttributeSeedCandidate, int(duplicate.id)) survivor = session.get(V2AttributeSeedCandidate, int(survivor.id)) assert duplicate.review_status == "merged" assert duplicate.merge_into_code == "assembled_or_rta" assert "assembly_type" in (survivor.aliases or []) approve_candidate(session, int(survivor.id)) apply_approved_attribute_seed(session) session.flush() live = session.execute(select(V2Attribute).where(V2Attribute.code == "assembled_or_rta")).scalar_one() assert session.execute(select(V2Attribute).where(V2Attribute.code == "assembly_type")).scalar_one_or_none() is None assert "assembly_type" in (live.aliases or []) def test_merge_counts_distinct_skus_not_summed_sources(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_depth", label="Assembled Depth", data_type="select") _add_definition(session, "depth_in", label="Depth In", data_type="select") products = [ MasterProduct(sku="SKU-BOTH-1", name="Both 1", row_hash="b1", is_active=True), MasterProduct(sku="SKU-BOTH-2", name="Both 2", row_hash="b2", is_active=True), MasterProduct(sku="SKU-A-ONLY", name="A only", row_hash="a1", is_active=True), MasterProduct(sku="SKU-B-ONLY", name="B only", row_hash="c1", is_active=True), ] session.add_all(products) session.flush() session.add_all( [ MasterProductAttributeValue( product_id=int(products[0].id), sku="SKU-BOTH-1", attribute_code="assembled_depth", value="24" ), MasterProductAttributeValue( product_id=int(products[0].id), sku="SKU-BOTH-1", attribute_code="depth_in", value="24" ), MasterProductAttributeValue( product_id=int(products[1].id), sku="SKU-BOTH-2", attribute_code="assembled_depth", value="24" ), MasterProductAttributeValue( product_id=int(products[1].id), sku="SKU-BOTH-2", attribute_code="depth_in", value="24" ), MasterProductAttributeValue( product_id=int(products[2].id), sku="SKU-A-ONLY", attribute_code="assembled_depth", value="24" ), MasterProductAttributeValue( product_id=int(products[3].id), sku="SKU-B-ONLY", attribute_code="depth_in", value="24" ), ] ) session.commit() seed_v2_attributes(session) session.flush() survivor = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembled_depth") ).scalar_one() duplicate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "depth_in") ).scalar_one() assert survivor.product_usage_count == 3 assert duplicate.product_usage_count == 3 bulk_merge_candidates(session, [int(survivor.id), int(duplicate.id)], int(survivor.id)) session.flush() survivor = session.get(V2AttributeSeedCandidate, int(survivor.id)) assert survivor.product_usage_count == 4 seed_v2_attributes(session) session.flush() survivor = session.get(V2AttributeSeedCandidate, int(survivor.id)) assert survivor.product_usage_count == 4 def test_bulk_merge_overlapping_numeric_values_can_retarget_survivor(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) for code, label in ( ("assembled_depth", "Assembled Depth"), ("depth", "Depth"), ("depth_in", "Depth In"), ("variation_depth_in", "Variation Depth In"), ): _add_definition(session, code, label=label, data_type="select") product = MasterProduct(sku="SKU-DEPTH", name="Base", row_hash="d-overlap", is_active=True) session.add(product) session.flush() for code in ("assembled_depth", "depth", "depth_in", "variation_depth_in"): session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-DEPTH", attribute_code=code, value="22", ) ) session.commit() seed_v2_attributes(session) session.flush() by_code = { row.source_code: row for row in session.scalars(select(V2AttributeSeedCandidate)).all() } first = bulk_merge_candidates( session, [int(by_code["depth"].id), int(by_code["assembled_depth"].id), int(by_code["depth_in"].id)], int(by_code["depth"].id), ) assert first["survivor_code"] == "depth" session.flush() second = bulk_merge_candidates( session, [ int(by_code["assembled_depth"].id), int(by_code["depth"].id), int(by_code["depth_in"].id), int(by_code["variation_depth_in"].id), ], int(by_code["assembled_depth"].id), ) assert second["survivor_code"] == "assembled_depth" session.flush() survivor = session.get(V2AttributeSeedCandidate, int(by_code["assembled_depth"].id)) assert survivor.review_status in {"modified", "approved", "pending_review"} assert survivor.is_variant_axis is True codes = { row.source_value_code for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth" ) ).all() if row.review_status != "merged" } assert "22" in codes def test_overrides_are_reproducible(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "construction_type", label="Construction Type", data_type="select") session.commit() seed_v2_attributes(session) session.flush() apply_override_payload( session, { "attributes": { "construction_type": { "review_status": "approved", "proposed_data_type": "select", "proposed_ui_control": "radio", } } }, ) apply_approved_attribute_seed(session) session.flush() row = session.execute(select(V2Attribute).where(V2Attribute.code == "construction_type")).scalar_one() assert row.ui_control == "radio" assert row.data_type == "select" def test_seed_discovers_codes_from_product_values_and_plytix_options(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) product = MasterProduct(sku="SKU-1", name="Door", row_hash="abc", is_active=True) session.add(product) session.flush() session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-1", attribute_code="finish", value="Natural Oak", ) ) session.add( AttributeDef( attribute_code="finish", frontend_label="Finish", type="select", is_required=False, allows_free_text=False, options_json=["Natural Oak", "Espresso"], ) ) session.add(AttributeOption(attribute_code="finish", option_value="White")) session.commit() summary = seed_v2_attributes(session) session.flush() finish = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "finish")).scalar_one() labels = { row.proposed_label for row in session.execute( select(V2AttributeValueSeedCandidate).where(V2AttributeValueSeedCandidate.attribute_source_code == "finish") ).scalars().all() } assert summary["candidates_created"] >= 1 assert finish.proposed_label == "Finish" assert finish.proposed_data_type == "select" assert finish.product_usage_count == 1 assert labels == {"Natural Oak", "Espresso", "White"} def test_seed_does_not_overwrite_existing_v2_attribute(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) session.add(V2Attribute(code="finish", label="Operator Finish", data_type="text", is_channel_required=False)) _add_definition(session, "finish", label="Finish", data_type="select") session.commit() seed_v2_attributes(session) _approve_all(session) apply_approved_attribute_seed(session) session.flush() row = session.execute(select(V2Attribute).where(V2Attribute.code == "finish")).scalar_one() assert row.label == "Operator Finish" assert row.data_type == "text" def test_seed_maps_magento_registry_ids_and_options(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) now = datetime.now(timezone.utc) _add_definition(session, "door_style", label="Door Style", data_type="select") vocab = ManualAttributeVocabulary(attribute_code="door_style", label="Door Style", is_active=True) session.add(vocab) session.flush() session.add(ManualAttributeVocabularyValue(vocabulary_id=int(vocab.id), value="Shaker", is_active=True)) session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="door_style", channel_attribute_code="cabinet_door_style", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) connection = MagentoConnection( store_base_url="https://example.test", environment="staging", consumer_key="k", consumer_secret="s", access_token="t", access_token_secret="ts", status="active", ) session.add(connection) session.flush() session.add( MagentoAttributeRegistry( connection_id=int(connection.id), attribute_code="cabinet_door_style", attribute_id=44, frontend_label="Door Style", frontend_input="select", fetched_at=now, ) ) session.add( MagentoAttributeOptionRegistry( connection_id=int(connection.id), attribute_code="cabinet_door_style", option_label="Shaker", option_label_norm="shaker", value_index=91, fetched_at=now, ) ) session.commit() _discover_apply(session) session.flush() door = session.execute(select(V2Attribute).where(V2Attribute.code == "door_style")).scalar_one() mapping = session.execute( select(V2ChannelAttributeMapping).where(V2ChannelAttributeMapping.canonical_attribute_id == door.id) ).scalar_one() value = session.execute( select(V2AttributeValue).where(V2AttributeValue.attribute_id == door.id, V2AttributeValue.display_label == "Shaker") ).scalar_one() option_map = session.execute( select(V2ChannelAttributeValueMapping).where( V2ChannelAttributeValueMapping.canonical_attribute_value_id == value.id ) ).scalar_one() assert mapping.channel_attribute_id == "44" assert mapping.representation["frontend_input"] == "select" assert option_map.channel == "magento" assert option_map.channel_option_id == "91" def test_approve_rejects_unmapped_select_values(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "door_style", label="Door Style", data_type="select") vocab = ManualAttributeVocabulary(attribute_code="door_style", label="Door Style", is_active=True) session.add(vocab) session.flush() session.add(ManualAttributeVocabularyValue(vocabulary_id=int(vocab.id), value="Shaker", is_active=True)) product = MasterProduct(sku="SKU-X", name="Door", row_hash="x", is_active=True) session.add(product) session.flush() session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-X", attribute_code="door_style", value="UnknownCustomValue", ) ) session.commit() seed_v2_attributes(session) session.flush() candidate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "door_style") ).scalar_one() custom = session.execute( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "door_style", V2AttributeValueSeedCandidate.proposed_label == "UnknownCustomValue", ) ).scalar_one() custom.review_status = "excluded" session.flush() from db.v2_attribute_review import AttributeReviewError try: approve_candidate(session, int(candidate.id)) assert False, "expected AttributeReviewError" except AttributeReviewError as exc: assert "UnknownCustomValue" in str(exc) assert session.get(V2AttributeSeedCandidate, candidate.id).review_status == "conflict" def test_seed_reset_rebuilds_candidates_not_live_rows(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) session.add(V2Attribute(code="stale", label="Stale", data_type="text")) _add_definition(session, "finish", label="Finish", data_type="text") session.commit() seed_v2_attributes(session, reset=True) session.flush() live_codes = {row.code for row in session.execute(select(V2Attribute)).scalars().all()} candidate_codes = {row.source_code for row in session.execute(select(V2AttributeSeedCandidate)).scalars().all()} assert live_codes == {"stale"} assert "finish" in candidate_codes assert "title" in candidate_codes def test_merge_variation_depth_into_depth_keeps_variant_axis(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "depth", label="Depth", data_type="select") _add_definition(session, "variation_depth_in", label="Variation Depth In", data_type="text") session.commit() seed_v2_attributes(session) session.flush() survivor = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "depth")).scalar_one() axis = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "variation_depth_in") ).scalar_one() assert axis.is_variant_axis is True merge_candidate(session, int(axis.id), "depth") session.flush() survivor = session.get(V2AttributeSeedCandidate, int(survivor.id)) assert survivor.is_variant_axis is True assert "variation_depth_in" in (survivor.aliases or []) def test_numeric_display_rename_becomes_option_code(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "depth", label="Depth", data_type="select") product = MasterProduct(sku="SKU-D", name="Base", row_hash="d1", is_active=True) session.add(product) session.flush() session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-D", attribute_code="depth", value="3390", ) ) session.commit() seed_v2_attributes(session) session.flush() value = session.execute( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "depth", V2AttributeValueSeedCandidate.source_value_code == "3390", ) ).scalar_one() from db.v2_attribute_review import update_value update_value(session, int(value.id), {"proposed_label": "90"}) session.flush() value = session.get(V2AttributeValueSeedCandidate, int(value.id)) assert value.proposed_label == "90" assert value.proposed_value_code == "90" assert value.source_value_code == "3390" candidate = session.execute(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "depth")).scalar_one() approve_candidate(session, int(candidate.id)) apply_approved_attribute_seed(session) session.flush() live = session.execute(select(V2Attribute).where(V2Attribute.code == "depth")).scalar_one() option = session.execute( select(V2AttributeValue).where(V2AttributeValue.attribute_id == live.id, V2AttributeValue.value_code == "90") ).scalar_one() assert option.display_label == "90" assert "3390" in (option.aliases or []) assert live.is_variant_axis is True def test_bulk_merge_values_folds_fraction_aliases(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_depth", label="Assembled Depth", data_type="select") first = MasterProduct(sku="SKU-FRAC-A", name="Filler A", row_hash="f1", is_active=True) second = MasterProduct(sku="SKU-FRAC-B", name="Filler B", row_hash="f2", is_active=True) session.add_all([first, second]) session.flush() session.add_all( [ MasterProductAttributeValue( product_id=int(first.id), sku="SKU-FRAC-A", attribute_code="assembled_depth", value="0.75" ), MasterProductAttributeValue( product_id=int(second.id), sku="SKU-FRAC-B", attribute_code="assembled_depth", value='3/4"' ), ] ) session.commit() seed_v2_attributes(session) session.flush() values = list( session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth" ) ).all() ) by_label = {row.proposed_label: row for row in values} decimal = by_label["0.75"] fraction = next(row for row in values if "3/4" in row.proposed_label or row.source_value_code == "3_4") result = bulk_merge_values(session, [int(decimal.id), int(fraction.id)], int(decimal.id)) session.flush() fraction = session.get(V2AttributeValueSeedCandidate, int(fraction.id)) decimal = session.get(V2AttributeValueSeedCandidate, int(decimal.id)) assert result["survivor_code"] in {"0_75", "0.75", "075"} assert fraction.review_status == "merged" assert "3_4" in (decimal.aliases or []) or fraction.source_value_code in (decimal.aliases or []) candidate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembled_depth") ).scalar_one() approve_candidate(session, int(candidate.id)) apply_approved_attribute_seed(session) session.flush() live = session.execute(select(V2Attribute).where(V2Attribute.code == "assembled_depth")).scalar_one() option = session.execute( select(V2AttributeValue).where(V2AttributeValue.attribute_id == live.id, V2AttributeValue.display_label == "0.75") ).scalar_one() aliases = option.aliases or [] assert any("3" in str(item) and "4" in str(item) for item in aliases) or fraction.source_value_code in aliases def test_bulk_merge_values_across_merged_attribute_sources(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "color", label="Color", data_type="select") _add_definition(session, "finish_color", label="Finish Color", data_type="select") products = [ MasterProduct(sku="SKU-C1", name="C1", row_hash="c1", is_active=True), MasterProduct(sku="SKU-F1", name="F1", row_hash="f1", is_active=True), MasterProduct(sku="SKU-F2", name="F2", row_hash="f2", is_active=True), MasterProduct(sku="SKU-C2", name="C2", row_hash="c2", is_active=True), ] session.add_all(products) session.flush() session.add_all( [ MasterProductAttributeValue( product_id=int(products[0].id), sku="SKU-C1", attribute_code="color", value="Navy" ), MasterProductAttributeValue( product_id=int(products[1].id), sku="SKU-F1", attribute_code="finish_color", value="Navy" ), MasterProductAttributeValue( product_id=int(products[2].id), sku="SKU-F2", attribute_code="finish_color", value="Navy Blue" ), MasterProductAttributeValue( product_id=int(products[3].id), sku="SKU-C2", attribute_code="color", value="Blue" ), ] ) session.commit() seed_v2_attributes(session) session.flush() color = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "color") ).scalar_one() finish = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "finish_color") ).scalar_one() bulk_merge_candidates(session, [int(color.id), int(finish.id)], int(color.id)) session.flush() leftover = V2AttributeValueSeedCandidate( attribute_source_code="finish_color", source_value_code="navy_from_finish", proposed_value_code="navy_from_finish", proposed_label="Navy", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) session.add(leftover) session.flush() navy = next( row for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "color" ) ).all() if row.proposed_label == "Navy" and row.review_status != "merged" ) navy_blue = next( row for row in session.scalars(select(V2AttributeValueSeedCandidate)).all() if row.proposed_label == "Navy Blue" and row.review_status != "merged" ) blue = next( row for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "color" ) ).all() if row.proposed_label == "Blue" and row.review_status != "merged" ) result = bulk_merge_values( session, [int(navy.id), int(leftover.id), int(navy_blue.id), int(blue.id)], int(navy.id), ) session.flush() leftover = session.get(V2AttributeValueSeedCandidate, int(leftover.id)) navy_blue = session.get(V2AttributeValueSeedCandidate, int(navy_blue.id)) blue = session.get(V2AttributeValueSeedCandidate, int(blue.id)) navy = session.get(V2AttributeValueSeedCandidate, int(navy.id)) assert leftover.review_status == "merged" assert navy_blue.review_status == "merged" assert blue.review_status == "merged" assert navy.review_status != "merged" assert int(navy.id) == int(result["survivor_id"]) assert "navy_from_finish" in (navy.aliases or []) def test_rediscover_after_attribute_merge_keeps_values_mergeable(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "color", label="Color", data_type="select") _add_definition(session, "finish_color", label="Finish Color", data_type="select") product = MasterProduct(sku="SKU-CF", name="Color Finish", row_hash="cf1", is_active=True) session.add(product) session.flush() session.add_all( [ MasterProductAttributeValue( product_id=int(product.id), sku="SKU-CF", attribute_code="color", value="Navy" ), MasterProductAttributeValue( product_id=int(product.id), sku="SKU-CF", attribute_code="finish_color", value="Navy Blue" ), ] ) session.commit() seed_v2_attributes(session) session.flush() color = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "color") ).scalar_one() finish = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "finish_color") ).scalar_one() bulk_merge_candidates(session, [int(color.id), int(finish.id)], int(color.id)) session.flush() seed_v2_attributes(session) session.flush() detail = candidate_detail(session, session.get(V2AttributeSeedCandidate, int(color.id))) pending = [row for row in detail["values"] if row["review_status"] != "merged"] assert pending survivor = next(row for row in pending if row["proposed_label"] == "Navy") ids = [int(row["id"]) for row in pending] if len(ids) >= 2: result = bulk_merge_values(session, ids, int(survivor["id"])) session.flush() assert int(result["survivor_id"]) == int(survivor["id"]) def test_approve_accepts_inch_fractions_merged_into_decimals(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_depth", label="Assembled Depth", data_type="select") decimal_product = MasterProduct(sku="SKU-125", name="A", row_hash="a1", is_active=True) fraction_product = MasterProduct(sku="SKU-118", name="B", row_hash="b1", is_active=True) session.add_all([decimal_product, fraction_product]) session.flush() session.add_all( [ MasterProductAttributeValue( product_id=int(decimal_product.id), sku="SKU-125", attribute_code="assembled_depth", value="1.125", ), MasterProductAttributeValue( product_id=int(fraction_product.id), sku="SKU-118", attribute_code="assembled_depth", value='1 1/8"', ), ] ) session.commit() seed_v2_attributes(session) session.flush() values = list( session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth" ) ).all() ) decimal = next(row for row in values if row.proposed_label == "1.125" or row.source_value_code == "1_125") fraction = next(row for row in values if "1/8" in row.proposed_label or row.source_value_code == "1_1_8") bulk_merge_values(session, [int(decimal.id), int(fraction.id)], int(decimal.id)) session.flush() candidate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembled_depth") ).scalar_one() approved = approve_candidate(session, int(candidate.id)) assert approved.review_status == "approved" def test_create_option_and_merge_magento_ids_allows_approve(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_depth", label="Assembled Depth", data_type="select") products = [ MasterProduct(sku="SKU-21-5", name="A", row_hash="a1", is_active=True), MasterProduct(sku="SKU-84", name="B", row_hash="b1", is_active=True), MasterProduct(sku="SKU-90", name="C", row_hash="c1", is_active=True), MasterProduct(sku="SKU-96", name="D", row_hash="d1", is_active=True), ] session.add_all(products) session.flush() session.add_all( [ MasterProductAttributeValue( product_id=int(products[0].id), sku="SKU-21-5", attribute_code="assembled_depth", value="21.5" ), MasterProductAttributeValue( product_id=int(products[1].id), sku="SKU-84", attribute_code="assembled_depth", value="3384" ), MasterProductAttributeValue( product_id=int(products[2].id), sku="SKU-90", attribute_code="assembled_depth", value="3390" ), MasterProductAttributeValue( product_id=int(products[3].id), sku="SKU-96", attribute_code="assembled_depth", value="3396" ), ] ) session.commit() seed_v2_attributes(session) session.flush() candidate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembled_depth") ).scalar_one() for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth" ) ).all(): if row.source_value_code in {"21_5", "21.5", "3384", "3390", "3396"} or row.proposed_label in { "21.5", "3384", "3390", "3396", }: row.review_status = "excluded" session.flush() from db.v2_attribute_review import AttributeReviewError try: approve_candidate(session, int(candidate.id)) assert False, "expected AttributeReviewError" except AttributeReviewError as exc: assert "21.5" in str(exc) assert "3390" in str(exc) report = unmapped_product_value_report(session, candidate) by_value = {item["value"]: item for item in report} assert by_value["21.5"]["reason"] == "excluded" assert by_value["3390"]["looks_like_magento_option_id"] is True created = create_value(session, int(candidate.id), {"proposed_label": "21.5"}) assert created.proposed_label == "21.5" create_and_merge_values(session, int(candidate.id), label="84", merge_values=["3384"]) create_and_merge_values(session, int(candidate.id), label="90", merge_values=["3390"]) create_and_merge_values(session, int(candidate.id), label="96", merge_values=["3396"]) session.flush() ninety = session.execute( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth", V2AttributeValueSeedCandidate.proposed_label == "90", ) ).scalar_one() assert "3390" in (ninety.aliases or []) or ninety.source_value_code == "3390" approved = approve_candidate(session, int(candidate.id)) assert approved.review_status == "approved" def test_create_value_retargets_existing_option(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_depth", label="Assembled Depth", data_type="select") product = MasterProduct(sku="SKU-KEEP", name="Keep", row_hash="k1", is_active=True) session.add(product) session.flush() session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-KEEP", attribute_code="assembled_depth", value="24" ) ) session.commit() seed_v2_attributes(session) session.flush() candidate = session.execute( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "assembled_depth") ).scalar_one() existing = session.execute( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth" ) ).scalar_one() created = create_value(session, int(candidate.id), {"proposed_label": "24"}) assert int(created.id) == int(existing.id) assert created.proposed_value_code == "24" def test_display_rename_forces_fraction_value_code(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "assembled_depth", label="Assembled Depth", data_type="select") product = MasterProduct(sku="SKU-215", name="Half", row_hash="h1", is_active=True) session.add(product) session.flush() session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-215", attribute_code="assembled_depth", value="21.5" ) ) session.commit() seed_v2_attributes(session) session.flush() value = session.execute( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "assembled_depth" ) ).scalar_one() from db.v2_attribute_review import update_value updated = update_value(session, int(value.id), {"proposed_label": "21 1/2"}) assert updated.proposed_label == "21 1/2" assert updated.proposed_value_code == "21_1_2" assert updated.source_value_code in {"21.5", "21_5"} assert updated.source_value_code in (updated.aliases or []) or "21.5" in (updated.aliases or []) or "21_5" in ( updated.aliases or [] ) forced = update_value(session, int(value.id), {"proposed_value_code": "twenty_one_half"}) assert forced.proposed_value_code == "twenty_one_half" assert "21_1_2" in (forced.aliases or []) def test_weight_mappings_are_compatible_with_decimal(): from db.v2_attribute_types import channel_compatibility magento = channel_compatibility( "decimal", "magento", {"channel_attribute_code": "weight", "frontend_input": "weight"}, ) shopify = channel_compatibility( "decimal", "shopify", {"channel_attribute_code": "weight", "metafield_type": "single_line_text_field"}, ) assert magento == "ok" assert shopify == "ok" def test_saving_decimal_clears_weight_type_conflict_and_incompatible_maps(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) now = datetime.now(timezone.utc) candidate = V2AttributeSeedCandidate( source_code="assembled_weight", proposed_code="assembled_weight", proposed_label="Assembled Weight", discovered_data_type="text", proposed_data_type="text", proposed_ui_control="text_field", review_status="pending_review", conflict_reasons=["source types disagree: decimal, text (proposed text)"], sources=["magento", "master_attribute_definition"], aliases=[], shopify_representation={}, created_at=now, updated_at=now, ) session.add(candidate) session.flush() session.add_all( [ V2AttributeSeedMappingCandidate( attribute_source_code="assembled_weight", channel="magento", channel_attribute_code="weight", representation={"frontend_input": "weight", "attribute_code": "weight"}, compatibility_status="incompatible", created_at=now, updated_at=now, ), V2AttributeSeedMappingCandidate( attribute_source_code="assembled_weight", channel="shopify", channel_attribute_code="weight", representation={"attribute_code": "weight", "metafield_type": "single_line_text_field"}, compatibility_status="incompatible", created_at=now, updated_at=now, ), ] ) session.flush() update_candidate( session, int(candidate.id), {"proposed_data_type": "decimal", "proposed_ui_control": "number_input"}, ) session.flush() detail = candidate_detail(session, session.get(V2AttributeSeedCandidate, int(candidate.id))) assert detail["candidate"]["review_status"] == "modified" assert detail["candidate"]["conflict_reasons"] == [] by_channel = {row["channel"]: row["compatibility_status"] for row in detail["mappings"]} assert by_channel["magento"] == "ok" assert by_channel["shopify"] == "ok" def test_discover_excludes_system_variant_fields(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "variants_list", label="Variants List", data_type="text") session.flush() stats = seed_v2_attributes(session) candidate = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "variants_list") ) assert candidate is not None assert candidate.review_status == "excluded" assert "system/dynamic" in str(candidate.notes or "").lower() assert stats["candidates_created"] >= 1 assert stats["review_status_counts"].get("excluded", 0) >= 1 def test_visibility_is_channel_policy_and_skipped_on_apply(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "visibility", label="Visibility", data_type="select") session.flush() seed_v2_attributes(session) candidate = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "visibility") ) assert candidate is not None assert candidate.review_status == "excluded" assert "channel listing policy" in str(candidate.notes or "").lower() or "shopify" in str(candidate.notes or "").lower() candidate.review_status = "approved" candidate.proposed_action = "keep" session.flush() apply_approved_attribute_seed(session) session.flush() assert session.scalar(select(V2Attribute).where(V2Attribute.code == "visibility")) is None def test_content_fields_are_seeded_without_option_values(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "description", label="Description", data_type="textarea") _add_definition(session, "door_style", label="Door Style", data_type="select") product = MasterProduct(sku="SKU-DESC", name="Base Cabinet", row_hash="d1", is_active=True) session.add(product) session.flush() session.add( MasterProductAttributeValue( product_id=int(product.id), sku="SKU-DESC", attribute_code="description", value="

A unique HTML description that must not become an option.

", ) ) session.commit() seed_v2_attributes(session) session.flush() by_code = { row.source_code: row for row in session.scalars(select(V2AttributeSeedCandidate)).all() } assert by_code["title"].proposed_label == "Product Title" assert by_code["description"].proposed_data_type == "html" assert by_code["short_description"].proposed_data_type == "html" assert by_code["features"].review_status == "pending_review" assert "overlay" in (by_code["features"].notes or "") assert not list( session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "description" ) ).all() ) title_maps = { row.channel: row.channel_attribute_code for row in session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == "title" ) ).all() } assert title_maps.get("magento") == "name" assert title_maps.get("shopify") == "title" approve_candidate(session, int(by_code["title"].id)) approve_candidate(session, int(by_code["short_description"].id)) apply_approved_attribute_seed(session) session.flush() assert session.scalar(select(V2Attribute).where(V2Attribute.code == "title")) is None short_desc = session.scalar(select(V2Attribute).where(V2Attribute.code == "short_description")) assert short_desc is not None assert short_desc.data_type == "html" def test_discover_loads_master_taxonomy_for_categories(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "categories", label="Categories", data_type="select") session.add_all( [ MasterTaxonomyNode( node_kind="category", code="living-room-sofas", name="Living Room Sofas", path_slug="furniture/living-room-sofas", sort_order=10, is_active=True, ), MasterTaxonomyNode( node_kind="category", code="retired-hutches", name="Retired Hutches", path_slug="furniture/retired-hutches", sort_order=20, is_active=False, ), ] ) session.flush() seed_v2_attributes(session) candidate = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "categories") ) assert candidate is not None assert candidate.review_status == "pending_review" assert candidate.proposed_data_type == "select" assert "taxonomy" in str(candidate.notes or "").lower() labels = { str(row.proposed_label) for row in session.scalars( select(V2AttributeValueSeedCandidate).where(V2AttributeValueSeedCandidate.attribute_source_code == "categories") ) } assert "Living Room Sofas" in labels assert "Retired Hutches" not in labels approve_candidate(session, int(candidate.id)) apply_approved_attribute_seed(session) live = session.scalar(select(V2Attribute).where(V2Attribute.code == "categories")) assert live is None assert candidate.review_status == "approved" assert "taxonomy" in str(candidate.notes or "").lower() def test_discover_marks_collection_landing_as_category_scope(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) now = datetime.now(timezone.utc) session.add( MagentoConnection( tenant_id="test", store_base_url="https://magento.test", magento_api_base_url="https://magento.test", rest_base_path="V1", environment="test", consumer_key="ck", consumer_secret="cs", access_token="at", access_token_secret="ats", status="active", ) ) session.flush() connection = session.scalar(select(MagentoConnection)) session.add( MagentoAttributeRegistry( connection_id=int(connection.id), attribute_code="hs_starting_price", attribute_id=4401, frontend_label="Starting Price", frontend_input="text", backend_type="varchar", is_user_defined=True, fetched_at=now, ) ) _add_definition(session, "door_style", label="Door Style", data_type="text") session.flush() seed_v2_attributes(session) landing = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "hs_starting_price") ) product = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "door_style") ) assert landing is not None assert landing.owner_scope == "category" assert product.owner_scope == "product" def test_operator_can_assign_and_preserve_channel_mapping(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) _add_definition(session, "average_price_per_linear_foot", label="Average Price Per Linear Foot", data_type="text") session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="average_price_per_linear_foot", channel_attribute_code="wrong_code", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) session.flush() seed_v2_attributes(session) session.flush() candidate = session.scalar( select(V2AttributeSeedCandidate).where( V2AttributeSeedCandidate.source_code == "average_price_per_linear_foot" ) ) assert candidate is not None mapping = upsert_candidate_mapping( session, int(candidate.id), {"channel": "magento", "channel_attribute_code": "dc_sample_group"}, ) assert mapping.channel_attribute_code == "dc_sample_group" assert mapping.is_manual is True update_candidate(session, int(candidate.id), {"owner_scope": "product"}) seed_v2_attributes(session) kept = session.scalar( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == "average_price_per_linear_foot", V2AttributeSeedMappingCandidate.channel == "magento", ) ) assert kept is not None assert kept.channel_attribute_code == "dc_sample_group" assert kept.is_manual is True refreshed = session.get(V2AttributeSeedCandidate, int(candidate.id)) assert refreshed.owner_scope == "product" def test_canonicalize_collection_family_keeps_collections_and_accessories(): collections = {"anna snow white": "Anna Snow White", "harriet": "Harriet"} families = {"cabinets": "Cabinets", "vanities": "Vanities"} assert ( canonicalize_collection_family_label("Kitchen Cabinets / Anna Snow White", collections, families) == "Anna Snow White" ) assert canonicalize_collection_family_label("Kitchen Cabinets Anna Snow White", collections, families) == "Anna Snow White" assert canonicalize_collection_family_label("Anna Snow White", collections, families) == "Anna Snow White" assert canonicalize_collection_family_label("Bathroom Vanities / Harriet", collections, families) == "Harriet" assert canonicalize_collection_family_label("Snow White - Accessories", collections, families) == "Snow White Accessories" assert canonicalize_collection_family_label("Snow White Accessories", collections, families) == "Snow White Accessories" assert canonicalize_collection_family_label("Cabinets", collections, families) is None assert ( canonicalize_collection_family_label("Kitchen Cabinets / Base Cabinets / Sink Base", collections, families) is None ) assert ( canonicalize_collection_family_label("Kitchen Cabinets / Accessories / Crown Molding", collections, families) is None ) def test_family_seed_collapses_prefixed_collections_and_keeps_accessories(catalog_intent_session: Session): session = catalog_intent_session _create_v2_attribute_tables(session) now = datetime.now(timezone.utc) _add_definition(session, "family", label="Family", data_type="select") _add_definition(session, "product_family", label="Product Family", data_type="select") session.add_all( [ MasterCollectionRegistry(code="asw", name="Anna Snow White", is_active=True), MasterCollectionRegistry(code="harriet", name="Harriet", is_active=True), MasterProductFamily(code="cabinets", name="Cabinets", is_active=True), MasterProductFamily(code="vanities", name="Vanities", is_active=True), ] ) session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="family", channel_attribute_code="family", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) session.add( ChannelAttributeAlias( channel_code="magento", canonical_code="product_family", channel_attribute_code="family", mapping_scope="dynamic", action="use_existing", is_active=True, ) ) connection = MagentoConnection( tenant_id="test", store_base_url="https://magento.test", magento_api_base_url="https://magento.test", rest_base_path="V1", environment="test", consumer_key="ck", consumer_secret="cs", access_token="at", access_token_secret="ats", status="active", ) session.add(connection) session.flush() session.add( MagentoAttributeRegistry( connection_id=int(connection.id), attribute_code="family", attribute_id=88, frontend_label="Family", frontend_input="select", fetched_at=now, ) ) for label, option_id in ( ("Anna Snow White", 1), ("Kitchen Cabinets / Anna Snow White", 2), ("Kitchen Cabinets / Base Cabinets / Sink Base", 3), ("Cabinets", 4), ("Snow White - Accessories", 5), ("Bathroom Vanities / Harriet", 6), ): session.add( MagentoAttributeOptionRegistry( connection_id=int(connection.id), attribute_code="family", option_label=label, option_label_norm=label.casefold(), value_index=option_id, fetched_at=now, ) ) session.commit() seed_v2_attributes(session) session.flush() family = session.scalar(select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "family")) assert family is not None assert "collection" in str(family.notes or "").lower() family_values = list( session.scalars( select(V2AttributeValueSeedCandidate).where(V2AttributeValueSeedCandidate.attribute_source_code == "family") ).all() ) live_labels = { str(row.proposed_label) for row in family_values if row.review_status not in {"merged", "excluded"} } assert live_labels == {"Anna Snow White", "Harriet", "Snow White Accessories"} anna = next(row for row in family_values if row.proposed_label == "Anna Snow White") anna_aliases = {str(item).casefold() for item in (anna.aliases or [])} assert any("kitchen cabinets" in item and "anna snow white" in item for item in anna_aliases) excluded = {str(row.proposed_label) for row in family_values if row.review_status == "excluded"} assert "Cabinets" in excluded or not any( str(row.proposed_label) == "Cabinets" and row.review_status == "pending_review" for row in family_values ) assert not any("Base Cabinets" in str(row.proposed_label) and row.review_status == "pending_review" for row in family_values) product_family = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == "product_family") ) assert product_family is not None product_family_labels = { str(row.proposed_label) for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == "product_family" ) ) if row.review_status not in {"merged", "excluded"} } assert product_family_labels == {"Cabinets", "Vanities"} assert "Anna Snow White" not in product_family_labels