Added e2e tests for commerce order assets

Author: Robert Segal

e2e_config.test.json

@@ -27,8 +27,11 @@
   "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.agreement.line.id": "ALI-9320-4904-7940-0001",
   "commerce.assets.id": "AST-0625-6526-6154",
   "commerce.assets.order.id": "ORD-7707-7765-8445",
+  "commerce.assets.draft.order.agreement.id": "AGR-9320-4904-7940",
+  "commerce.assets.draft.order.id": "ORD-9272-6817-2305",
   "commerce.assets.product.item.id": "ITM-1767-7355-0002",
   "commerce.assets.product.template.id": "",
   "commerce.authorization.id": "AUT-0031-2873",

pyproject.toml

@@ -132,6 +132,7 @@ per-file-ignores = [
   "tests/e2e/commerce/agreement/*.py: WPS202",
   "tests/e2e/commerce/agreement/attachment/*.py: WPS202",
   "tests/e2e/commerce/order/*.py: WPS202 WPS204",
+  "tests/e2e/commerce/order/asset/*.py: WPS211 WPS202",
   "tests/e2e/commerce/subscription/*.py: WPS202",
   "tests/unit/http/test_async_service.py: WPS204 WPS202",
   "tests/unit/http/test_service.py: WPS204 WPS202",

tests/e2e/commerce/asset/conftest.py

@@ -1,16 +1,6 @@
 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(

tests/e2e/commerce/conftest.py

@@ -49,3 +49,13 @@ def asset_item_id(e2e_config):
 @pytest.fixture
 def asset_agreement_id(e2e_config):
     return e2e_config["commerce.assets.agreement.id"]
+
+
+@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"

tests/e2e/commerce/order/asset/conftest.py

@@ -0,0 +1,57 @@
+import pytest
+
+
+@pytest.fixture
+def draft_order_asset_agreement_id(e2e_config):
+    return e2e_config["commerce.assets.draft.order.agreement.id"]
+
+
+@pytest.fixture
+def asset_agreement_line_id(e2e_config):
+    return e2e_config["commerce.assets.agreement.line.id"]
+
+
+@pytest.fixture
+def commerce_asset_draft_order_id(e2e_config):
+    return e2e_config["commerce.assets.draft.order.id"]
+
+
+@pytest.fixture
+def order_asset_factory(
+    draft_order_asset_agreement_id,
+    buyer_id,
+    asset_agreement_line_id,
+    buyer_account_id,
+    seller_id,
+    commerce_product_id,
+    commerce_asset_draft_order_id,
+):
+    def factory(
+        name: str = "E2E Created Order Asset",
+        quantity: int = 1,
+        external_vendor_id: str = "ext-vendor-id",
+    ):
+        return {
+            "name": name,
+            "externalIds": {"vendor": external_vendor_id},
+            "lines": [
+                {
+                    "id": asset_agreement_line_id,
+                    "agreement": {"id": draft_order_asset_agreement_id},
+                    "buyer": {"id": buyer_id},
+                    "client": {"id": buyer_account_id},
+                    "oldQuantity": 0,
+                    "quantity": quantity,
+                    "price": {
+                        "unitPP": 10,
+                        "PPx1": 100,
+                        "currency": "USD",
+                    },
+                }
+            ],
+            "order": {"id": commerce_asset_draft_order_id},
+            "product": {"id": commerce_product_id},
+            "seller": {"id": seller_id},
+        }
+
+    return factory

tests/e2e/commerce/order/asset/test_async_asset.py

@@ -0,0 +1,122 @@
+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_order_asset(async_mpt_vendor, order_asset_factory, commerce_asset_draft_order_id):
+    # Must use this fixture for all tests to prevent api errors
+    asset_data = order_asset_factory()
+    orders = async_mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+    asset = await assets.create(asset_data)
+
+    yield asset
+
+    try:
+        await assets.delete(asset.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete order asset: {getattr(error, 'title', str(error))}")  # noqa: WPS421
+
+
+async def test_get_order_asset_by_id(
+    async_mpt_vendor, created_order_asset, commerce_asset_draft_order_id
+):
+    asset_id = created_order_asset.id
+    orders = async_mpt_vendor.commerce.orders.assets(commerce_asset_draft_order_id)
+
+    result = await orders.get(asset_id)
+
+    assert result is not None
+
+
+async def test_list_order_assets(
+    async_mpt_vendor, created_order_asset, commerce_asset_draft_order_id
+):
+    limit = 10
+    orders = async_mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    result = await assets.fetch_page(limit=limit)
+
+    assert result is not None
+
+
+async def test_get_order_asset_by_id_not_found(
+    async_mpt_vendor, created_order_asset, commerce_asset_draft_order_id, invalid_agreement_asset_id
+):
+    orders = async_mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    with pytest.raises(MPTAPIError, match="404 Not Found"):
+        await assets.get(invalid_agreement_asset_id)
+
+
+async def test_filter_order_assets(
+    async_mpt_vendor, created_order_asset, commerce_asset_draft_order_id
+):
+    select_fields = ["-externalIds"]
+    asset_id = created_order_asset.id
+    assets = async_mpt_vendor.commerce.orders.assets(commerce_asset_draft_order_id)
+    filtered_assets = (
+        assets.filter(RQLQuery(id=asset_id))
+        .filter(RQLQuery(name="E2E Created Order Asset"))
+        .select(*select_fields)
+    )
+
+    result = [asset async for asset in filtered_assets.iterate()]
+
+    assert len(result) == 1
+
+
+def test_create_order_asset(created_order_asset):
+    result = created_order_asset
+
+    assert result is not None
+
+
+async def test_update_order_asset(
+    async_mpt_vendor,
+    created_order_asset,
+    commerce_asset_draft_order_id,
+):
+    asset_id = created_order_asset.id
+    updated_asset_data = {
+        "name": "E2E Updated Order Asset",
+    }
+    orders = async_mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    result = await assets.update(asset_id, updated_asset_data)
+
+    assert result is not None
+
+
+async def test_delete_order_asset(
+    async_mpt_vendor,
+    created_order_asset,
+    commerce_asset_draft_order_id,
+):
+    asset_id = created_order_asset.id
+    orders = async_mpt_vendor.commerce.orders
+
+    result = orders.assets(commerce_asset_draft_order_id)
+
+    await result.delete(asset_id)
+
+
+async def test_render_order_asset(
+    async_mpt_vendor,
+    created_order_asset,
+    commerce_asset_draft_order_id,
+):
+    asset_id = created_order_asset.id
+    orders = async_mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    result = await assets.render(asset_id)
+
+    assert result is not None

tests/e2e/commerce/order/asset/test_sync_asset.py

@@ -0,0 +1,116 @@
+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_order_asset(mpt_vendor, order_asset_factory, commerce_asset_draft_order_id):
+    # Must use this fixture for all tests to prevent api errors
+    asset_data = order_asset_factory()
+    orders = mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+    asset = assets.create(asset_data)
+
+    yield asset
+
+    try:
+        assets.delete(asset.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete order asset: {getattr(error, 'title', str(error))}")  # noqa: WPS421
+
+
+def test_get_order_asset_by_id(mpt_vendor, created_order_asset, commerce_asset_draft_order_id):
+    asset_id = created_order_asset.id
+    orders = mpt_vendor.commerce.orders.assets(commerce_asset_draft_order_id)
+
+    result = orders.get(asset_id)
+
+    assert result is not None
+
+
+def test_list_order_assets(mpt_vendor, created_order_asset, commerce_asset_draft_order_id):
+    limit = 10
+    orders = mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    result = assets.fetch_page(limit=limit)
+
+    assert result is not None
+
+
+def test_get_order_asset_by_id_not_found(
+    mpt_vendor, created_order_asset, commerce_asset_draft_order_id, invalid_agreement_asset_id
+):
+    orders = mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    with pytest.raises(MPTAPIError, match="404 Not Found"):
+        assets.get(invalid_agreement_asset_id)
+
+
+def test_filter_order_assets(mpt_vendor, created_order_asset, commerce_asset_draft_order_id):
+    select_fields = ["-externalIds"]
+    asset_id = created_order_asset.id
+    assets = mpt_vendor.commerce.orders.assets(commerce_asset_draft_order_id)
+    filtered_assets = (
+        assets.filter(RQLQuery(id=asset_id))
+        .filter(RQLQuery(name="E2E Created Order Asset"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_assets.iterate())
+
+    assert len(result) == 1
+
+
+def test_create_order_asset(created_order_asset):
+    result = created_order_asset
+
+    assert result is not None
+
+
+def test_update_order_asset(
+    mpt_vendor,
+    created_order_asset,
+    commerce_asset_draft_order_id,
+):
+    asset_id = created_order_asset.id
+    updated_asset_data = {
+        "name": "E2E Updated Order Asset",
+    }
+    orders = mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    result = assets.update(asset_id, updated_asset_data)
+
+    assert result is not None
+
+
+def test_delete_order_asset(
+    mpt_vendor,
+    created_order_asset,
+    commerce_asset_draft_order_id,
+):
+    asset_id = created_order_asset.id
+    orders = mpt_vendor.commerce.orders
+
+    result = orders.assets(commerce_asset_draft_order_id)
+
+    result.delete(asset_id)
+
+
+def test_render_order_asset(
+    mpt_vendor,
+    created_order_asset,
+    commerce_asset_draft_order_id,
+):
+    asset_id = created_order_asset.id
+    orders = mpt_vendor.commerce.orders
+    assets = orders.assets(commerce_asset_draft_order_id)
+
+    result = assets.render(asset_id)
+
+    assert result is not None