""" Sanity test: given one parent row with configurable_variations + configurable_attributes, verify cache refresh logic and that we can build a POST payload with attribute_id + value_index. """ from datetime import datetime, timedelta, timezone from typing import Any, Dict, List from unittest.mock import MagicMock import pytest from magento.sync_service import ( _extract_needed_attribute_codes_from_rows, _is_configurable_parent_static, _norm_option_label, MagentoSyncService, ) def test_extract_needed_attribute_codes(): """Extract attr codes from configurable_attributes (code=label) and configurable_variations.""" rows = [ { "sku": "PARENT", "product_type": "configurable", "configurable_attributes": "width=Width (inches),color=Color", "configurable_variations": "sku=CHILD1,width=24|sku=CHILD2,width=18", }, ] codes, code_to_label = _extract_needed_attribute_codes_from_rows(rows) assert "width" in codes assert "color" in codes assert "sku" not in codes assert code_to_label.get("width") == "Width (inches)" assert code_to_label.get("color") == "Color" def test_is_configurable_parent_static(): """Configurable parent has variant_list or sku= in configurable_variations.""" assert _is_configurable_parent_static({ "product_type": "configurable", "variant_list": "C1,C2", }) assert _is_configurable_parent_static({ "product_type": "configurable", "configurable_variations": "sku=C1,width=24", }) assert not _is_configurable_parent_static({ "product_type": "simple", }) assert not _is_configurable_parent_static({ "product_type": "configurable", "configurable_variations": "", }) def test_norm_option_label(): """Normalize for cache lookup.""" assert _norm_option_label("24") == "24" assert _norm_option_label(" 24 ") == "24" assert _norm_option_label("Width (inches)") == "width (inches)" assert _norm_option_label("") == "" def test_build_configurable_option_payload_with_cache(): """Given attribute cache has attribute_id and options, build payload with value_index.""" from magento.sync_hashes import _parse_children_from_row row = { "sku": "PARENT", "product_type": "configurable", "configurable_attributes": "width=Width (inches)", "configurable_variations": "sku=CHILD1,width=24|sku=CHILD2,width=18", } children, mappings, _ = _parse_children_from_row(row) assert children == ["CHILD1", "CHILD2"] assert len(mappings) == 2 assert mappings[0].get("width") == "24" assert mappings[1].get("width") == "18" # Simulate cache: attribute_id=92, options 24->583, 18->589 class FakeAttrCacheRepo: def get_attributes(self, conn_id: int, codes: List[str], ttl) -> Dict[str, dict]: if "width" in [c.lower() for c in codes]: return {"width": {"attribute_id": 92, "frontend_label": "Width (inches)"}} return {} def get_options(self, conn_id: int, code: str, ttl) -> Dict[str, int]: if code.lower() == "width": return {"24": 583, "18": 589} return {} mock_api = MagicMock() svc = MagicMock() svc._api = mock_api svc._attr_cache_repo = FakeAttrCacheRepo() svc._conn_id = 1 # Use real _build_configurable_option_payload logic from magento.sync_service import MagentoSyncService real_svc = MagentoSyncService(mock_api, {"id": 1}, attr_cache_repo=FakeAttrCacheRepo()) payload, err = real_svc._build_configurable_option_payload( "PARENT", "width", mappings, 0, ) assert err is None assert payload is not None assert payload["attribute_id"] == "92" assert payload["values"] == [{"value_index": 583}, {"value_index": 589}] mock_api.get_attribute_detail.assert_not_called() mock_api.get_attribute_options.assert_not_called() def test_transform_custom_attributes_for_magento_coerces_boolean_values(): class FakeAttrCacheRepo: def get_attributes(self, conn_id: int, codes: List[str], ttl) -> Dict[str, dict]: return { "soft_close": {"frontend_input": "boolean", "backend_type": "int"}, "warranty": {"frontend_input": "text", "backend_type": "text"}, } def get_options(self, conn_id: int, code: str, ttl) -> Dict[str, int]: return {} class FakeAttrSetRepo: def get_attributes_for_set(self, conn_id: int, attribute_set_id: int) -> Dict[str, int]: return {"soft_close": 1, "warranty": 2} svc = MagentoSyncService( MagicMock(), {"id": 1}, attr_cache_repo=FakeAttrCacheRepo(), attr_set_attrs_repo=FakeAttrSetRepo(), ) svc._default_attribute_set_id = 4 payload = svc._transform_custom_attributes_for_magento( { "sku": "SKU-1", "custom_attributes": [ {"attribute_code": "soft_close", "value": "Yes"}, {"attribute_code": "warranty", "value": "5 Year Limited Warranty"}, ], }, 1, ) assert payload["custom_attributes"] == [ {"attribute_code": "soft_close", "value": 1}, {"attribute_code": "warranty", "value": "5 Year Limited Warranty"}, ] def test_transform_custom_attributes_drops_status_and_visibility(): """Master 'active' must never be resolved as Magento status select option.""" class FakeAttrCacheRepo: def get_attributes(self, conn_id: int, codes: List[str], ttl) -> Dict[str, dict]: return { "status": {"frontend_input": "select", "backend_type": "int"}, "visibility": {"frontend_input": "select", "backend_type": "int"}, "color": {"frontend_input": "select", "backend_type": "int"}, } def get_options(self, conn_id: int, code: str, ttl) -> Dict[str, int]: if code == "status": return {"enabled": 1, "disabled": 2} if code == "visibility": return {"not visible individually": 1, "catalog, search": 4} if code == "color": return {"white": 10} return {} class FakeAttrSetRepo: def get_attributes_for_set(self, conn_id: int, attribute_set_id: int) -> Dict[str, int]: return {"status": 1, "visibility": 2, "color": 3} svc = MagentoSyncService( MagicMock(), {"id": 1}, attr_cache_repo=FakeAttrCacheRepo(), attr_set_attrs_repo=FakeAttrSetRepo(), ) svc._default_attribute_set_id = 4 payload = svc._transform_custom_attributes_for_magento( { "sku": "SKU-1", "status": 1, "visibility": 4, "custom_attributes": [ {"attribute_code": "status", "value": "active"}, {"attribute_code": "visibility", "value": "Catalog, Search"}, {"attribute_code": "color", "value": "White"}, ], }, 1, ) assert payload["status"] == 1 assert payload["visibility"] == 4 assert payload["custom_attributes"] == [ {"attribute_code": "color", "value": "10"}, ]