MPT-19905: add /integration/extensions/{id}/media service

mpt_api_client/resources/integration/extension_media.py

@@ -0,0 +1,91 @@
+from mpt_api_client.http import AsyncService, Service
+from mpt_api_client.http.mixins import (
+    AsyncCollectionMixin,
+    AsyncCreateFileMixin,
+    AsyncDeleteMixin,
+    AsyncDownloadFileMixin,
+    AsyncModifiableResourceMixin,
+    CollectionMixin,
+    CreateFileMixin,
+    DeleteMixin,
+    DownloadFileMixin,
+    ModifiableResourceMixin,
+)
+from mpt_api_client.models import Model
+from mpt_api_client.models.model import BaseModel
+from mpt_api_client.resources.integration.mixins import (
+    AsyncMediaMixin,
+    AsyncPublishableMixin,
+    MediaMixin,
+    PublishableMixin,
+)
+
+
+class ExtensionMedia(Model):
+    """Extension Media resource.
+
+    Attributes:
+        name: Media name.
+        revision: Revision number.
+        type: Media type (Video or Image).
+        description: Media description.
+        status: Media status (Draft, Published, Unpublished, Deleted).
+        filename: Original file name.
+        size: File size in bytes.
+        content_type: MIME content type.
+        display_order: Display order.
+        url: URL to access the media.
+        extension: Reference to the extension.
+        audit: Audit information (created, updated, published, unpublished).
+    """
+
+    name: str | None
+    revision: int | None
+    type: str | None
+    description: str | None
+    status: str | None
+    filename: str | None
+    size: int | None
+    content_type: str | None
+    display_order: int | None
+    url: str | None
+    extension: BaseModel | None
+    audit: BaseModel | None
+
+
+class ExtensionMediaServiceConfig:
+    """Extension Media service configuration."""
+
+    _endpoint = "/public/v1/integration/extensions/{extension_id}/media"
+    _model_class = ExtensionMedia
+    _collection_key = "data"
+    _upload_file_key = "file"
+    _upload_data_key = "media"
+
+
+class ExtensionMediaService(
+    MediaMixin[ExtensionMedia],
+    PublishableMixin[ExtensionMedia],
+    DownloadFileMixin[ExtensionMedia],
+    CreateFileMixin[ExtensionMedia],
+    ModifiableResourceMixin[ExtensionMedia],
+    DeleteMixin,
+    CollectionMixin[ExtensionMedia],
+    Service[ExtensionMedia],
+    ExtensionMediaServiceConfig,
+):
+    """Sync service for the /public/v1/integration/extensions/{extensionId}/media endpoint."""
+
+
+class AsyncExtensionMediaService(
+    AsyncMediaMixin[ExtensionMedia],
+    AsyncPublishableMixin[ExtensionMedia],
+    AsyncDownloadFileMixin[ExtensionMedia],
+    AsyncCreateFileMixin[ExtensionMedia],
+    AsyncModifiableResourceMixin[ExtensionMedia],
+    AsyncDeleteMixin,
+    AsyncCollectionMixin[ExtensionMedia],
+    AsyncService[ExtensionMedia],
+    ExtensionMediaServiceConfig,
+):
+    """Async service for the /public/v1/integration/extensions/{extensionId}/media endpoint."""

mpt_api_client/resources/integration/extensions.py

@@ -13,6 +13,10 @@
 )
 from mpt_api_client.models import Model
 from mpt_api_client.models.model import BaseModel
+from mpt_api_client.resources.integration.extension_media import (
+    AsyncExtensionMediaService,
+    ExtensionMediaService,
+)
 from mpt_api_client.resources.integration.extension_terms import (
     AsyncExtensionTermsService,
     ExtensionTermsService,
@@ -89,6 +93,19 @@ def terms(self, extension_id: str) -> ExtensionTermsService:
             http_client=self.http_client, endpoint_params={"extension_id": extension_id}
         )
 
+    def media(self, extension_id: str) -> ExtensionMediaService:
+        """Return the media service for the given extension.
+
+        Args:
+            extension_id: Extension ID.
+
+        Returns:
+            ExtensionMediaService instance.
+        """
+        return ExtensionMediaService(
+            http_client=self.http_client, endpoint_params={"extension_id": extension_id}
+        )
+
 
 class AsyncExtensionsService(
     AsyncExtensionMixin[Extension],
@@ -107,3 +124,16 @@ def terms(self, extension_id: str) -> AsyncExtensionTermsService:
         return AsyncExtensionTermsService(
             http_client=self.http_client, endpoint_params={"extension_id": extension_id}
         )
+
+    def media(self, extension_id: str) -> AsyncExtensionMediaService:
+        """Return the async media service for the given extension.
+
+        Args:
+            extension_id: Extension ID.
+
+        Returns:
+            AsyncExtensionMediaService instance.
+        """
+        return AsyncExtensionMediaService(
+            http_client=self.http_client, endpoint_params={"extension_id": extension_id}
+        )

mpt_api_client/resources/integration/mixins/__init__.py

@@ -2,14 +2,20 @@
     AsyncExtensionMixin,
     ExtensionMixin,
 )
+from mpt_api_client.resources.integration.mixins.media_mixin import (
+    AsyncMediaMixin,
+    MediaMixin,
+)
 from mpt_api_client.resources.integration.mixins.publishable_mixin import (
     AsyncPublishableMixin,
     PublishableMixin,
 )
 
 __all__ = [  # noqa: WPS410
     "AsyncExtensionMixin",
+    "AsyncMediaMixin",
     "AsyncPublishableMixin",
     "ExtensionMixin",
+    "MediaMixin",
     "PublishableMixin",
 ]

mpt_api_client/resources/integration/mixins/media_mixin.py

@@ -0,0 +1,46 @@
+from mpt_api_client.http.types import FileTypes
+from mpt_api_client.http.url_utils import join_url_path
+
+
+class MediaMixin[Model]:
+    """Mixin that adds media-specific actions: upload_image."""
+
+    def upload_image(self, resource_id: str, file: FileTypes) -> Model:  # noqa: WPS110
+        """Upload or replace the image binary for this media item.
+
+        Args:
+            resource_id: Media item ID.
+            file: Binary image file to upload.
+
+        Returns:
+            Updated media item.
+        """
+        url = join_url_path(self.path, resource_id, "image")  # type: ignore[attr-defined]
+        response = self.http_client.request(  # type: ignore[attr-defined]
+            "put",
+            url,
+            files={"file": file},
+        )
+        return self._model_class.from_response(response)  # type: ignore[attr-defined, no-any-return]
+
+
+class AsyncMediaMixin[Model]:
+    """Async mixin that adds media-specific actions: upload_image."""
+
+    async def upload_image(self, resource_id: str, file: FileTypes) -> Model:  # noqa: WPS110
+        """Upload or replace the image binary for this media item.
+
+        Args:
+            resource_id: Media item ID.
+            file: Binary image file to upload.
+
+        Returns:
+            Updated media item.
+        """
+        url = join_url_path(self.path, resource_id, "image")  # type: ignore[attr-defined]
+        response = await self.http_client.request(  # type: ignore[attr-defined]
+            "put",
+            url,
+            files={"file": file},
+        )
+        return self._model_class.from_response(response)  # type: ignore[attr-defined, no-any-return]

tests/e2e/integration/extension_media/conftest.py

@@ -0,0 +1,53 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+
+
+@pytest.fixture(scope="session")
+def extension_id(e2e_config):
+    return e2e_config["integration.extension.id"]
+
+
+@pytest.fixture
+def extension_media_service(mpt_vendor, extension_id):
+    return mpt_vendor.integration.extensions.media(extension_id)
+
+
+@pytest.fixture
+def async_extension_media_service(async_mpt_vendor, extension_id):
+    return async_mpt_vendor.integration.extensions.media(extension_id)
+
+
+@pytest.fixture
+def media_data(short_uuid):
+    return {
+        "name": f"e2e - please delete {short_uuid}",
+        "description": "Created by automated E2E tests. Safe to delete.",
+        "mediaType": "Image",
+        "url": "https://example.com/image.png",
+        "displayOrder": 1,
+    }
+
+
+@pytest.fixture
+def created_media(extension_media_service, media_data, logo_fd):
+    media = extension_media_service.create(media_data, file=logo_fd)
+
+    yield media
+
+    try:
+        extension_media_service.delete(media.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete media {media.id}: {error.title}")  # noqa: WPS421
+
+
+@pytest.fixture
+async def async_created_media(async_extension_media_service, media_data, logo_fd):
+    media = await async_extension_media_service.create(media_data, file=logo_fd)
+
+    yield media
+
+    try:
+        await async_extension_media_service.delete(media.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete media {media.id}: {error.title}")  # noqa: WPS421

tests/e2e/integration/extension_media/test_async_extension_media.py

@@ -0,0 +1,53 @@
+import pytest
+
+from tests.e2e.helper import assert_async_service_filter_with_iterate
+
+pytestmark = [
+    pytest.mark.flaky,
+]
+
+
+def test_create_extension_media_async(async_created_media, media_data):
+    result = async_created_media.name
+
+    assert result == media_data["name"]
+
+
+async def test_filter_extension_media_async(async_extension_media_service, async_created_media):
+    await assert_async_service_filter_with_iterate(
+        async_extension_media_service, async_created_media.id, None
+    )  # act
+
+
+async def test_update_extension_media_async(
+    async_extension_media_service, async_created_media, short_uuid
+):
+    update_data = {"name": f"e2e updated {short_uuid}"}
+
+    result = await async_extension_media_service.update(async_created_media.id, update_data)
+
+    assert result.name == update_data["name"]
+
+
+async def test_publish_extension_media_async(async_extension_media_service, async_created_media):
+    result = await async_extension_media_service.publish(async_created_media.id)
+
+    assert result.status == "Published"
+
+
+async def test_download_extension_media_async(async_extension_media_service, async_created_media):
+    result = await async_extension_media_service.download(async_created_media.id)
+
+    assert result.file_contents is not None
+
+
+async def test_unpublish_extension_media_async(async_extension_media_service, async_created_media):
+    await async_extension_media_service.publish(async_created_media.id)
+
+    result = await async_extension_media_service.unpublish(async_created_media.id)
+
+    assert result.status == "Unpublished"
+
+
+async def test_delete_extension_media_async(async_extension_media_service, async_created_media):
+    await async_extension_media_service.delete(async_created_media.id)  # act

tests/e2e/integration/extension_media/test_sync_extension_media.py

@@ -0,0 +1,49 @@
+import pytest
+
+from tests.e2e.helper import assert_service_filter_with_iterate
+
+pytestmark = [
+    pytest.mark.flaky,
+]
+
+
+def test_create_extension_media(created_media, media_data):
+    result = created_media.name
+
+    assert result == media_data["name"]
+
+
+def test_filter_extension_media(extension_media_service, created_media):
+    assert_service_filter_with_iterate(extension_media_service, created_media.id, None)  # act
+
+
+def test_update_extension_media(extension_media_service, created_media, short_uuid):
+    update_data = {"name": f"e2e updated {short_uuid}"}
+
+    result = extension_media_service.update(created_media.id, update_data)
+
+    assert result.name == update_data["name"]
+
+
+def test_publish_extension_media(extension_media_service, created_media):
+    result = extension_media_service.publish(created_media.id)
+
+    assert result.status == "Published"
+
+
+def test_download_extension_media(extension_media_service, created_media):
+    result = extension_media_service.download(created_media.id)
+
+    assert result.file_contents is not None
+
+
+def test_unpublish_extension_media(extension_media_service, created_media):
+    extension_media_service.publish(created_media.id)
+
+    result = extension_media_service.unpublish(created_media.id)
+
+    assert result.status == "Unpublished"
+
+
+def test_delete_extension_media(extension_media_service, created_media):
+    extension_media_service.delete(created_media.id)  # act