from datetime import datetime, timezone

from sqlalchemy import select

from db.master_taxonomy_sync import (
    assign_skus_to_taxonomy_node,
    bootstrap_master_taxonomy,
    coerce_taxonomy_node_ids,
    ensure_collection_landing_link,
    find_taxonomy_channel_candidates,
    link_products_to_taxonomy_from_master,
    link_taxonomy_node_channel,
    push_taxonomy_nodes_to_channels,
    reparent_taxonomy_node,
    _assign_global_root_channel_memberships,
    _should_replace_channel_taxonomy_assignment,
    upsert_taxonomy_node,
)
from db.models import (
    CollectionLandingPage,
    MagentoCategoryRegistry,
    MagentoConnection,
    MasterProduct,
    MasterTaxonomyChannelLink,
    MasterTaxonomyNode,
    ProductChannelTaxonomyAssignment,
    ShopifyCollectionRegistry,
    ShopifyConnection,
)
from db.product_channel_taxonomy import (
    MAGENTO_CATEGORY_KINDS,
    MAGENTO_KIND,
    SHOPIFY_COLLECTION_KIND,
    SHOPIFY_PRODUCT_CATEGORY_KIND,
)


def _seed_shopify_connection(session) -> int:
    existing = session.get(ShopifyConnection, 1)
    if existing is None:
        session.add(
            ShopifyConnection(
                id=1,
                shop_code="test",
                shop_domain="test.myshopify.com",
                status="active",
            )
        )
        session.flush()
    return 1


def test_bootstrap_builds_tree_from_products(catalog_intent_session):
    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            category_l1="Kitchen Cabinets",
            collection="Anna Snow White",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()

    result = bootstrap_master_taxonomy(session, dry_run=False)
    assert result["steps"]["backfill"]["collections_created"] >= 0
    assert result["tree"]["count"] >= 2


def test_replace_channel_taxonomy_assignment_is_connection_scoped():
    magento_one = ProductChannelTaxonomyAssignment(
        master_sku="ASG-B15",
        channel_code="magento",
        connection_id=1,
        taxonomy_kind=MAGENTO_KIND,
        remote_id="297",
    )
    magento_two = ProductChannelTaxonomyAssignment(
        master_sku="ASG-B15",
        channel_code="magento",
        connection_id=2,
        taxonomy_kind=MAGENTO_KIND,
        remote_id="397",
    )
    shopify_collection = ProductChannelTaxonomyAssignment(
        master_sku="ASG-B15",
        channel_code="shopify",
        connection_id=1000001,
        taxonomy_kind=SHOPIFY_COLLECTION_KIND,
        remote_id="gid://shopify/Collection/1",
    )
    shopify_product_category = ProductChannelTaxonomyAssignment(
        master_sku="ASG-B15",
        channel_code="shopify",
        connection_id=1000001,
        taxonomy_kind=SHOPIFY_PRODUCT_CATEGORY_KIND,
        remote_id="gid://shopify/TaxonomyCategory/1",
    )

    kwargs = {
        "magento_connection_id": 2,
        "shopify_connection_id": None,
        "magento_category_kinds": MAGENTO_CATEGORY_KINDS,
        "shopify_product_category_kind": SHOPIFY_PRODUCT_CATEGORY_KIND,
    }

    assert not _should_replace_channel_taxonomy_assignment(magento_one, **kwargs)
    assert _should_replace_channel_taxonomy_assignment(magento_two, **kwargs)
    assert not _should_replace_channel_taxonomy_assignment(shopify_collection, **kwargs)
    assert not _should_replace_channel_taxonomy_assignment(shopify_product_category, **kwargs)


def test_upsert_and_assign_skus_dry_run(catalog_intent_session):
    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            category_l1="Kitchen Cabinets",
            collection="Anna Snow White",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()
    bootstrap_master_taxonomy(session, dry_run=False)
    session.flush()

    collection = next(n for n in bootstrap_master_taxonomy(session, dry_run=True)["tree"]["collections"])
    upserted = upsert_taxonomy_node(
        session,
        {"id": collection["id"], "name": "Anna Snow White Renamed", "sync_listing_path": True},
    )
    assert "Renamed" in upserted["node"]["name"]

    assign = assign_skus_to_taxonomy_node(
        session,
        node_id=collection["id"],
        skus=["ASG-B15"],
        dry_run=True,
    )
    assert assign["sku_count"] == 1
    assert assign["assignment_count"] >= 1


def test_ensure_collection_landing_link_creates_page(catalog_intent_session):
    from sqlalchemy import select

    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            category_l1="Kitchen Cabinets",
            collection="Anna Snow White",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()
    bootstrap_master_taxonomy(session, dry_run=False)
    node = session.scalar(
        select(MasterTaxonomyNode).where(MasterTaxonomyNode.node_kind == "collection").limit(1)
    )
    assert node is not None

    result = ensure_collection_landing_link(session, node)
    session.flush()
    assert result["ensured"] is True
    landing = session.scalar(
        select(CollectionLandingPage).where(CollectionLandingPage.path_slug == node.path_slug).limit(1)
    )
    assert landing is not None
    assert node.collection_registry_id is not None
    assert landing.collection_registry_id == node.collection_registry_id


def test_upsert_path_slug_cascades_child_paths(catalog_intent_session):
    from sqlalchemy import select

    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            category_l1="Kitchen Cabinets",
            collection="Anna Snow White",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()
    bootstrap_master_taxonomy(session, dry_run=False)
    session.flush()

    hub = session.scalar(select(MasterTaxonomyNode).where(MasterTaxonomyNode.node_kind == "hub").limit(1))
    child = session.scalar(
        select(MasterTaxonomyNode)
        .where(MasterTaxonomyNode.parent_id == hub.id)
        .where(MasterTaxonomyNode.node_kind == "collection")
        .limit(1)
    )
    assert hub is not None and child is not None
    old_hub_slug = hub.path_slug
    old_child_slug = child.path_slug
    new_hub_slug = f"{old_hub_slug}-renamed"

    result = upsert_taxonomy_node(
        session,
        {"id": hub.id, "name": hub.name, "path_slug": new_hub_slug},
    )
    session.flush()

    hub_row = session.get(MasterTaxonomyNode, hub.id)
    child_row = session.get(MasterTaxonomyNode, child.id)
    assert hub_row.path_slug == new_hub_slug
    assert child_row.path_slug == new_hub_slug + old_child_slug[len(old_hub_slug) :]
    assert result["path_cascade"]["cascade_count"] >= 1


def test_reparent_cascades_child_collection_paths(catalog_intent_session):
    from sqlalchemy import select

    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            category_l1="Kitchen Cabinets",
            collection="Anna Snow White",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()
    bootstrap_master_taxonomy(session, dry_run=False)
    session.flush()

    hub = session.scalar(
        select(MasterTaxonomyNode)
        .where(MasterTaxonomyNode.node_kind == "hub")
        .limit(1)
    )
    child = session.scalar(
        select(MasterTaxonomyNode)
        .where(MasterTaxonomyNode.parent_id == hub.id)
        .where(MasterTaxonomyNode.node_kind == "collection")
        .limit(1)
    )
    assert hub is not None
    assert child is not None
    old_hub_slug = hub.path_slug
    old_child_slug = child.path_slug
    assert old_child_slug.startswith(old_hub_slug + "/")

    doors_node = upsert_taxonomy_node(
        session,
        {"name": "Doors", "node_kind": "hub", "path_slug": "doors"},
    )["node"]
    session.flush()
    hub_id = hub.id
    child_id = child.id

    result = reparent_taxonomy_node(session, hub_id, parent_id=doors_node["id"])
    session.flush()

    hub_row = session.get(MasterTaxonomyNode, hub_id)
    child_row = session.get(MasterTaxonomyNode, child_id)
    hub_leaf = old_hub_slug.split("/")[-1]
    assert hub_row.path_slug == f"doors/{hub_leaf}"
    assert child_row.path_slug == hub_row.path_slug + old_child_slug[len(old_hub_slug) :]
    assert result["path_cascade"]["cascade_count"] >= 1


def test_shopify_title_and_handle_for_deep_path(catalog_intent_session):
    from db.master_taxonomy_sync import _shopify_handle_for_node, _shopify_title_for_node

    session = catalog_intent_session
    kitchen = upsert_taxonomy_node(
        session,
        {"name": "Kitchen Cabinets", "path_slug": "kitchen-cabinets", "node_kind": "hub"},
    )["node"]
    accessories = upsert_taxonomy_node(
        session,
        {
            "name": "Accessories",
            "path_slug": "kitchen-cabinets/accessories",
            "parent_id": kitchen["id"],
            "node_kind": "category",
        },
    )["node"]
    leaf = upsert_taxonomy_node(
        session,
        {
            "name": "Panels and Fillers",
            "path_slug": "kitchen-cabinets/accessories/panels-and-fillers",
            "parent_id": accessories["id"],
            "node_kind": "category",
        },
    )["node"]
    row = session.get(MasterTaxonomyNode, leaf["id"])
    assert _shopify_handle_for_node(row) == "kitchen-cabinets-accessories-panels-and-fillers"
    assert _shopify_title_for_node(session, row) == "Kitchen Cabinets Accessories Panels and Fillers"


def test_write_listing_target_ensures_listing_path(catalog_intent_session):
    from sqlalchemy import delete, select

    from db.master_taxonomy_sync import _write_listing_target
    from db.models import ChannelListingPath

    session = catalog_intent_session
    node_dict = upsert_taxonomy_node(
        session,
        {
            "name": "Panels and Fillers",
            "path_slug": "kitchen-cabinets/accessories/panels-and-fillers",
            "node_kind": "category",
        },
    )["node"]
    row = session.get(MasterTaxonomyNode, node_dict["id"])
    session.execute(delete(ChannelListingPath).where(ChannelListingPath.path_slug == row.path_slug))
    session.flush()
    _write_listing_target(session, row, "magento", 1, "99", "category", "Panels and Fillers")
    assert session.scalar(
        select(ChannelListingPath).where(ChannelListingPath.path_slug == row.path_slug).limit(1)
    ) is not None


def test_shopify_handle_for_nested_path(catalog_intent_session):
    from db.master_taxonomy_sync import _shopify_handle_for_node, _shopify_handle_suggestions

    session = catalog_intent_session
    parent = upsert_taxonomy_node(
        session,
        {"name": "Kitchen Cabinets", "path_slug": "kitchen-cabinets", "node_kind": "hub"},
    )["node"]
    child = upsert_taxonomy_node(
        session,
        {
            "name": "Doors",
            "path_slug": "kitchen-cabinets/doors",
            "parent_id": parent["id"],
            "node_kind": "category",
        },
    )["node"]
    row = session.get(MasterTaxonomyNode, child["id"])
    assert _shopify_handle_for_node(row) == "kitchen-cabinets-doors"
    suggestions = _shopify_handle_suggestions(row)
    assert "kitchen-cabinets-doors" in suggestions
    assert "kitchen_doors" in suggestions


def test_reparent_hub_under_hub_becomes_category(catalog_intent_session):
    from sqlalchemy import select

    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            category_l1="Kitchen Cabinets",
            collection="Anna Snow White",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()
    bootstrap_master_taxonomy(session, dry_run=False)
    hub = session.scalar(
        select(MasterTaxonomyNode).where(MasterTaxonomyNode.node_kind == "hub").limit(1)
    )
    assert hub is not None
    doors = upsert_taxonomy_node(
        session,
        {"name": "Doors", "node_kind": "hub", "path_slug": "doors-reparent-kind-test"},
    )["node"]
    reparent_taxonomy_node(session, hub.id, parent_id=doors["id"])
    session.flush()
    row = session.get(MasterTaxonomyNode, hub.id)
    assert row is not None
    assert row.node_kind == "category"


def test_find_taxonomy_channel_candidates_shopify(catalog_intent_session):
    session = catalog_intent_session
    connection_id = _seed_shopify_connection(session)
    node = upsert_taxonomy_node(
        session,
        {"name": "Doors", "node_kind": "hub", "path_slug": "doors-conflict-test"},
    )["node"]
    session.add(
        ShopifyCollectionRegistry(
            shop_code="test",
            connection_id=connection_id,
            collection_id="gid://shopify/Collection/123",
            handle="doors-conflict-test",
            title="Doors",
            collection_type="manual",
            fetched_at=datetime.now(timezone.utc),
        )
    )
    session.flush()
    row = session.get(MasterTaxonomyNode, node["id"])
    result = find_taxonomy_channel_candidates(
        session,
        row,
        channel_code="shopify",
        connection_id=connection_id,
    )
    assert result["candidate_count"] >= 1


def test_shopify_handle_conflict_returns_candidates(catalog_intent_session):
    session = catalog_intent_session
    connection_id = _seed_shopify_connection(session)
    node = upsert_taxonomy_node(
        session,
        {"name": "Doors", "node_kind": "hub", "path_slug": "doors-conflict-test"},
    )["node"]
    session.add(
        ShopifyCollectionRegistry(
            shop_code="test",
            connection_id=connection_id,
            collection_id="gid://shopify/Collection/123",
            handle="doors-conflict-test",
            title="Doors",
            collection_type="manual",
            fetched_at=datetime.now(timezone.utc),
        )
    )
    session.flush()
    row = session.get(MasterTaxonomyNode, node["id"])
    preflight = find_taxonomy_channel_candidates(
        session, row, channel_code="shopify", connection_id=connection_id
    )
    assert preflight["candidate_count"] >= 1

    result = push_taxonomy_nodes_to_channels(
        session,
        node_ids=[node["id"]],
        shopify_connection_id=connection_id,
        shopify_client=object(),
        shopify_shop_code="test",
        dry_run=False,
        create_missing=True,
        include_magento=False,
    )
    assert result["error_count"] == 1
    err = result["errors"][0]
    assert err["channel"] == "shopify"
    assert "conflict" in err, err.get("error")
    assert err["conflict"]["handle"] == "doors-conflict-test"
    assert len(err["conflict"]["candidates"]) >= 1
    assert err["conflict"]["candidates"][0]["remote_id"] == "gid://shopify/Collection/123"


def test_link_shopify_collection_to_taxonomy_node(catalog_intent_session):
    from sqlalchemy import select

    session = catalog_intent_session
    connection_id = _seed_shopify_connection(session)
    node = upsert_taxonomy_node(
        session,
        {"name": "Doors", "node_kind": "hub", "path_slug": "doors-link-test"},
    )["node"]
    session.add(
        ShopifyCollectionRegistry(
            shop_code="test",
            connection_id=connection_id,
            collection_id="gid://shopify/Collection/456",
            handle="doors-link-test",
            title="Doors",
            collection_type="manual",
            fetched_at=datetime.now(timezone.utc),
        )
    )
    session.flush()

    result = link_taxonomy_node_channel(
        session,
        node["id"],
        channel_code="shopify",
        remote_id="gid://shopify/Collection/456",
        connection_id=connection_id,
        shop_code="test",
    )
    assert result["linked"] is True
    link = session.scalar(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.taxonomy_node_id == node["id"])
        .where(MasterTaxonomyChannelLink.channel_code == "shopify")
        .limit(1)
    )
    assert link is not None
    assert link.remote_id == "gid://shopify/Collection/456"


def test_coerce_taxonomy_node_ids_accepts_scalar_and_list():
    assert coerce_taxonomy_node_ids(1) == [1]
    assert coerce_taxonomy_node_ids("22") == [22]
    assert coerce_taxonomy_node_ids([1, "2", 2]) == [1, 2, 2]
    assert coerce_taxonomy_node_ids(None) is None
    assert coerce_taxonomy_node_ids([]) is None


class _MockMagentoRenameApi:
    def __init__(self) -> None:
        self.update_calls: list[tuple[int, dict]] = []
        self.categories = {
            501: {"id": 501, "name": "Bathroom Cabinetry", "parent_id": 2, "is_active": True},
        }

    def search_categories(self, *, name=None, parent_id=None, url_key=None):
        if url_key == "kitchen-cabinets":
            return 200, [{"id": 501}]
        return 200, []

    def get_category(self, category_id: int):
        category = self.categories.get(int(category_id))
        if category is None:
            return 404, None
        return 200, dict(category)

    def update_category(self, category_id: int, payload: dict):
        self.update_calls.append((category_id, payload))
        self.categories[int(category_id)] = {
            **self.categories[int(category_id)],
            **payload,
        }
        return 200, {"id": category_id}


def test_push_magento_renames_category_matched_by_url_key(catalog_intent_session):
    session = catalog_intent_session
    connection_id = 1
    if session.get(MagentoConnection, connection_id) is None:
        session.add(
            MagentoConnection(
                id=connection_id,
                store_base_url="https://magento.test",
                environment="test",
                consumer_key="ck",
                consumer_secret="cs",
                access_token="at",
                access_token_secret="ts",
                status="active",
            )
        )
        session.flush()
    node = upsert_taxonomy_node(
        session,
        {"name": "Kitchen Cabinets", "node_kind": "hub", "path_slug": "kitchen-cabinets"},
    )["node"]
    session.flush()

    api = _MockMagentoRenameApi()
    result = push_taxonomy_nodes_to_channels(
        session,
        node_ids=[node["id"]],
        magento_connection_id=connection_id,
        magento_api=api,
        dry_run=False,
        create_missing=False,
        include_shopify=False,
    )

    assert result["magento_updated"] == 1
    assert api.update_calls
    assert api.update_calls[0][1]["name"] == "Kitchen Cabinets"
    link = session.scalar(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.taxonomy_node_id == node["id"])
        .where(MasterTaxonomyChannelLink.channel_code == "magento")
        .where(MasterTaxonomyChannelLink.connection_id == connection_id)
        .limit(1)
    )
    assert link is not None
    assert link.remote_id == "501"


class _MockMagentoStaleCategoryApi:
    def __init__(self) -> None:
        self.update_calls: list[tuple[int, dict]] = []
        self.create_calls: list[dict] = []
        self.next_id = 999

    def get_category(self, category_id: int):
        return 404, None

    def search_categories(self, *, name=None, parent_id=None, url_key=None):
        return 200, []

    def update_category(self, category_id: int, payload: dict):
        self.update_calls.append((category_id, payload))
        return 404, {"error": "No such entity", "body": {"message": "No such entity"}}

    def create_category(self, payload: dict):
        self.create_calls.append(payload)
        category_id = self.next_id
        self.next_id += 1
        return 200, {"id": category_id}


def test_push_magento_recreates_when_linked_category_missing(catalog_intent_session):
    session = catalog_intent_session
    connection_id = 4
    if session.get(MagentoConnection, connection_id) is None:
        session.add(
            MagentoConnection(
                id=connection_id,
                store_base_url="https://magento.test",
                environment="test",
                consumer_key="ck",
                consumer_secret="cs",
                access_token="at",
                access_token_secret="ts",
                status="active",
            )
        )
        session.flush()

    node = upsert_taxonomy_node(
        session,
        {"name": "Start Shopping", "node_kind": "category", "path_slug": "start-shopping"},
    )["node"]
    stale = MasterTaxonomyChannelLink(
        taxonomy_node_id=node["id"],
        channel_code="magento",
        connection_id=connection_id,
        taxonomy_kind=MAGENTO_KIND,
        remote_id="300",
        remote_parent_id="2",
        remote_path="Start Shopping",
        is_active=True,
    )
    session.add(stale)
    session.flush()

    api = _MockMagentoStaleCategoryApi()
    result = push_taxonomy_nodes_to_channels(
        session,
        node_ids=[node["id"]],
        magento_connection_id=connection_id,
        magento_api=api,
        dry_run=False,
        create_missing=True,
        include_shopify=False,
    )

    assert result["error_count"] == 0, result.get("errors")
    assert result["magento_created"] == 1
    assert result["magento_updated"] == 0
    assert api.update_calls == []
    assert len(api.create_calls) == 1

    link = session.scalar(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.taxonomy_node_id == node["id"])
        .where(MasterTaxonomyChannelLink.channel_code == "magento")
        .where(MasterTaxonomyChannelLink.connection_id == connection_id)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
        .limit(1)
    )
    assert link is not None
    # Unique (node, channel, connection, kind) row is reused after stale remote clear.
    assert link.remote_id == "999"
    assert link.id == stale.id


class _MockMagentoCreateTreeApi:
    def __init__(self, *, existing_ids: set[int] | None = None) -> None:
        self.existing_ids = set(existing_ids or [])
        self.create_calls: list[dict] = []
        self.update_calls: list[tuple[int, dict]] = []
        self.next_id = 500

    def get_category(self, category_id: int):
        category_id = int(category_id)
        if category_id in self.existing_ids:
            return 200, {"id": category_id, "name": "Base Cabinets", "parent_id": 5}
        return 404, None

    def search_categories(self, *, name=None, parent_id=None, url_key=None):
        return 200, []

    def update_category(self, category_id: int, payload: dict):
        self.update_calls.append((int(category_id), payload))
        if int(category_id) not in self.existing_ids:
            return 404, {"error": "missing"}
        return 200, {"id": int(category_id)}

    def create_category(self, payload: dict):
        self.create_calls.append(payload)
        category_id = self.next_id
        self.next_id += 1
        self.existing_ids.add(category_id)
        return 200, {"id": category_id}


def test_push_magento_creates_descendants_under_parent(catalog_intent_session):
    session = catalog_intent_session
    connection_id = 4
    if session.get(MagentoConnection, connection_id) is None:
        session.add(
            MagentoConnection(
                id=connection_id,
                store_base_url="https://magento.test",
                environment="test",
                consumer_key="ck",
                consumer_secret="cs",
                access_token="at",
                access_token_secret="ts",
                status="active",
            )
        )
        session.flush()

    parent = upsert_taxonomy_node(
        session,
        {"name": "Base Cabinets", "node_kind": "category", "path_slug": "kitchen-cabinets/base-cabinets"},
    )["node"]
    child = upsert_taxonomy_node(
        session,
        {
            "name": "Drawer Base",
            "node_kind": "category",
            "parent_id": parent["id"],
            "path_slug": "kitchen-cabinets/base-cabinets/drawer-base",
        },
    )["node"]
    session.add(
        MasterTaxonomyChannelLink(
            taxonomy_node_id=parent["id"],
            channel_code="magento",
            connection_id=connection_id,
            taxonomy_kind=MAGENTO_KIND,
            remote_id="36",
            remote_parent_id="5",
            remote_path="Base Cabinets",
            is_active=True,
        )
    )
    session.flush()

    api = _MockMagentoCreateTreeApi(existing_ids={36})
    result = push_taxonomy_nodes_to_channels(
        session,
        node_ids=[parent["id"]],
        magento_connection_id=connection_id,
        magento_api=api,
        dry_run=False,
        create_missing=True,
        include_shopify=False,
        include_descendants=True,
    )

    assert result["error_count"] == 0, result.get("errors")
    assert result["magento_updated"] == 1
    assert result["magento_created"] == 1
    assert api.update_calls and api.update_calls[0][0] == 36
    assert len(api.create_calls) == 1
    assert api.create_calls[0]["name"] == "Drawer Base"
    assert api.create_calls[0]["parent_id"] == 36

    child_link = session.scalar(
        select(MasterTaxonomyChannelLink)
        .where(MasterTaxonomyChannelLink.taxonomy_node_id == child["id"])
        .where(MasterTaxonomyChannelLink.channel_code == "magento")
        .where(MasterTaxonomyChannelLink.connection_id == connection_id)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
        .limit(1)
    )
    assert child_link is not None
    assert child_link.remote_id == "500"


def test_assign_global_root_channel_memberships(catalog_intent_session):
    session = catalog_intent_session
    now = datetime.now(timezone.utc)
    session.add(
        MagentoConnection(
            id=1,
            store_base_url="https://magento.test",
            environment="test",
            consumer_key="ck",
            consumer_secret="cs",
            access_token="at",
            access_token_secret="ts",
            status="active",
        )
    )
    session.add(
        ShopifyConnection(
            id=1,
            shop_code="test",
            shop_domain="test.myshopify.com",
            status="active",
        )
    )
    session.flush()
    session.add(
        MagentoCategoryRegistry(
            connection_id=1,
            category_id=2,
            parent_id=1,
            name="Default Category",
            path="/1/2",
            path_names="Default Category",
            level=1,
            is_active=True,
            fetched_at=now,
        )
    )
    session.add(
        ShopifyCollectionRegistry(
            connection_id=1,
            shop_code="test",
            collection_id="gid://shopify/Collection/all",
            handle="all",
            title="All",
            collection_type="smart",
            fetched_at=now,
        )
    )
    session.add(
        MasterProduct(
            sku="ACH-B12",
            name="B12",
            category_l1="Kitchen Cabinets",
            collection="Anna Caramel Harvest",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()

    stats = _assign_global_root_channel_memberships(
        session,
        ["ACH-B12"],
        magento_connection_id=1,
        shopify_connection_id=1,
    )
    assert stats["magento_remote_id"] == "2"
    assert stats["magento_remote_path"] == "Default Category"
    assert stats["magento_assigned"] == 1
    assert stats["shopify_remote_id"] == "gid://shopify/Collection/all"
    assert stats["shopify_assigned"] == 1
    assert stats["magento_missing"] is False
    assert stats["shopify_missing"] is False

    magento_row = session.scalar(
        select(ProductChannelTaxonomyAssignment).where(
            ProductChannelTaxonomyAssignment.master_sku == "ACH-B12",
            ProductChannelTaxonomyAssignment.channel_code == "magento",
            ProductChannelTaxonomyAssignment.remote_id == "2",
            ProductChannelTaxonomyAssignment.assignment_status == "active",
        )
    )
    shopify_row = session.scalar(
        select(ProductChannelTaxonomyAssignment).where(
            ProductChannelTaxonomyAssignment.master_sku == "ACH-B12",
            ProductChannelTaxonomyAssignment.channel_code == "shopify",
            ProductChannelTaxonomyAssignment.remote_id == "gid://shopify/Collection/all",
            ProductChannelTaxonomyAssignment.assignment_status == "active",
        )
    )
    assert magento_row is not None
    assert shopify_row is not None


def test_link_products_keeps_all_products_listing_path(catalog_intent_session):
    from db.channel_listing_path import resolve_listing_paths_for_sku

    session = catalog_intent_session
    hub = MasterTaxonomyNode(
        parent_id=None,
        node_kind="hub",
        code="kitchen-cabinets",
        name="Kitchen Cabinets",
        path_slug="kitchen-cabinets",
        is_active=True,
    )
    session.add(hub)
    session.flush()
    session.add(
        MasterTaxonomyNode(
            parent_id=hub.id,
            node_kind="collection",
            code="anna-caramel-harvest",
            name="Kitchen Cabinets / Anna Caramel Harvest",
            path_slug="kitchen-cabinets/anna-caramel-harvest",
            is_active=True,
        )
    )
    session.add(
        MasterProduct(
            sku="ACH-B12",
            name="B12",
            category_l1="Kitchen Cabinets",
            collection="Anna Caramel Harvest",
            row_hash="h1",
            is_active=True,
        )
    )
    session.flush()

    result = link_products_to_taxonomy_from_master(
        session,
        skus=["ACH-B12"],
        replace_existing=True,
        sync_channels=False,
        dry_run=False,
    )
    assert result["upserted"] >= 1
    assigned = {row["path_slug"] for row in resolve_listing_paths_for_sku(session, "ACH-B12")}
    assert "all-products" in assigned
    assert "kitchen-cabinets/anna-caramel-harvest" in assigned