MPT-14890 Tests for catalog/product/media

- Add E2E tests
- Fix async service for catalog/product/media

Author: albertsola

mpt_api_client/resources/catalog/mixins.py

@@ -169,7 +169,7 @@ class MediaMixin[Model](
     DownloadFileMixin[Model],
     PublishableMixin[Model],
 ):
-    """Document mixin."""
+    """Media mixin."""
 
     _upload_file_key = "file"
     _upload_data_key = "media"
@@ -227,3 +227,13 @@ async def deactivate(
         return await self._resource_action(  # type: ignore[attr-defined, no-any-return]
             resource_id, "POST", "deactivate", json=resource_data
         )
+
+class AsyncMediaMixin[Model](
+    AsyncCreateFileMixin[Model],
+    AsyncDownloadFileMixin[Model],
+    AsyncPublishableMixin[Model],
+):
+    """Media mixin."""
+
+    _upload_file_key = "file"
+    _upload_data_key = "media"

mpt_api_client/resources/catalog/products_media.py

@@ -12,6 +12,7 @@
 )
 from mpt_api_client.models import Model, ResourceData
 from mpt_api_client.resources.catalog.mixins import (
+    AsyncMediaMixin,
     AsyncPublishableMixin,
     MediaMixin,
 )
@@ -40,30 +41,10 @@ class MediaService(
 
 
 class AsyncMediaService(
-    AsyncFilesOperationsMixin[Media],
-    AsyncPublishableMixin[Media],
+    AsyncMediaMixin[Media],
     AsyncModifiableResourceMixin[Media],
     AsyncCollectionMixin[Media],
     AsyncService[Media],
     MediaServiceConfig,
 ):
     """Media service."""
-
-    @override
-    async def create(
-        self,
-        resource_data: ResourceData | None = None,
-        files: dict[str, FileTypes] | None = None,
-        data_key: str = "_media_data",
-    ) -> Media:
-        """Create Media resource.
-
-        Args:
-            resource_data: Resource data.
-            files: Files data.
-            data_key: Key to use for the JSON data in the multipart form.
-
-        Returns:
-            Media resource.
-        """
-        return await super().create(resource_data=resource_data, files=files, data_key=data_key)

tests/e2e/catalog/product/media/conftest.py

@@ -0,0 +1,19 @@
+import pytest
+
+
+@pytest.fixture
+def media_data():
+    return {
+        "name": "e2e test media - please delete",
+        "description": "E2E test media for automated testing",
+        "displayOrder": 1,
+        "type": "Image",
+        "mediatype": "Image",
+        "url": "",
+        "language": "en-gb",
+    }
+
+
+@pytest.fixture
+def test_media_file(logo_fd):
+    return logo_fd

tests/e2e/catalog/product/media/test_async_media.py

@@ -0,0 +1,82 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+
+pytestmark = [pytest.mark.flaky, pytest.mark.asyncio]
+
+
+@pytest.fixture
+def async_media_service(async_mpt_vendor, product_id):
+    return async_mpt_vendor.catalog.products.media(product_id)
+
+
+@pytest.fixture
+def async_vendor_media_service(async_mpt_vendor, product_id):
+    return async_mpt_vendor.catalog.products.media(product_id)
+
+
+@pytest.fixture
+async def created_media_from_file_async(logger, async_media_service, media_data, test_media_file):
+    media = await async_media_service.create(media_data, test_media_file)
+    yield media
+    try:
+        await async_media_service.delete(media.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete media {media.id}: {error.title}")
+
+
+@pytest.fixture
+async def created_media_from_url_async(logger, async_media_service, media_data, jpg_url):
+    media_data["url"] = jpg_url
+    media = await async_media_service.create(media_data)
+    yield media
+    try:
+        await async_media_service.delete(media.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete media {media.id}: {error.title}")
+
+
+def test_create_media_async(created_media_from_file_async, media_data):
+    assert created_media_from_file_async.name == media_data["name"]
+    assert created_media_from_file_async.description == media_data["description"]
+
+
+def test_create_media_async_from_url(created_media_from_file_async, media_data):
+    assert created_media_from_file_async.name == media_data["name"]
+    assert created_media_from_file_async.description == media_data["description"]
+
+
+async def test_update_media_async(async_media_service, created_media_from_file_async):
+    update_data = {"name": "Updated e2e test media - please delete"}
+    media = await async_media_service.update(created_media_from_file_async.id, update_data)
+    assert media.name == update_data["name"]
+
+
+async def test_media_lifecycle_async(async_mpt_vendor, async_mpt_ops, created_media_from_file_async):
+    await async_mpt_vendor.catalog.products.media(created_media_from_file_async.product.id).review(created_media_from_file_async.id)
+    await async_mpt_ops.catalog.products.media(created_media_from_file_async.product.id).publish(created_media_from_file_async.id)
+    await async_mpt_vendor.catalog.products.media(created_media_from_file_async.product.id).unpublish(
+        created_media_from_file_async.id
+    )
+
+async def test_delete_media_async(async_vendor_media_service, created_media_from_file_async):
+    await async_vendor_media_service.delete(created_media_from_file_async.id)
+    with pytest.raises(MPTAPIError):
+        await async_vendor_media_service.get(created_media_from_file_async.id)
+
+
+async def test_get_media_async(async_vendor_media_service, created_media_from_file_async):
+    media = await async_vendor_media_service.get(created_media_from_file_async.id)
+    assert media.id == created_media_from_file_async.id
+
+
+async def test_download_media_async(async_vendor_media_service, created_media_from_file_async):
+    file_response = await async_vendor_media_service.download(created_media_from_file_async.id)
+    assert file_response.file_contents is not None
+    assert file_response.filename == "logo.png"
+
+
+async def test_get_not_found_media_async(async_vendor_media_service):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await async_vendor_media_service.get("INVALID-ID")
+

tests/e2e/catalog/product/media/test_sync_media.py

@@ -0,0 +1,76 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+
+pytestmark = [pytest.mark.flaky]
+
+
+@pytest.fixture
+def vendor_media_service(mpt_vendor, product_id):
+    return mpt_vendor.catalog.products.media(product_id)
+
+
+@pytest.fixture
+def created_media_from_file(logger, vendor_media_service, media_data, test_media_file):
+    media = vendor_media_service.create(media_data, test_media_file)
+    yield media
+    try:
+        vendor_media_service.delete(media.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete media {media.id}: {error.title}")
+
+@pytest.fixture
+def created_media_from_url(logger, vendor_media_service, media_data, jpg_url):
+    media_data["url"] = jpg_url
+    media = vendor_media_service.create(media_data)
+    yield media
+    try:
+        vendor_media_service.delete(media.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete media {media.id}: {error.title}")
+
+
+def test_create_media(created_media_from_file, media_data):
+    assert created_media_from_file.name == media_data["name"]
+    assert created_media_from_file.description == media_data["description"]
+
+
+def test_create_media_from_url(created_media_from_file, media_data):
+    assert created_media_from_file.name == media_data["name"]
+    assert created_media_from_file.description == media_data["description"]
+
+
+
+def test_update_media(vendor_media_service, created_media_from_file):
+    update_data = {"name": "Updated e2e test media - please delete"}
+    media = vendor_media_service.update(created_media_from_file.id, update_data)
+    assert media.name == update_data["name"]
+
+
+def test_media_lifecycle(mpt_vendor, mpt_ops, created_media_from_file):
+    mpt_vendor.catalog.products.media(created_media_from_file.product.id).review(created_media_from_file.id)
+    mpt_ops.catalog.products.media(created_media_from_file.product.id).publish(created_media_from_file.id)
+    mpt_vendor.catalog.products.media(created_media_from_file.product.id).unpublish(
+        created_media_from_file.id
+    )
+
+def test_delete_media(vendor_media_service, created_media_from_file):
+    vendor_media_service.delete(created_media_from_file.id)
+    with pytest.raises(MPTAPIError):
+        vendor_media_service.get(created_media_from_file.id)
+
+
+def test_get_media(vendor_media_service, created_media_from_file):
+    media = vendor_media_service.get(created_media_from_file.id)
+    assert media.id == created_media_from_file.id
+
+
+def test_download_media(vendor_media_service, created_media_from_file):
+    file_response = vendor_media_service.download(created_media_from_file.id)
+    assert file_response.file_contents is not None
+    assert file_response.filename == "logo.png"
+
+
+async def test_get_not_found_media(vendor_media_service):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await vendor_media_service.get("INVALID-ID")

tests/e2e/conftest.py

@@ -79,6 +79,9 @@ def pdf_fd():
 def pdf_url():
     return "https://sample-files.com/downloads/documents/pdf/basic-text.pdf"
 
+@pytest.fixture
+def jpg_url():
+    return "https://sample-files.com/downloads/images/jpg/color_test_800x600_118kb.jpg"
 
 @pytest.fixture
 def e2e_config(project_root_path):

tests/unit/http/test_mixins.py

@@ -235,18 +235,18 @@ async def test_async_file_create_with_data(async_media_service):
                 json=media_data,
             )
         )
-        files = {"media": ("test.jpg", io.BytesIO(b"Image content"), "image/jpeg")}
-        new_media = await async_media_service.create({"name": "Product image"}, files=files)
+        file = ("test.jpg", io.BytesIO(b"Image content"), "image/jpeg")
+        new_media = await async_media_service.create({"name": "Product image"}, file=file)
 
     request: httpx.Request = mock_route.calls[0].request
 
     assert (
-        b'Content-Disposition: form-data; name="_media_data"\r\n'
+        b'Content-Disposition: form-data; name="media"\r\n'
         b"Content-Type: application/json\r\n\r\n"
         b'{"name":"Product image"}\r\n' in request.content
     )
     assert (
-        b'Content-Disposition: form-data; name="media"; filename="test.jpg"\r\n'
+        b'Content-Disposition: form-data; name="file"; filename="test.jpg"\r\n'
         b"Content-Type: image/jpeg\r\n\r\n"
         b"Image content\r\n" in request.content
     )
@@ -265,13 +265,12 @@ async def test_async_file_create_no_data(async_media_service):
                 json=media_data,
             )
         )
-        files = {"media": ("test.jpg", io.BytesIO(b"Image content"), "image/jpeg")}
-        new_media = await async_media_service.create(files=files)
+        new_media = await async_media_service.create({}, file=("test.jpg", io.BytesIO(b"Image content"), "image/jpeg"))
 
     request: httpx.Request = mock_route.calls[0].request
 
     assert (
-        b'Content-Disposition: form-data; name="media"; filename="test.jpg"\r\n'
+        b'Content-Disposition: form-data; name="file"; filename="test.jpg"\r\n'
         b"Content-Type: image/jpeg\r\n\r\n"
         b"Image content\r\n" in request.content
     )