Added seeding and e2e tests for commerce assets

Author: Robert Segal

e2e_config.test.json

@@ -26,6 +26,11 @@
   "catalog.unit.id": "UNT-1229",
   "commerce.agreement.attachment.id": "ATT-9850-2169-6098-0001",
   "commerce.agreement.id": "AGR-9850-2169-6098",
+  "commerce.assets.agreement.id": "AGR-2473-3299-1721",
+  "commerce.assets.id": "AST-0625-6526-6154",
+  "commerce.assets.order.id": "ORD-7707-7765-8445",
+  "commerce.assets.product.item.id": "ITM-1767-7355-0002",
+  "commerce.assets.product.template.id": "",
   "commerce.authorization.id": "AUT-0031-2873",
   "commerce.client.id": "ACC-1086-6867",
   "commerce.order.id": "ORD-6969-3541-5426",

mpt_api_client/resources/commerce/assets.py

@@ -49,7 +49,7 @@ def terminate(self, asset_id: str, resource_data: ResourceData | None = None) ->
         response = self._resource_do_request(asset_id, "POST", "terminate", json=resource_data)
         return self._model_class.from_response(response)
 
-    def render(self, asset_id: str) -> AssetTemplate:
+    def render(self, asset_id: str) -> str:
         """Renders the template for the given Asset id.
 
         Args:
@@ -60,7 +60,7 @@ def render(self, asset_id: str) -> AssetTemplate:
         """
         response = self._resource_do_request(asset_id, action="render")
 
-        return AssetTemplate.from_response(response)
+        return response.text
 
 
 class AsyncAssetService(
@@ -86,15 +86,15 @@ async def terminate(self, asset_id: str, resource_data: ResourceData | None = No
 
         return self._model_class.from_response(response)
 
-    async def render(self, asset_id: str) -> AssetTemplate:
+    async def render(self, asset_id: str) -> str:
         """Renders the template for the given Asset id.
 
         Args:
             asset_id: Asset ID.
 
         Returns:
-            Render asset template json.
+            Render asset template string.
         """
         response = await self._resource_do_request(asset_id, action="render")
 
-        return AssetTemplate.from_response(response)
+        return response.text

tests/e2e/commerce/asset/conftest.py

@@ -0,0 +1,37 @@
+import pytest
+
+
+@pytest.fixture
+def agreement_asset_id(e2e_config):
+    return e2e_config["commerce.assets.id"]
+
+
+@pytest.fixture
+def invalid_agreement_asset_id():
+    return "AST-0000-0000-0000"
+
+
+@pytest.fixture
+def asset_factory(asset_item_id, asset_agreement_id):
+    def factory(
+        name: str = "E2E Created Asset",
+        quantity: int = 1,
+        external_vendor_id: str = "ext-vendor-id",
+    ):
+        return {
+            "name": name,
+            "agreement": {"id": asset_agreement_id},
+            "externalIds": {"vendor": external_vendor_id},
+            "lines": [
+                {
+                    "item": {"id": asset_item_id},
+                    "quantity": quantity,
+                    "price": {
+                        "unitPP": 10,
+                        "PPx12": 10,
+                    },
+                }
+            ],
+        }
+
+    return factory

tests/e2e/commerce/asset/test_async_asset.py

@@ -0,0 +1,83 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+from mpt_api_client.rql.query_builder import RQLQuery
+
+pytestmark = [pytest.mark.flaky]
+
+
+@pytest.fixture
+async def created_asset(async_mpt_vendor, asset_factory):
+    new_asset_request_data = asset_factory(
+        name="E2E Created Asset",
+    )
+
+    result = await async_mpt_vendor.commerce.assets.create(new_asset_request_data)
+
+    yield result
+
+    try:
+        await async_mpt_vendor.commerce.assets.terminate(result.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to terminate asset: {getattr(error, 'title', str(error))}")  # noqa: WPS421
+
+
+async def test_get_asset_by_id(async_mpt_vendor, agreement_asset_id):
+    result = await async_mpt_vendor.commerce.assets.get(agreement_asset_id)
+
+    assert result is not None
+
+
+async def test_list_assets(async_mpt_vendor):
+    limit = 10
+
+    result = await async_mpt_vendor.commerce.assets.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_get_asset_by_id_not_found(async_mpt_vendor, invalid_agreement_asset_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await async_mpt_vendor.commerce.assets.get(invalid_agreement_asset_id)
+
+
+async def test_filter_assets(async_mpt_vendor, agreement_asset_id):
+    select_fields = ["-externalIds"]
+    filtered_assets = (
+        async_mpt_vendor.commerce.assets.filter(RQLQuery(id=agreement_asset_id))
+        .filter(RQLQuery(name="E2E Seeded Order Asset"))
+        .select(*select_fields)
+    )
+
+    result = [asset async for asset in filtered_assets.iterate()]
+
+    assert len(result) == 1
+
+
+def test_create_asset(created_asset):
+    result = created_asset
+
+    assert result is not None
+
+
+async def test_update_asset(async_mpt_vendor, created_asset):
+    updated_asset_data = {
+        "name": "E2E Updated Asset",
+        "parameters": {"fulfillment": []},
+    }
+
+    result = await async_mpt_vendor.commerce.assets.update(created_asset.id, updated_asset_data)
+
+    assert result is not None
+
+
+async def test_terminate_asset(async_mpt_vendor, created_asset):
+    result = await async_mpt_vendor.commerce.assets.terminate(created_asset.id)
+
+    assert result is not None
+
+
+async def test_render_asset(async_mpt_vendor, created_asset):
+    result = await async_mpt_vendor.commerce.assets.render(created_asset.id)
+
+    assert result is not None

tests/e2e/commerce/asset/test_sync_asset.py

@@ -0,0 +1,83 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+from mpt_api_client.rql.query_builder import RQLQuery
+
+pytestmark = [pytest.mark.flaky]
+
+
+@pytest.fixture
+def created_asset(mpt_vendor, asset_factory):
+    new_asset_request_data = asset_factory(
+        name="E2E Created Asset",
+    )
+
+    result = mpt_vendor.commerce.assets.create(new_asset_request_data)
+
+    yield result
+
+    try:
+        mpt_vendor.commerce.assets.terminate(result.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to terminate asset: {getattr(error, 'title', str(error))}")  # noqa: WPS421
+
+
+def test_get_asset_by_id(mpt_vendor, agreement_asset_id):
+    result = mpt_vendor.commerce.assets.get(agreement_asset_id)
+
+    assert result is not None
+
+
+def test_list_assets(mpt_vendor):
+    limit = 10
+
+    result = mpt_vendor.commerce.assets.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+def test_get_asset_by_id_not_found(mpt_vendor, invalid_agreement_asset_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        mpt_vendor.commerce.assets.get(invalid_agreement_asset_id)
+
+
+def test_filter_assets(mpt_vendor, agreement_asset_id):
+    select_fields = ["-externalIds"]
+    filtered_assets = (
+        mpt_vendor.commerce.assets.filter(RQLQuery(id=agreement_asset_id))
+        .filter(RQLQuery(name="E2E Seeded Order Asset"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_assets.iterate())
+
+    assert len(result) == 1
+
+
+def test_create_asset(created_asset):
+    result = created_asset
+
+    assert result is not None
+
+
+def test_update_asset(mpt_vendor, created_asset):
+    updated_asset_data = {
+        "name": "E2E Updated Asset",
+        "parameters": {"fulfillment": []},
+    }
+
+    result = mpt_vendor.commerce.assets.update(created_asset.id, updated_asset_data)
+
+    assert result is not None
+
+
+def test_terminate_asset(mpt_vendor, created_asset):
+    result = mpt_vendor.commerce.assets.terminate(created_asset.id)
+
+    assert result is not None
+
+
+def test_render_asset(mpt_vendor, created_asset):
+    result = mpt_vendor.commerce.assets.render(created_asset.id)
+
+    assert result is not None

tests/e2e/commerce/conftest.py

@@ -39,3 +39,13 @@ def subscription_item_id(e2e_config):
 @pytest.fixture
 def subscription_agreement_id(e2e_config):
     return e2e_config["commerce.subscription.agreement.id"]
+
+
+@pytest.fixture
+def asset_item_id(e2e_config):
+    return e2e_config["commerce.assets.product.item.id"]
+
+
+@pytest.fixture
+def asset_agreement_id(e2e_config):
+    return e2e_config["commerce.assets.agreement.id"]