"""Tests for MASTER_TAXONOMY_AUTO_INVENT freeze policy.""" from contextlib import contextmanager from unittest.mock import patch import pytest from sqlalchemy import select from app.jobs.sync_master_taxonomy_pipeline import run_master_taxonomy_pipeline from db.master_taxonomy_hierarchy import build_hierarchy_from_master_catalog from db.master_taxonomy_intent import apply_taxonomy_proposals from db.master_taxonomy_registry import backfill_master_taxonomy_from_products, upsert_collection_registry from db.models import MasterCollectionRegistry, MasterProduct, MasterTaxonomyNode from db.taxonomy_invent_policy import ( can_auto_invent, coerce_taxonomy_mode_when_frozen, taxonomy_auto_invent_enabled, ) def test_can_auto_invent_defaults_frozen(monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) assert can_auto_invent() is False assert can_auto_invent(allow_invent=True) is True assert can_auto_invent(allow_invent=False) is False def test_coerce_full_to_link_only_when_frozen(monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) mode, requested, coerced = coerce_taxonomy_mode_when_frozen("full") assert mode == "link_only" assert requested == "full" assert coerced is True def test_coerce_full_allowed_when_env_unlocks(monkeypatch): monkeypatch.setenv("MASTER_TAXONOMY_AUTO_INVENT", "true") mode, requested, coerced = coerce_taxonomy_mode_when_frozen("full") assert mode == "full" assert requested == "full" assert coerced is False def test_pipeline_records_enforcement_note(monkeypatch, catalog_intent_session): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) session = catalog_intent_session @contextmanager def _session_scope(): yield session with ( patch("app.jobs.sync_master_taxonomy_pipeline.get_session", _session_scope), patch( "app.jobs.sync_master_taxonomy_pipeline._active_connection_id", return_value=None, ), patch( "app.jobs.master_taxonomy_enqueue._resolve_connections", return_value=(None, None, None, None, None), ), patch( "db.master_taxonomy_sync.link_products_to_taxonomy_from_master", return_value={"linked": 0}, ), patch( "app.jobs.sync_master_taxonomy_pipeline._push_products_for_skus", return_value={"sku_count": 0}, ), ): result = run_master_taxonomy_pipeline( dry_run=True, taxonomy_mode="full", pull_remotes=False, create_missing_remotes=False, push_products=False, ) assert result["taxonomy_mode"] == "link_only" assert result["taxonomy_mode_requested"] == "full" assert result["taxonomy_mode_enforced"] == "link_only" assert "frozen" in (result.get("note") or "").lower() def test_hierarchy_frozen_does_not_invent_unknown_collection(catalog_intent_session, monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) session = catalog_intent_session session.add( MasterProduct( sku="NEW-COLL-1", name="Base", brand="Port & Bell", product_family="Cabinets", category_l1="Kitchen Cabinets", collection="Brand New Invented Collection XYZ", row_hash="h-new", is_active=True, ) ) session.flush() before_colls = list(session.scalars(select(MasterCollectionRegistry)).all()) stats = build_hierarchy_from_master_catalog(session, dry_run=False) session.flush() assert stats["invent_frozen"] is True assert stats["nodes_created"] == 0 after_colls = list(session.scalars(select(MasterCollectionRegistry)).all()) assert len(after_colls) == len(before_colls) assert ( session.scalar( select(MasterCollectionRegistry).where( MasterCollectionRegistry.name == "Brand New Invented Collection XYZ" ) ) is None ) def test_hierarchy_unlock_allows_invent(catalog_intent_session, monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) session = catalog_intent_session session.add( MasterProduct( sku="UNLOCK-1", name="Base", brand="Port & Bell", product_family="Cabinets", category_l1="Kitchen Cabinets", collection="Anna Snow White", row_hash="h-unlock", is_active=True, ) ) session.flush() stats = build_hierarchy_from_master_catalog(session, dry_run=False, allow_invent=True) session.flush() assert stats["invent_frozen"] is False assert stats["nodes_created"] >= 1 nodes = list(session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True))).all()) assert any("anna-snow-white" in (n.path_slug or "") or "Anna Snow White" in (n.name or "") for n in nodes) # Explicit registry invent (dashboard-style) still works under freeze when unlocked. row = upsert_collection_registry( session, {"name": "Dashboard Unlock Coll", "code": "dashboard-unlock-coll", "category_l1": "Kitchen Cabinets"}, allow_invent=True, ) assert row["id"] def test_backfill_frozen_links_existing_only(catalog_intent_session, monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) session = catalog_intent_session session.add( MasterCollectionRegistry( code="existing-coll", name="Existing Coll", path_slug="kitchen-cabinets/existing-coll", is_active=True, ) ) session.add( MasterProduct( sku="BF-1", name="P", row_hash="bf1", is_active=True, category_l1="Kitchen Cabinets", collection="Existing Coll", ) ) session.add( MasterProduct( sku="BF-2", name="P2", row_hash="bf2", is_active=True, category_l1="Kitchen Cabinets", collection="Should Not Invent", ) ) session.flush() result = backfill_master_taxonomy_from_products(session, dry_run=False) assert result["invent_frozen"] is True assert result["collections_created"] == 0 assert ( session.scalar( select(MasterCollectionRegistry).where(MasterCollectionRegistry.name == "Should Not Invent") ) is None ) product = session.scalar(select(MasterProduct).where(MasterProduct.sku == "BF-1")) assert product.collection_registry_id is not None def test_apply_proposals_frozen_skips_materialize(catalog_intent_session, monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) session = catalog_intent_session result = apply_taxonomy_proposals( session, [{"path_slug": "kitchen-cabinets/new-proposed", "name": "New Proposed", "node_kind": "collection"}], dry_run=False, ) assert result["invent_frozen"] is True assert result["created_count"] == 0 assert ( session.scalar( select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == "kitchen-cabinets/new-proposed") ) is None ) def test_upsert_collection_registry_intentional_api_allows_create(catalog_intent_session, monkeypatch): monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) session = catalog_intent_session row = upsert_collection_registry( session, {"name": "Dashboard Created Coll", "code": "dashboard-created-coll", "category_l1": "Kitchen Cabinets"}, allow_invent=True, ) assert row["id"] assert row["name"] == "Dashboard Created Coll" def test_coerce_remote_provision_when_frozen(monkeypatch): from db.taxonomy_invent_policy import coerce_remote_provision_when_frozen monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) effective, coerced = coerce_remote_provision_when_frozen(True) assert effective is False assert coerced is True effective, coerced = coerce_remote_provision_when_frozen(True, allow_provision=True) assert effective is True assert coerced is False def test_ensure_magento_category_path_refuses_create_when_frozen(monkeypatch): from db.channel_catalog_provision import _ensure_magento_category_path monkeypatch.delenv("MASTER_TAXONOMY_AUTO_INVENT", raising=False) class _Repo: def resolve_path(self, *_a, **_k): return None def upsert_category(self, *_a, **_k): return None class _Api: def search_categories(self, **_k): return 200, [] def create_category(self, *_a, **_k): raise AssertionError("create_category must not run while frozen") monkeypatch.setattr( "db.channel_catalog_provision.SqlAlchemyCategoryRegistryRepository", lambda _session: _Repo(), ) with pytest.raises(ValueError, match="Remote taxonomy provision is frozen"): _ensure_magento_category_path( session=object(), api=_Api(), connection_id=1, path="Kitchen Cabinets/Brand New Polluted", )