"""Tests for Magento async bulk product writer.""" from __future__ import annotations import json from typing import Any, Dict, List, Optional, Tuple from magento.product_bulk_writer import ( BulkProductOp, MagentoProductBulkWriter, sku_from_bulk_operation, ) class FakeBulkApi: def __init__(self) -> None: self.put_calls: List[List[Dict[str, Any]]] = [] self.post_calls: List[List[Dict[str, Any]]] = [] self.status_bodies: Dict[str, Dict[str, Any]] = {} self.existing_skus: Dict[str, Dict[str, Any]] = {} self.put_uuid = "put-uuid-1" self.post_uuid = "post-uuid-1" def put_products_bulk_async( self, products: List[Dict[str, Any]] ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: self.put_calls.append(products) return 200, { "bulk_uuid": self.put_uuid, "request_items": [{"id": i, "status": "accepted"} for i in range(len(products))], "errors": False, }, None def post_products_bulk_async( self, products: List[Dict[str, Any]] ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: self.post_calls.append(products) return 200, { "bulk_uuid": self.post_uuid, "request_items": [{"id": i, "status": "accepted"} for i in range(len(products))], "errors": False, }, None def get_bulk_status( self, bulk_uuid: str ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: return 200, self.status_bodies.get(bulk_uuid, {}), None def get_bulk_detailed_status( self, bulk_uuid: str ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: return 200, self.status_bodies.get(bulk_uuid, {}), None def get_product(self, sku: str) -> Optional[Dict[str, Any]]: return self.existing_skus.get(sku) def _op(sku: str) -> BulkProductOp: return BulkProductOp( sku=sku, put_payload={"sku": sku, "name": f"Name {sku}", "status": 1}, create_payload={"sku": sku, "name": f"Name {sku}", "status": 1, "attribute_set_id": 4}, row={"sku": sku}, ) def _op_status(sku: str, status: int, message: str = "Service execution success") -> Dict[str, Any]: return { "id": 1, "status": status, "result_message": message, "serialized_data": json.dumps({"meta_information": {"product": {"sku": sku}}}), } def test_sku_from_bulk_operation_reads_meta_information(): op = { "serialized_data": json.dumps({"meta_information": {"product": {"sku": "ABC-1"}}}), } assert sku_from_bulk_operation(op) == "ABC-1" def test_bulk_writer_all_put_success(): api = FakeBulkApi() api.status_bodies[api.put_uuid] = { "operations_list": [ _op_status("SKU1", 1), _op_status("SKU2", 1), ], "operation_count": 2, } writer = MagentoProductBulkWriter(api=api, poll_interval_s=0, sleep=lambda _: None) results = writer.execute([_op("SKU1"), _op("SKU2")]) assert [r["status"] for r in results] == ["success", "success"] assert all(r["is_new"] is False for r in results) assert len(api.put_calls) == 1 assert api.post_calls == [] def test_bulk_writer_put_missing_falls_back_to_post(): api = FakeBulkApi() api.status_bodies[api.put_uuid] = { "operations_list": [ _op_status("SKU1", 1), _op_status("SKU2", 3, "No such entity with sku = SKU2"), ], "operation_count": 2, } api.status_bodies[api.post_uuid] = { "operations_list": [ _op_status("SKU2", 1, "Service execution success Magento\\Catalog\\Model\\ProductRepository::save"), ], "operation_count": 1, } writer = MagentoProductBulkWriter(api=api, poll_interval_s=0, sleep=lambda _: None) results = writer.execute([_op("SKU1"), _op("SKU2")]) assert results[0]["status"] == "success" assert results[0]["is_new"] is False assert results[1]["status"] == "success" assert results[1]["is_new"] is True assert len(api.post_calls) == 1 assert api.post_calls[0][0]["sku"] == "SKU2" def test_bulk_writer_partial_failure_preserves_per_sku_errors(): api = FakeBulkApi() api.status_bodies[api.put_uuid] = { "operations_list": [ _op_status("SKU1", 1), _op_status("SKU2", 3, "Invalid attribute value"), ], "operation_count": 2, } writer = MagentoProductBulkWriter(api=api, poll_interval_s=0, sleep=lambda _: None) results = writer.execute([_op("SKU1"), _op("SKU2")]) assert results[0]["status"] == "success" assert results[1]["status"] == "failed" assert "Invalid attribute value" in results[1]["error"] assert api.post_calls == [] def test_bulk_writer_positional_fallback_when_serialized_data_missing(): api = FakeBulkApi() api.status_bodies[api.put_uuid] = { "operations_list": [ {"id": 10, "status": 1, "result_message": "ok"}, {"id": 11, "status": 1, "result_message": "ok"}, ], "operation_count": 2, } writer = MagentoProductBulkWriter(api=api, poll_interval_s=0, sleep=lambda _: None) results = writer.execute([_op("A"), _op("B")]) assert [r["status"] for r in results] == ["success", "success"] def test_bulk_writer_timeout_marks_open_as_failed(): api = FakeBulkApi() api.status_bodies[api.put_uuid] = { "operations_list": [ _op_status("SKU1", 4, "Open"), ], "operation_count": 1, } writer = MagentoProductBulkWriter( api=api, poll_interval_s=0, poll_timeout_s=0, # immediate timeout sleep=lambda _: None, ) results = writer.execute([_op("SKU1")]) assert results[0]["status"] == "failed" assert "timed out" in results[0]["error"] def test_bulk_writer_refuses_create_when_get_finds_existing_after_invalid_product_data(): api = FakeBulkApi() api.existing_skus["SKU2"] = {"sku": "SKU2", "id": 99} api.status_bodies[api.put_uuid] = { "operations_list": [ _op_status("SKU2", 3, "Invalid product data"), ], "operation_count": 1, } writer = MagentoProductBulkWriter(api=api, poll_interval_s=0, sleep=lambda _: None) results = writer.execute([_op("SKU2")]) assert results[0]["status"] == "failed" assert results[0]["needs_single_fallback"] is True assert "refusing POST create" in results[0]["error"] assert api.post_calls == [] def test_bulk_writer_skips_post_when_product_appears_before_create(): api = FakeBulkApi() api.status_bodies[api.put_uuid] = { "operations_list": [ _op_status("SKU2", 3, "No such entity with sku = SKU2"), ], "operation_count": 1, } # Product is created by a concurrent worker between PUT failure and POST. api.existing_skus["SKU2"] = {"sku": "SKU2", "id": 100} writer = MagentoProductBulkWriter(api=api, poll_interval_s=0, sleep=lambda _: None) results = writer.execute([_op("SKU2")]) assert results[0]["needs_single_fallback"] is True assert api.post_calls == []