from unittest.mock import MagicMock, patch from magento.collection_landing_push import sync_magento_collection_products def _landing(path_slug: str, start_shopping_config=None): landing = MagicMock() landing.path_slug = path_slug landing.is_active = True landing.start_shopping_config = start_shopping_config or {} return landing def _mock_shopping_attributes_ready(api: MagicMock) -> None: api.get_attribute.return_value = ( 200, { "attribute_code": "shopping_collection", "is_searchable": True, "is_filterable": 1, "is_filterable_in_search": True, }, ) def test_sync_magento_collection_products_assigns_only_missing_channel_skus(): session = MagicMock() session.scalars.return_value.all.return_value = [] api = MagicMock() _mock_shopping_attributes_ready(api) api.get_category_products.return_value = (200, [{"sku": "ACH-B12"}], None) api.assign_product_to_category.return_value = (200, {"success": True}, None) api.put_product.return_value = (200, {"sku": "ACH-W15"}) resolved = { "skus": [ {"master_sku": "ACH-B12", "match_source": "master_collection"}, {"master_sku": "ACH-W15", "match_source": "master_collection"}, ] } with patch( "db.collection_sku_mapping.resolve_collection_skus", return_value=resolved ), patch( "db.channel_sku_mapping.mapping_by_master", return_value={"ACH-B12": "ACH-B12", "ACH-W15": "ACH-W15"}, ), patch("db.product_channel_taxonomy.upsert_sku_taxonomy_assignment") as upsert: result = sync_magento_collection_products( session, api, category_id=296, collection_row={ "path_slug": "kitchen-cabinets/anna-caramel-harvest", "category_l1": "Kitchen Cabinets", "collection": "Anna Caramel Harvest", }, connection_id=1, ) assert result == { "dry_run": False, "resolved": 2, "processed": 2, "assigned": 1, "skipped_existing": 1, "failed": 0, "errors": [], } api.assign_product_to_category.assert_called_once_with(296, "ACH-W15", position=1) assert upsert.call_count == 2 def test_sync_magento_collection_products_dry_run_does_not_call_magento(): session = MagicMock() api = MagicMock() _mock_shopping_attributes_ready(api) with patch( "db.collection_sku_mapping.resolve_collection_skus", return_value={"skus": [{"master_sku": "ACH-B12"}]}, ), patch( "db.channel_sku_mapping.mapping_by_master", return_value={"ACH-B12": "ACH-B12"} ): result = sync_magento_collection_products( session, api, category_id=296, collection_row={"collection": "Anna Caramel Harvest"}, connection_id=1, dry_run=True, ) assert result["would_assign"] == 1 api.get_category_products.assert_not_called() def test_sync_magento_collection_products_continues_after_per_sku_failure(): session = MagicMock() session.scalars.return_value.all.return_value = [] api = MagicMock() _mock_shopping_attributes_ready(api) api.get_category_products.return_value = (200, [], None) api.assign_product_to_category.return_value = (200, {"success": True}, None) api.put_product.return_value = (200, {"sku": "ACH-B12"}) def upsert_side_effect(*_args, **kwargs): if kwargs.get("master_sku") == "BAD-SKU": raise ValueError("Unknown master SKU: BAD-SKU") return {} with patch( "db.channel_sku_mapping.mapping_by_master", return_value={"ACH-B12": "ACH-B12", "BAD-SKU": "BAD-SKU", "ACH-W15": "ACH-W15"}, ), patch( "db.product_channel_taxonomy.upsert_sku_taxonomy_assignment", side_effect=upsert_side_effect, ): result = sync_magento_collection_products( session, api, category_id=296, collection_row={"path_slug": "kitchen-cabinets/anna-caramel-harvest"}, connection_id=1, master_skus=["ACH-B12", "BAD-SKU", "ACH-W15"], ) assert result["processed"] == 3 assert result["assigned"] == 2 assert result["failed"] == 1 assert result["errors"][0]["master_sku"] == "BAD-SKU" def test_sync_magento_collection_products_retries_shopping_filter_write_via_post(): session = MagicMock() api = MagicMock() _mock_shopping_attributes_ready(api) api.get_category_products.return_value = (200, [], None) api.assign_product_to_category.return_value = (200, {"success": True}, None) api.put_product.return_value = ( 400, {"message": "Invalid product data: Invalid attribute set entity type"}, ) api.get_product.return_value = { "sku": "ACH-B12", "type_id": "simple", "attribute_set_id": 4, "custom_attributes": [{"attribute_code": "url_key", "value": "ach-b12"}], } api.post_product.return_value = (200, {"sku": "ACH-B12"}) with patch( "db.channel_sku_mapping.mapping_by_master", return_value={"ACH-B12": "ACH-B12"}, ), patch( "db.product_channel_taxonomy.upsert_sku_taxonomy_assignment", return_value={}, ), patch( "db.channel_listing_path.resolve_listing_paths_for_sku", return_value=[ {"path_slug": "kitchen-cabinets/base-cabinets/single-door"}, ], ): session.scalars.return_value.all.return_value = [ _landing( "kitchen-cabinets/anna-caramel-harvest", { "items": [ { "l1_path_slug": "kitchen-cabinets/base-cabinets", "l2_path_slugs": ["kitchen-cabinets/base-cabinets/single-door"], } ] }, ) ] result = sync_magento_collection_products( session, api, category_id=296, collection_row={"path_slug": "kitchen-cabinets/anna-caramel-harvest"}, connection_id=1, master_skus=["ACH-B12"], ) assert result["failed"] == 0 api.post_product.assert_called_once() retry_payload = api.post_product.call_args.args[0] assert retry_payload["attribute_set_id"] == 4 assert retry_payload["type_id"] == "simple" attrs = {item["attribute_code"]: item["value"] for item in retry_payload["custom_attributes"]} assert attrs["url_key"] == "ach-b12" assert attrs["shopping_collection"] == "anna-caramel-harvest" assert attrs["shopping_l1"] == "base-cabinets" assert attrs["shopping_l2"] == "base-cabinets/single-door" def test_sync_magento_collection_products_skips_rta_skus(): session = MagicMock() session.scalars.return_value.all.return_value = [] api = MagicMock() _mock_shopping_attributes_ready(api) api.get_category_products.return_value = (200, [], None) api.assign_product_to_category.return_value = (200, {"success": True}, None) api.put_product.return_value = (200, {"sku": "ACH-B12"}) with patch( "db.channel_sku_mapping.mapping_by_master", return_value={"RTA-ACH-B12": "RTA-ACH-B12", "ACH-B12": "ACH-B12"}, ), patch( "db.product_channel_taxonomy.upsert_sku_taxonomy_assignment", return_value={}, ), patch("db.channel_listing_path.resolve_listing_paths_for_sku", return_value=[]): result = sync_magento_collection_products( session, api, category_id=296, collection_row={"path_slug": "kitchen-cabinets/anna-caramel-harvest"}, connection_id=1, master_skus=["RTA-ACH-B12", "ACH-B12"], ) assert result["processed"] == 1 api.assign_product_to_category.assert_called_once_with(296, "ACH-B12", position=0) def test_sync_magento_collection_products_derives_l1_from_leaf_assignment_paths(): session = MagicMock() api = MagicMock() _mock_shopping_attributes_ready(api) api.get_category_products.return_value = (200, [], None) api.assign_product_to_category.return_value = (200, {"success": True}, None) api.put_product.return_value = (200, {"sku": "CSG-B12"}) with patch( "db.channel_sku_mapping.mapping_by_master", return_value={"CSG-B12": "CSG-B12"}, ), patch( "db.product_channel_taxonomy.upsert_sku_taxonomy_assignment", return_value={}, ), patch( "db.channel_listing_path.resolve_listing_paths_for_sku", return_value=[ {"path_slug": "kitchen-cabinets/base-cabinets/single-door"}, ], ): session.scalars.return_value.all.return_value = [ _landing( "kitchen-cabinets/cooper-slate-gray", { "items": [ { "l1_path_slug": "kitchen-cabinets/base-cabinets", "l2_path_slugs": ["kitchen-cabinets/base-cabinets/single-door"], } ] }, ) ] result = sync_magento_collection_products( session, api, category_id=393, collection_row={"path_slug": "kitchen-cabinets/cooper-slate-gray"}, connection_id=1, master_skus=["CSG-B12"], ) assert result["failed"] == 0 put_payload = api.put_product.call_args.args[1] attrs = {item["attribute_code"]: item["value"] for item in put_payload["custom_attributes"]} assert attrs["shopping_collection"] == "cooper-slate-gray" assert attrs["shopping_l1"] == "base-cabinets" assert attrs["shopping_l2"] == "base-cabinets/single-door" def test_sync_magento_collection_products_ignores_stale_magento_assignment_paths_for_l2(): session = MagicMock() api = MagicMock() _mock_shopping_attributes_ready(api) api.get_category_products.return_value = (200, [], None) api.assign_product_to_category.return_value = (200, {"success": True}, None) api.put_product.return_value = (200, {"sku": "ASG-W1230"}) with patch( "db.channel_sku_mapping.mapping_by_master", return_value={"ASG-W1230": "ASG-W1230"}, ), patch( "db.product_channel_taxonomy.upsert_sku_taxonomy_assignment", return_value={}, ), patch( "db.channel_listing_path.resolve_listing_paths_for_sku", return_value=[ {"path_slug": "kitchen-cabinets/wall-cabinets/basic-wall"}, ], ): session.scalars.return_value.all.return_value = [ _landing( "kitchen-cabinets/anna-stone-gray", { "items": [ { "l1_path_slug": "kitchen-cabinets/wall-cabinets", "l2_path_slugs": [ "kitchen-cabinets/wall-cabinets/basic-wall", "kitchen-cabinets/wall-cabinets/single-door-glass-wall-cabinet", ], } ] }, ) ] result = sync_magento_collection_products( session, api, category_id=299, collection_row={"path_slug": "kitchen-cabinets/anna-stone-gray"}, connection_id=4, master_skus=["ASG-W1230"], ) assert result["failed"] == 0 put_payload = api.put_product.call_args.args[1] attrs = {item["attribute_code"]: item["value"] for item in put_payload["custom_attributes"]} assert attrs["shopping_collection"] == "anna-stone-gray" assert attrs["shopping_l1"] == "wall-cabinets" assert attrs["shopping_l2"] == "wall-cabinets/basic-wall"