MPT-14900 E2E tests for catalog pricing-policies attachments (#133)

Author: albertsola

mpt_api_client/resources/catalog/pricing_policy_attachments.py

@@ -1,14 +1,17 @@
 from mpt_api_client.http import AsyncService, Service
 from mpt_api_client.http.mixins import (
     AsyncCollectionMixin,
-    AsyncFilesOperationsMixin,
+    AsyncDownloadFileMixin,
     AsyncModifiableResourceMixin,
     CollectionMixin,
-    FilesOperationsMixin,
+    DownloadFileMixin,
     ModifiableResourceMixin,
 )
 from mpt_api_client.models import Model
-from mpt_api_client.resources.catalog.mixins import ActivatableMixin, AsyncActivatableMixin
+from mpt_api_client.resources.catalog.mixins import (
+    AsyncCreateFileMixin,
+    CreateFileMixin,
+)
 
 
 class PricingPolicyAttachment(Model):
@@ -21,11 +24,13 @@ class PricingPolicyAttachmentsServiceConfig:
     _endpoint = "/public/v1/catalog/pricing-policies/{pricing_policy_id}/attachments"
     _model_class = PricingPolicyAttachment
     _collection_key = "data"
+    _upload_file_key = "file"
+    _upload_data_key = "attachment"
 
 
 class PricingPolicyAttachmentsService(
-    FilesOperationsMixin[PricingPolicyAttachment],
-    ActivatableMixin[PricingPolicyAttachment],
+    CreateFileMixin[PricingPolicyAttachment],
+    DownloadFileMixin[PricingPolicyAttachment],
     ModifiableResourceMixin[PricingPolicyAttachment],
     CollectionMixin[PricingPolicyAttachment],
     Service[PricingPolicyAttachment],
@@ -35,8 +40,8 @@ class PricingPolicyAttachmentsService(
 
 
 class AsyncPricingPolicyAttachmentsService(
-    AsyncFilesOperationsMixin[PricingPolicyAttachment],
-    AsyncActivatableMixin[PricingPolicyAttachment],
+    AsyncCreateFileMixin[PricingPolicyAttachment],
+    AsyncDownloadFileMixin[PricingPolicyAttachment],
     AsyncModifiableResourceMixin[PricingPolicyAttachment],
     AsyncCollectionMixin[PricingPolicyAttachment],
     AsyncService[PricingPolicyAttachment],

tests/e2e/catalog/items/test_async_item.py

@@ -7,7 +7,7 @@
 
 
 @pytest.fixture
-async def async_created_item(logger, async_mpt_vendor, item_data):
+async def async_created_item(async_mpt_vendor, item_data):
     service = async_mpt_vendor.catalog.items
     item = await service.create(item_data)
     yield item

tests/e2e/catalog/items/test_sync_item.py

@@ -7,7 +7,7 @@
 
 
 @pytest.fixture
-def created_item(logger, mpt_vendor, item_data):
+def created_item(mpt_vendor, item_data):
     service = mpt_vendor.catalog.items
     item = service.create(item_data)
     yield item

tests/e2e/catalog/pricing_policies/attachments/__init__.py

@@ -0,0 +1,31 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+
+
+@pytest.fixture
+def attachment_id(created_attachment):
+    return created_attachment.id
+
+
+@pytest.fixture
+def attachment_data():
+    return {
+        "name": "e2e test attachment - please delete",
+        "description": "E2E test attachment for automated testing",
+    }
+
+
+@pytest.fixture
+def attachment_service(mpt_ops, pricing_policy_id):
+    return mpt_ops.catalog.pricing_policies.attachments(pricing_policy_id)
+
+
+@pytest.fixture
+def created_attachment(attachment_service, attachment_data, pdf_fd):
+    attachment = attachment_service.create(attachment_data, file=pdf_fd)
+    yield attachment
+    try:
+        attachment_service.delete(attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete attachment {attachment.id}: {error.title}")

tests/e2e/catalog/pricing_policies/attachments/conftest.py

@@ -0,0 +1,76 @@
+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 async_attachment_service(async_mpt_ops, pricing_policy_id):
+    return async_mpt_ops.catalog.pricing_policies.attachments(pricing_policy_id)
+
+
+@pytest.fixture
+async def created_attachment_async(async_attachment_service, attachment_data, pdf_fd):
+    attachment = await async_attachment_service.create(attachment_data, file=pdf_fd)
+    yield attachment
+    try:
+        await async_attachment_service.delete(attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete attachment {attachment.id}: {error.title}")
+
+
+def test_create_attachment_async(created_attachment_async, attachment_data):
+    result = created_attachment_async
+
+    assert result.name == attachment_data["name"]
+    assert result.description == attachment_data["description"]
+
+
+async def test_update_attachment_async(async_attachment_service, created_attachment_async):
+    update_data = {"name": "Updated e2e test attachment - please delete"}
+
+    result = await async_attachment_service.update(created_attachment_async.id, update_data)
+
+    assert result.name == update_data["name"]
+
+
+async def test_get_attachment_async(async_attachment_service, attachment_id):
+    result = await async_attachment_service.get(attachment_id)
+
+    assert result.id == attachment_id
+
+
+async def test_download_attachment_async(async_attachment_service, attachment_id):
+    result = await async_attachment_service.download(attachment_id)
+
+    assert result.file_contents is not None
+    assert result.filename == "empty.pdf"
+
+
+async def test_iterate_attachments_async(async_attachment_service, created_attachment_async):
+    result = [att async for att in async_attachment_service.iterate()]
+
+    assert any(att.id == created_attachment_async.id for att in result)
+
+
+async def test_filter_attachments_async(async_attachment_service, created_attachment_async):
+    filtered_service = async_attachment_service.filter(RQLQuery(id=created_attachment_async.id))
+
+    result = [att async for att in filtered_service.iterate()]
+
+    assert len(result) == 1
+    assert result[0].id == created_attachment_async.id
+
+
+async def test_not_found_async(async_attachment_service):
+    with pytest.raises(MPTAPIError):
+        await async_attachment_service.get("ATT-000-000-000")
+
+
+async def test_delete_attachment_async(async_attachment_service, created_attachment_async):
+    await async_attachment_service.delete(created_attachment_async.id)
+
+    with pytest.raises(MPTAPIError):
+        await async_attachment_service.get(created_attachment_async.id)

tests/e2e/catalog/pricing_policies/attachments/test_async_attachment.py

@@ -0,0 +1,59 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+from mpt_api_client.rql.query_builder import RQLQuery
+
+pytestmark = [pytest.mark.flaky]
+
+
+def test_create_attachment(created_attachment, attachment_data):
+    result = created_attachment
+
+    assert result.name == attachment_data["name"]
+    assert result.description == attachment_data["description"]
+
+
+def test_update_attachment(attachment_service, created_attachment):
+    update_data = {"name": "Updated e2e test attachment - please delete"}
+
+    result = attachment_service.update(created_attachment.id, update_data)
+
+    assert result.name == update_data["name"]
+
+
+def test_get_attachment(attachment_service, attachment_id):
+    result = attachment_service.get(attachment_id)
+
+    assert result.id == attachment_id
+
+
+def test_download_attachment(attachment_service, attachment_id):
+    result = attachment_service.download(attachment_id)
+
+    assert result.file_contents is not None
+    assert result.filename == "empty.pdf"
+
+
+def test_iterate_attachments(attachment_service, created_attachment):
+    result = list(attachment_service.iterate())
+
+    assert any(att.id == created_attachment.id for att in result)
+
+
+def test_filter_attachments(attachment_service, created_attachment):
+    result = list(attachment_service.filter(RQLQuery(id=created_attachment.id)).iterate())
+
+    assert len(result) == 1
+    assert result[0].id == created_attachment.id
+
+
+def test_not_found(attachment_service):
+    with pytest.raises(MPTAPIError):
+        attachment_service.get("ATT-000-000-000")
+
+
+def test_delete_attachment(attachment_service, created_attachment):
+    attachment_service.delete(created_attachment.id)
+
+    with pytest.raises(MPTAPIError):
+        attachment_service.get(created_attachment.id)

tests/e2e/catalog/pricing_policies/attachments/test_sync_attachment.py

@@ -1,5 +1,7 @@
 import pytest
 
+from mpt_api_client.exceptions import MPTAPIError
+
 
 @pytest.fixture
 def buyer_id(e2e_config):
@@ -19,5 +21,22 @@ def pricing_policy_data(buyer_id, product_id):
 
 
 @pytest.fixture
-def pricing_policy_id(e2e_config):
-    return e2e_config.get("catalog.pricing_policy.id")
+def pricing_policies_service(mpt_ops):
+    return mpt_ops.catalog.pricing_policies
+
+
+@pytest.fixture
+def created_pricing_policy(pricing_policies_service, pricing_policy_data):
+    policy = pricing_policies_service.create(pricing_policy_data)
+
+    yield policy
+
+    try:
+        pricing_policies_service.delete(policy.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete pricing policy {policy.id}: {error.title}")
+
+
+@pytest.fixture
+def pricing_policy_id(created_pricing_policy):
+    return created_pricing_policy.id

tests/e2e/catalog/pricing_policies/conftest.py

@@ -35,12 +35,10 @@ async def test_get_pricing_policy(async_pricing_policies_service, async_created_
     assert result.id == async_created_pricing_policy.id
 
 
-async def test_get_pricing_policy_by_id(
-    async_pricing_policies_service, async_created_pricing_policy
-):
-    result = await async_pricing_policies_service.get(async_created_pricing_policy.id)
+async def test_get_pricing_policy_by_id(async_pricing_policies_service, pricing_policy_id):
+    result = await async_pricing_policies_service.get(pricing_policy_id)
 
-    assert result.id == async_created_pricing_policy.id
+    assert result.id == pricing_policy_id
 
 
 async def test_iterate_pricing_policies(

tests/e2e/catalog/pricing_policies/test_async_pricing_policies.py

@@ -6,23 +6,6 @@
 pytestmark = [pytest.mark.flaky]
 
 
-@pytest.fixture
-def pricing_policies_service(mpt_ops):
-    return mpt_ops.catalog.pricing_policies
-
-
-@pytest.fixture
-def created_pricing_policy(pricing_policies_service, pricing_policy_data):
-    policy = pricing_policies_service.create(pricing_policy_data)
-
-    yield policy
-
-    try:
-        pricing_policies_service.delete(policy.id)
-    except MPTAPIError as error:
-        print(f"TEARDOWN - Unable to delete pricing policy {policy.id}: {error.title}")
-
-
 def test_create_pricing_policy(created_pricing_policy, pricing_policy_data):
     result = created_pricing_policy
 
@@ -36,9 +19,6 @@ def test_get_pricing_policy(pricing_policies_service, created_pricing_policy):
 
 
 def test_get_pricing_policy_by_id(pricing_policies_service, pricing_policy_id):
-    if not pricing_policy_id:
-        pytest.skip("No pricing_policy_id configured")
-
     result = pricing_policies_service.get(pricing_policy_id)
 
     assert result.id == pricing_policy_id