"""Tests for kitchen package resolution and channel bundle payloads.""" from __future__ import annotations from dataclasses import replace from sqlalchemy import select from db.kitchen_packages import ( KitchenPackageComponent, package_master_sku, resolve_kitchen_packages_from_payload, upsert_package_master_products, ) from db.models import MasterProduct, MasterProductPrice from magento.bundle_product import build_magento_fixed_bundle_payload from magento.bundle_product import upsert_magento_kitchen_package from shopify.bundle_product import ( build_shopify_package_product_set_input, build_shopify_variant_bundle_input, ) def _grace_payload(): return { "source_document": {"collection_code": "GSW"}, "collection": { "display_name": "Grace Snow White", "category_l1": "Kitchen Cabinets", "brand_label": "Home Surplus", "canonical_code": "grace-snow-white", }, "kitchen_packages": [ { "package_code": "l_shape_8x10", "label": "8 x 10 L-Shape Kitchen", "price_usd": 1999, "includes": [ { "qty": 2, "description": "Wall 12 x 30", "mapped_brochure_sku": "W1230", "mapping_confidence": "high", }, { "qty": 1, "description": "Base Sink 36", "mapped_brochure_sku": "SB36", "mapping_confidence": "high", }, ], }, { "package_code": "u_shape_10x10x10", "label": "10 x 10 x 10 U-Shape Kitchen", "price_usd": 4685, "includes": [ { "qty": 1, "description": "Base 18", "mapped_brochure_sku": "B18", "mapping_confidence": "high", } ], }, ], } def _anna_payload(): return { "source_document": {"collection_code": "ACH"}, "collection": { "display_name": "Anna Caramel Harvest", "category_l1": "Kitchen Cabinets", "brand_label": "Home Surplus", "canonical_code": "anna-caramel-harvest", }, "kitchen_packages": [ { "package_code": "l_shape_8x10", "label": "8 x 10 L-Shape Kitchen", "price_usd": 3116, "includes": [ {"qty": 1, "description": 'Wall Corner 24" x 30"', "mapped_brochure_sku": "LS36"}, {"qty": 2, "description": 'Wall 12" x 30"', "mapped_brochure_sku": "W1230"}, {"qty": 1, "description": 'Wall 36" x 15"', "mapped_brochure_sku": "W3615"}, {"qty": 1, "description": 'Wall 18" x 30"', "mapped_brochure_sku": "W1830"}, {"qty": 1, "description": 'Wall 30" x 15"', "mapped_brochure_sku": "W3015"}, {"qty": 1, "description": 'Wall 24" x 30"', "mapped_brochure_sku": "W2430"}, {"qty": 1, "description": "Base Sink 36", "mapped_brochure_sku": "SB36"}, {"qty": 1, "description": "Lazy Susan 36", "mapped_brochure_sku": "LS36"}, {"qty": 1, "description": "Base 18", "mapped_brochure_sku": "B18"}, ], } ], } def _with_existing_components(pkg): return replace( pkg, components=[ replace(c, master_exists=True) if isinstance(c, KitchenPackageComponent) else c for c in pkg.components ], ) def test_package_master_sku_suffixes(): assert package_master_sku("GSW", "l_shape_8x10") == "GSW-PKG-L810" assert package_master_sku("ACH", "u_shape_10x10x10") == "ACH-PKG-U101010" def test_resolve_kitchen_packages_from_payload_builds_two_packages(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) assert len(packages) == 2 by_code = {p.package_code: p for p in packages} assert by_code["l_shape_8x10"].package_master_sku == "GSW-PKG-L810" assert by_code["l_shape_8x10"].price_usd == 1999 assert [c.master_sku for c in by_code["l_shape_8x10"].components] == ["GSW-W1230", "GSW-SB36"] assert by_code["l_shape_8x10"].components[0].qty == 2 assert by_code["u_shape_10x10x10"].package_master_sku == "GSW-PKG-U101010" assert by_code["u_shape_10x10x10"].price_usd == 4685 def test_resolve_kitchen_packages_prefers_brochure_description_skus_for_packages(): packages = resolve_kitchen_packages_from_payload(_anna_payload()) assert len(packages) == 1 pkg = packages[0] assert [component.master_sku for component in pkg.components] == [ "ACH-WCOR2430", "ACH-W1230", "ACH-W3615", "ACH-W1830", "ACH-W3015", "ACH-W2430", "ACH-SB36", "ACH-BCOR36L/R", "ACH-B18", ] def test_magento_fixed_bundle_payload_sets_price_type_fixed(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) payload = build_magento_fixed_bundle_payload(pkg, attribute_set_id=4, website_ids=[1]) assert payload["type_id"] == "bundle" assert payload["price"] == 1999 assert payload["sku"] == "GSW-PKG-L810" attrs = {a["attribute_code"]: a["value"] for a in payload["custom_attributes"]} assert attrs["price_type"] == "1" assert attrs["sku_type"] == "1" assert attrs["weight_type"] == "1" assert attrs["shipment_type"] == "0" assert attrs["options_container"] == "container2" assert attrs["tax_class_id"] == "2" options = payload["extension_attributes"]["bundle_product_options"] assert len(options) == 2 assert options[0]["required"] is True assert options[0]["option_id"] == 0 assert options[0]["product_links"][0]["sku"] == "GSW-W1230" assert options[0]["product_links"][0]["option_id"] == 1 assert options[0]["sku"].startswith("GSW-PKG-L810-") assert options[0]["product_links"][0]["qty"] == 2.0 assert options[0]["product_links"][0]["can_change_quantity"] == 0 assert options[0]["product_links"][0]["is_default"] is True assert payload["extension_attributes"]["stock_item"]["is_in_stock"] is True def test_shopify_package_input_has_fixed_price_and_bom_metafield(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) product_input = build_shopify_package_product_set_input(pkg) assert product_input["variants"][0]["sku"] == "GSW-PKG-L810" assert product_input["variants"][0]["price"] == "1999.00" assert product_input["productType"] == "Kitchen Package" assert product_input["productOptions"][0]["name"] == "Title" keys = {m["key"] for m in product_input["metafields"]} assert "kitchen_package_bom" in keys assert "kitchen_package_code" in keys def test_shopify_bundle_input_uses_component_channel_skus_and_quantities(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) bundle_input = build_shopify_variant_bundle_input( pkg, parent_variant_id="gid://shopify/ProductVariant/PARENT", component_channel_skus={ "GSW-W1230": "SHOP-W1230", "GSW-SB36": "SHOP-SB36", }, ) assert bundle_input["parentProductVariantId"] == "gid://shopify/ProductVariant/PARENT" assert bundle_input["productVariantRelationshipsToCreate"] == [ {"sku": "SHOP-W1230", "quantity": 2}, {"sku": "SHOP-SB36", "quantity": 1}, ] def test_upsert_package_master_products(catalog_intent_session): session = catalog_intent_session packages = resolve_kitchen_packages_from_payload(_grace_payload(), session=session) result = upsert_package_master_products(session, packages) session.flush() assert result["created"] == 2 row = session.scalar(select(MasterProduct).where(MasterProduct.sku == "GSW-PKG-L810")) assert row is not None assert row.assembly_type == "bundle" price = session.scalar( select(MasterProductPrice).where( MasterProductPrice.sku == "GSW-PKG-L810", MasterProductPrice.price_type == "base", ) ) assert price is not None assert float(price.amount) == 1999.0 class _FakeMagentoApi: def __init__( self, *, existing_component_skus=None, component_types=None, configurable_children=None, inline_options_on_save=True, ): self.post_calls = [] self.put_calls = [] self.add_option_calls = [] self.add_link_calls = [] self.update_option_calls = [] self.delete_option_calls = [] self.delete_product_calls = [] self.existing_component_skus = { str(s).strip().upper() for s in (existing_component_skus or []) } self.component_types = { str(k).strip().upper(): str(v).strip().lower() for k, v in (component_types or {}).items() } self.configurable_children = { str(k).strip().upper(): list(v) for k, v in (configurable_children or {}).items() } self.inline_options_on_save = inline_options_on_save self._options_by_sku = {} self._products = {} def get_product(self, sku): key = str(sku or "").strip().upper() if key in self._products: return self._products[key] if key in self.existing_component_skus: return { "sku": sku, "type_id": self.component_types.get(key, "simple"), "id": 1, } return None def get_configurable_children(self, sku): key = str(sku or "").strip().upper() return 200, list(self.configurable_children.get(key, [])), None def post_product(self, payload, *, save_options=False): self.post_calls.append({"payload": payload, "save_options": save_options}) body = {"id": 101, "sku": payload["sku"], "extension_attributes": {}} key = str(payload["sku"]).upper() if save_options and self.inline_options_on_save: options = (payload.get("extension_attributes") or {}).get("bundle_product_options") or [] saved = [{**opt, "option_id": idx + 1} for idx, opt in enumerate(options)] body["extension_attributes"]["bundle_product_options"] = saved self._options_by_sku[key] = saved self._products[key] = body return 200, body def put_product(self, sku, payload, *, save_options=False): self.put_calls.append({"sku": sku, "payload": payload, "save_options": save_options}) body = {"id": 101, "sku": sku, "extension_attributes": {}} key = str(sku).upper() if save_options and self.inline_options_on_save: options = (payload.get("extension_attributes") or {}).get("bundle_product_options") or [] saved = [{**opt, "option_id": idx + 1} for idx, opt in enumerate(options)] body["extension_attributes"]["bundle_product_options"] = saved self._options_by_sku[key] = saved self._products[key] = body return 200, body def delete_product(self, sku): self.delete_product_calls.append(sku) key = str(sku).upper() self._products.pop(key, None) self._options_by_sku.pop(key, None) return 200, None def get_bundle_options(self, sku): return 200, list(self._options_by_sku.get(str(sku).upper(), [])), None def add_bundle_option(self, option): self.add_option_calls.append(option) option_id = len(self.add_option_calls) sku = str(option.get("sku") or "").upper() stored = dict(option) stored["option_id"] = option_id self._options_by_sku.setdefault(sku, []).append(stored) return 200, option_id, None def add_bundle_product_link(self, sku, option_id, linked_product): self.add_link_calls.append( {"sku": sku, "option_id": option_id, "linked_product": linked_product} ) key = str(sku).upper() for opt in self._options_by_sku.get(key, []): if int(opt.get("option_id") or 0) == int(option_id): links = list(opt.get("product_links") or []) links.append(dict(linked_product)) opt["product_links"] = links break return 200, True, None def update_bundle_option(self, option_id, option): self.update_option_calls.append({"option_id": option_id, "option": option}) return 200, option_id, None def delete_bundle_option(self, sku, option_id): self.delete_option_calls.append({"sku": sku, "option_id": option_id}) return 204, None def test_magento_upsert_updates_existing_in_place(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) api = _FakeMagentoApi( existing_component_skus=["GSW-W1230", "GSW-SB36"], inline_options_on_save=True, ) api._products["GSW-PKG-L810"] = {"sku": "GSW-PKG-L810", "type_id": "bundle", "id": 55} result = upsert_magento_kitchen_package( api, pkg, attribute_set_id=4, website_ids=[1], dry_run=False, ) assert result["status"] == "ok" assert result["action"] == "update" assert api.delete_product_calls == [] assert api.put_calls[0]["save_options"] is True assert api.post_calls == [] assert result["bundle_option_results"][0]["step"] == "verify_options" def test_magento_upsert_force_recreate_deletes_then_creates(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) api = _FakeMagentoApi( existing_component_skus=["GSW-W1230", "GSW-SB36"], inline_options_on_save=True, ) api._products["GSW-PKG-L810"] = {"sku": "GSW-PKG-L810", "type_id": "bundle", "id": 55} result = upsert_magento_kitchen_package( api, pkg, attribute_set_id=4, website_ids=[1], dry_run=False, force_recreate=True, ) assert result["status"] == "ok" assert result["action"] == "recreate" assert result["force_recreate"] is True assert "entity_id" in result["recreate_warning"] assert result["previous_magento_product_id"] == 55 assert api.delete_product_calls == ["GSW-PKG-L810"] assert api.post_calls[0]["save_options"] is True def test_magento_upsert_rejects_configurable_components_without_remap(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) api = _FakeMagentoApi( existing_component_skus=["GSW-W1230", "GSW-W1230-WH", "GSW-SB36"], component_types={"GSW-W1230": "configurable", "GSW-W1230-WH": "simple", "GSW-SB36": "simple"}, configurable_children={"GSW-W1230": ["GSW-W1230-WH"]}, inline_options_on_save=True, ) result = upsert_magento_kitchen_package( api, pkg, attribute_set_id=4, website_ids=[1], dry_run=False, ) assert result["status"] == "failed" assert result["action"] == "preflight" reasons = {i["reason"] for i in result["error"]["issues"]} assert "configurable_parent_not_allowed" in reasons assert api.post_calls == [] assert api.put_calls == [] assert api.delete_product_calls == [] def test_magento_upsert_preflight_fails_for_missing_or_composite_components(): packages = resolve_kitchen_packages_from_payload(_grace_payload()) pkg = _with_existing_components(packages[0]) api = _FakeMagentoApi( existing_component_skus=["GSW-W1230"], component_types={"GSW-W1230": "configurable"}, configurable_children={"GSW-W1230": []}, ) result = upsert_magento_kitchen_package( api, pkg, attribute_set_id=4, website_ids=[1], dry_run=False, ) assert result["status"] == "failed" assert result["action"] == "preflight" reasons = {i["reason"] for i in result["error"]["issues"]} assert "configurable_parent_not_allowed" in reasons assert "missing_in_magento" in reasons assert api.post_calls == [] assert "remapped" not in result.get("link_resolution", {})