Added e2e tests for commerce order subscription

Author: Robert Segal

e2e_config.test.json

@@ -26,6 +26,7 @@
   "catalog.unit.id": "UNT-1229",
   "commerce.agreement.attachment.id": "ATT-9850-2169-6098-0001",
   "commerce.agreement.id": "AGR-9850-2169-6098",
+  "commerce.agreement.subscription.line.id": "ALI-9850-2169-6098-0001",
   "commerce.assets.agreement.id": "AGR-2473-3299-1721",
   "commerce.assets.id": "AST-0625-6526-6154",
   "commerce.assets.order.id": "ORD-7707-7765-8445",

tests/e2e/commerce/conftest.py

@@ -1,4 +1,5 @@
 import pytest
+from freezegun import freeze_time
 
 
 @pytest.fixture
@@ -49,3 +50,47 @@ def asset_item_id(e2e_config):
 @pytest.fixture
 def asset_agreement_id(e2e_config):
     return e2e_config["commerce.assets.agreement.id"]
+
+
+@pytest.fixture
+def subscription_id(e2e_config):
+    return e2e_config["commerce.subscription.id"]
+
+
+@pytest.fixture
+def agreement_subscription_line_id(e2e_config):
+    return e2e_config["commerce.agreement.subscription.line.id"]
+
+
+@pytest.fixture
+def invalid_subscription_id():
+    return "SUB-0000-0000-0000"
+
+
+@pytest.fixture
+def subscription_factory(subscription_agreement_id, subscription_item_id):
+    @freeze_time("2025-11-14T09:00:00.000Z")
+    def factory(
+        name: str = "E2E Created Subscription",
+        external_vendor_id: str = "ext-vendor-id",
+        quantity: int = 1,
+    ):
+        return {
+            "name": name,
+            "startDate": "2025-11-03T09:00:00.000Z",
+            "commitmentDate": "2026-11-02T09:00:00.000Z",
+            "autoRenew": True,
+            "agreement": {"id": subscription_agreement_id},
+            "externalIds": {"vendor": external_vendor_id},
+            "template": None,
+            "lines": [
+                {
+                    "item": {"id": subscription_item_id},
+                    "quantity": quantity,
+                    "price": {"unitPP": 10},
+                }
+            ],
+            "parameters": {"fulfillment": []},
+        }
+
+    return factory

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

@@ -0,0 +1,24 @@
+import pytest
+from freezegun import freeze_time
+
+
+@pytest.fixture
+def order_subscription_factory(agreement_subscription_line_id):
+    @freeze_time("2025-11-14T09:00:00.000Z")
+    def factory(
+        name: str = "E2E Created Order Subscription",
+        external_vendor_id: str = "ext-vendor-id",
+        quantity: int = 1,
+    ):
+        return {
+            "name": name,
+            "startDate": "2025-11-03T09:00:00.000Z",
+            "commitmentDate": "2026-11-02T09:00:00.000Z",
+            "autoRenew": True,
+            "externalIds": {"vendor": external_vendor_id},
+            "template": None,
+            "lines": [{"id": agreement_subscription_line_id}],
+            "parameters": {"fulfillment": []},
+        }
+
+    return factory

tests/e2e/commerce/order/subscription/test_async_subscription.py

@@ -0,0 +1,95 @@
+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_subscription(async_mpt_vendor, order_subscription_factory, order_id):
+    # Must use this fixture for all tests to prevent api failures
+    subscription_data = order_subscription_factory()
+    orders = async_mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+    subscription = await subscriptions.create(subscription_data)
+
+    yield subscription
+
+    try:
+        await subscriptions.delete(subscription.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete subscription: {getattr(error, 'title', str(error))}")  # noqa: WPS421
+
+
+async def test_get_order_subscription_by_id(async_mpt_vendor, created_order_subscription, order_id):
+    subscription_id = created_order_subscription.id
+    subscriptions = async_mpt_vendor.commerce.orders.subscriptions(order_id)
+
+    result = await subscriptions.get(subscription_id)
+
+    assert result is not None
+
+
+async def test_list_order_subscriptions(async_mpt_vendor, created_order_subscription, order_id):
+    limit = 10
+    orders = async_mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+
+    result = await subscriptions.fetch_page(limit=limit)
+
+    assert result is not None
+
+
+async def test_get_order_subscription_by_id_not_found(
+    async_mpt_vendor, created_order_subscription, order_id, invalid_subscription_id
+):
+    orders = async_mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+
+    with pytest.raises(MPTAPIError, match="404 Not Found"):
+        await subscriptions.get(invalid_subscription_id)
+
+
+async def test_filter_order_subscriptions(
+    async_mpt_vendor, created_order_subscription, order_id, subscription_id
+):
+    select_fields = ["-externalIds"]
+    subscription_id = created_order_subscription.id
+    subscriptions = async_mpt_vendor.commerce.orders.subscriptions(order_id)
+    filtered_subscriptions = (
+        subscriptions.filter(RQLQuery(id=subscription_id))
+        .filter(RQLQuery(name="E2E Created Order Subscription"))
+        .select(*select_fields)
+    )
+
+    result = [subscription async for subscription in filtered_subscriptions.iterate()]
+
+    assert len(result) == 1
+
+
+def test_create_order_subscription(created_order_subscription):
+    result = created_order_subscription
+
+    assert result is not None
+
+
+async def test_update_order_subscription(async_mpt_vendor, created_order_subscription, order_id):
+    subscription_id = created_order_subscription.id
+    updated_subscription_data = {
+        "name": "E2E Updated Order Subscription",
+    }
+    orders = async_mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+
+    result = await subscriptions.update(subscription_id, updated_subscription_data)
+
+    assert result is not None
+
+
+async def test_delete_order_subscription(async_mpt_vendor, created_order_subscription, order_id):
+    subscription_id = created_order_subscription.id
+    orders = async_mpt_vendor.commerce.orders
+
+    subscriptions = orders.subscriptions(order_id)
+    await subscriptions.delete(subscription_id)

tests/e2e/commerce/order/subscription/test_sync_subscription.py

@@ -0,0 +1,96 @@
+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_subscription(mpt_vendor, order_subscription_factory, order_id):
+    # Must use this fixture for all tests to prevent api failures
+    subscription_data = order_subscription_factory()
+    orders = mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+    subscription = subscriptions.create(subscription_data)
+
+    yield subscription
+
+    try:
+        subscriptions.delete(subscription.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete subscription: {getattr(error, 'title', str(error))}")  # noqa: WPS421
+
+
+def test_get_order_subscription_by_id(mpt_vendor, created_order_subscription, order_id):
+    subscription_id = created_order_subscription.id
+    subscriptions = mpt_vendor.commerce.orders.subscriptions(order_id)
+
+    result = subscriptions.get(subscription_id)
+
+    assert result is not None
+
+
+def test_list_order_subscriptions(mpt_vendor, created_order_subscription, order_id):
+    limit = 10
+    orders = mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+
+    result = subscriptions.fetch_page(limit=limit)
+
+    assert result is not None
+
+
+def test_get_order_subscription_by_id_not_found(
+    mpt_vendor, created_order_subscription, order_id, invalid_subscription_id
+):
+    orders = mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+
+    with pytest.raises(MPTAPIError, match="404 Not Found"):
+        subscriptions.get(invalid_subscription_id)
+
+
+def test_filter_order_subscriptions(
+    mpt_vendor, created_order_subscription, order_id, subscription_id
+):
+    select_fields = ["-externalIds"]
+    subscription_id = created_order_subscription.id
+    subscriptions = mpt_vendor.commerce.orders.subscriptions(order_id)
+    filtered_subscriptions = (
+        subscriptions.filter(RQLQuery(id=subscription_id))
+        .filter(RQLQuery(name="E2E Created Order Subscription"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_subscriptions.iterate())
+
+    assert len(result) == 1
+
+
+def test_create_order_subscription(created_order_subscription):
+    result = created_order_subscription
+
+    assert result is not None
+
+
+def test_update_order_subscription(mpt_vendor, created_order_subscription, order_id):
+    subscription_id = created_order_subscription.id
+    updated_subscription_data = {
+        "name": "E2E Updated Order Subscription",
+    }
+    orders = mpt_vendor.commerce.orders
+    subscriptions = orders.subscriptions(order_id)
+
+    result = subscriptions.update(subscription_id, updated_subscription_data)
+
+    assert result is not None
+
+
+def test_delete_order_subscription(mpt_vendor, created_order_subscription, order_id):
+    subscription_id = created_order_subscription.id
+    orders = mpt_vendor.commerce.orders
+
+    result = orders.subscriptions(order_id)
+
+    result.delete(subscription_id)