MPT-20438: Added endpoints and e2e tests for program enrollment attachments

e2e_config.test.json

@@ -72,6 +72,7 @@
   "integration.term.id": "ETC-6587-4477-0062",
   "program.document.file.id": "PDM-9643-3741-0001",
   "program.enrollment.assignee.id": "USR-6337-1324",
+  "program.enrollment.attachment.id": "ENA-3965-5056-7966-0001",
   "program.enrollment.id": "ENR-3965-5056-7966",
   "program.enrollment.process.template.id": "PTM-9643-3741-0001",
   "program.enrollment.query.template.id": "PTM-9643-3741-0002",

mpt_api_client/resources/program/enrollments.py

@@ -7,6 +7,10 @@
 )
 from mpt_api_client.models import Model
 from mpt_api_client.models.model import BaseModel, ResourceData
+from mpt_api_client.resources.program.enrollments_attachments import (
+    AsyncEnrollmentAttachmentsService,
+    EnrollmentAttachmentsService,
+)
 from mpt_api_client.resources.program.mixins.render_mixin import AsyncRenderMixin, RenderMixin
 
 
@@ -59,6 +63,19 @@ class EnrollmentService(
 ):
     """Program enrollment service."""
 
+    def attachments(self, enrollment_id: str) -> EnrollmentAttachmentsService:
+        """Get enrollment attachments service.
+
+        Args:
+            enrollment_id: Enrollment ID
+
+        Returns:
+            Enrollment attachments service.
+        """
+        return EnrollmentAttachmentsService(
+            http_client=self.http_client, endpoint_params={"enrollment_id": enrollment_id}
+        )
+
     def validate(self, resource_id: str, resource_data: ResourceData | None = None) -> Enrollment:
         """Validate enrollment.
 
@@ -141,6 +158,19 @@ class AsyncEnrollmentService(
 ):
     """Async program enrollment service."""
 
+    def attachments(self, enrollment_id: str) -> AsyncEnrollmentAttachmentsService:
+        """Get enrollment attachments service.
+
+        Args:
+            enrollment_id: Enrollment ID
+
+        Returns:
+            Enrollment attachments service.
+        """
+        return AsyncEnrollmentAttachmentsService(
+            http_client=self.http_client, endpoint_params={"enrollment_id": enrollment_id}
+        )
+
     async def validate(
         self, resource_id: str, resource_data: ResourceData | None = None
     ) -> Enrollment:

mpt_api_client/resources/program/enrollments_attachments.py

@@ -0,0 +1,63 @@
+from mpt_api_client.http import AsyncService, Service
+from mpt_api_client.http.mixins import (
+    AsyncCollectionMixin,
+    CollectionMixin,
+)
+from mpt_api_client.models import Model
+from mpt_api_client.models.model import BaseModel
+from mpt_api_client.resources.program.mixins.attachment_mixin import (
+    AsyncAttachmentMixin,
+    AttachmentMixin,
+)
+
+
+class EnrollmentAttachment(Model):
+    """Enrollment Attachment resource.
+
+    Attributes:
+        name: The name of the attachment.
+        description: The description of the attachment.
+        type: The type of the attachment.
+        filename: The filename of the attachment.
+        size: The size of the attachment in bytes.
+        content_type: The content type of the attachment.
+        enrollment: The enrollment associated with the attachment.
+        audit: The audit information for the attachment.
+    """
+
+    name: str | None = None
+    description: str | None = None
+    type: str | None = None
+    filename: str | None = None
+    size: int | None = None
+    content_type: str | None = None
+    enrollment: BaseModel | None = None
+    audit: BaseModel | None = None
+
+
+class EnrollmentAttachmentsServiceConfig:
+    """Enrollment Attachments service configuration."""
+
+    _endpoint = "/public/v1/program/enrollments/{enrollment_id}/attachments"
+    _model_class = EnrollmentAttachment
+    _collection_key = "data"
+    _upload_file_key = "file"
+    _upload_data_key = "attachment"
+
+
+class EnrollmentAttachmentsService(
+    AttachmentMixin[EnrollmentAttachment],
+    CollectionMixin[EnrollmentAttachment],
+    Service[EnrollmentAttachment],
+    EnrollmentAttachmentsServiceConfig,
+):
+    """Enrollment Attachments service."""
+
+
+class AsyncEnrollmentAttachmentsService(
+    AsyncAttachmentMixin[EnrollmentAttachment],
+    AsyncCollectionMixin[EnrollmentAttachment],
+    AsyncService[EnrollmentAttachment],
+    EnrollmentAttachmentsServiceConfig,
+):
+    """Enrollment Attachments service."""

mpt_api_client/resources/program/mixins/attachment_mixin.py

@@ -0,0 +1,32 @@
+from mpt_api_client.http.mixins import (
+    AsyncCreateFileMixin,
+    AsyncDeleteMixin,
+    AsyncDownloadFileMixin,
+    AsyncGetMixin,
+    AsyncUpdateMixin,
+    CreateFileMixin,
+    DeleteMixin,
+    DownloadFileMixin,
+    GetMixin,
+    UpdateMixin,
+)
+
+
+class AttachmentMixin[Model](
+    CreateFileMixin[Model],
+    UpdateMixin[Model],
+    DeleteMixin,
+    DownloadFileMixin[Model],
+    GetMixin[Model],
+):
+    """Attachment mixin."""
+
+
+class AsyncAttachmentMixin[Model](
+    AsyncCreateFileMixin[Model],
+    AsyncUpdateMixin[Model],
+    AsyncDeleteMixin,
+    AsyncDownloadFileMixin[Model],
+    AsyncGetMixin[Model],
+):
+    """Async Attachment mixin."""

pyproject.toml

@@ -126,7 +126,7 @@ per-file-ignores = [
   "mpt_api_client/resources/exchange/*.py: WPS235 WPS215",
   "mpt_api_client/resources/integration/*.py: WPS214 WPS215 WPS235",
   "mpt_api_client/resources/helpdesk/*.py: WPS204 WPS215 WPS214",
-  "mpt_api_client/resources/program/*.py: WPS204 WPS215",
+  "mpt_api_client/resources/program/*.py: WPS204 WPS215 WPS235",
   "mpt_api_client/rql/query_builder.py: WPS110 WPS115 WPS210 WPS214",
   "tests/e2e/accounts/*.py: WPS430 WPS202",
   "tests/e2e/billing/*.py: WPS202 WPS421 WPS118",

tests/e2e/program/enrollment/attachment/conftest.py

@@ -0,0 +1,22 @@
+import pytest
+
+
+@pytest.fixture
+def attachment_id(e2e_config):
+    return e2e_config["program.enrollment.attachment.id"]
+
+
+@pytest.fixture
+def invalid_attachment_id():
+    return "ENA-0000-0000-0000-0000"
+
+
+@pytest.fixture
+def enrollment_attachment_factory():
+    def factory(name: str = "E2E Created Program Enrollment Attachment"):
+        return {
+            "name": name,
+            "description": name,
+        }
+
+    return factory

tests/e2e/program/enrollment/attachment/test_async_attachment.py

@@ -0,0 +1,94 @@
+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_enrollment_attachment(
+    async_mpt_vendor, enrollment_attachment_factory, enrollment_id, pdf_fd
+):
+    new_enrollment_attachment_request_data = enrollment_attachment_factory(
+        name="E2E Created Program Enrollment Attachment",
+    )
+    enrollment_attachments = async_mpt_vendor.program.enrollments.attachments(enrollment_id)
+
+    created_enrollment_attachment = await enrollment_attachments.create(
+        new_enrollment_attachment_request_data, file=pdf_fd
+    )
+
+    yield created_enrollment_attachment
+
+    try:
+        await enrollment_attachments.delete(created_enrollment_attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete enrollment attachment: {error.title}")  # noqa: WPS421
+
+
+@pytest.fixture
+def enrollment_attachments(async_mpt_vendor, enrollment_id):
+    return async_mpt_vendor.program.enrollments.attachments(enrollment_id)
+
+
+async def test_get_enrollment_attachment_by_id(enrollment_attachments, attachment_id):
+    result = await enrollment_attachments.get(attachment_id)
+
+    assert result is not None
+
+
+async def test_get_enrollment_attachment_not_found(enrollment_attachments, invalid_attachment_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await enrollment_attachments.get(invalid_attachment_id)
+
+
+async def test_list_enrollment_attachments(enrollment_attachments):
+    limit = 10
+
+    result = await enrollment_attachments.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_filter_enrollment_attachments(enrollment_attachments, attachment_id):
+    select_fields = ["-description"]
+    filtered_attachments = (
+        enrollment_attachments
+        .filter(RQLQuery(id=attachment_id))
+        .filter(RQLQuery(name="E2E Seeded Program Enrollment Attachment"))
+        .select(*select_fields)
+    )
+
+    result = [attachment async for attachment in filtered_attachments.iterate()]
+
+    assert len(result) == 1
+
+
+def test_create_enrollment_attachment(created_enrollment_attachment):
+    result = created_enrollment_attachment
+
+    assert result is not None
+
+
+async def test_update_enrollment_attachment(enrollment_attachments, created_enrollment_attachment):
+    updated_data = {
+        "name": "E2E Updated Program Enrollment Attachment",
+        "description": "E2E Updated Program Enrollment Attachment",
+    }
+
+    result = await enrollment_attachments.update(created_enrollment_attachment.id, updated_data)
+
+    assert result is not None
+
+
+async def test_delete_enrollment_attachment(enrollment_attachments, created_enrollment_attachment):
+    result = created_enrollment_attachment
+
+    await enrollment_attachments.delete(result.id)
+
+
+async def test_download_enrollment_attachment(enrollment_attachments, attachment_id):
+    result = await enrollment_attachments.download(attachment_id)
+
+    assert result.file_contents is not None

tests/e2e/program/enrollment/attachment/test_sync_attachment.py

@@ -0,0 +1,92 @@
+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_enrollment_attachment(mpt_vendor, enrollment_attachment_factory, enrollment_id, pdf_fd):
+    new_enrollment_attachment_request_data = enrollment_attachment_factory(
+        name="E2E Created Program Enrollment Attachment",
+    )
+    enrollment_attachments = mpt_vendor.program.enrollments.attachments(enrollment_id)
+
+    created_enrollment_attachment = enrollment_attachments.create(
+        new_enrollment_attachment_request_data, file=pdf_fd
+    )
+
+    yield created_enrollment_attachment
+
+    try:
+        enrollment_attachments.delete(created_enrollment_attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete enrollment attachment: {error.title}")  # noqa: WPS421
+
+
+@pytest.fixture
+def enrollment_attachments(mpt_vendor, enrollment_id):
+    return mpt_vendor.program.enrollments.attachments(enrollment_id)
+
+
+def test_get_enrollment_attachment_by_id(enrollment_attachments, attachment_id):
+    result = enrollment_attachments.get(attachment_id)
+
+    assert result is not None
+
+
+def test_get_enrollment_attachment_not_found(enrollment_attachments, invalid_attachment_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        enrollment_attachments.get(invalid_attachment_id)
+
+
+def test_list_enrollment_attachments(enrollment_attachments):
+    limit = 10
+
+    result = enrollment_attachments.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+def test_filter_enrollment_attachments(enrollment_attachments, attachment_id):
+    select_fields = ["-description"]
+    filtered_attachments = (
+        enrollment_attachments
+        .filter(RQLQuery(id=attachment_id))
+        .filter(RQLQuery(name="E2E Seeded Program Enrollment Attachment"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_attachments.iterate())
+
+    assert len(result) == 1
+
+
+def test_create_enrollment_attachment(created_enrollment_attachment):
+    result = created_enrollment_attachment
+
+    assert result is not None
+
+
+def test_update_enrollment_attachment(enrollment_attachments, created_enrollment_attachment):
+    updated_data = {
+        "name": "E2E Updated Program Enrollment Attachment",
+        "description": "E2E Updated Program Enrollment Attachment",
+    }
+
+    result = enrollment_attachments.update(created_enrollment_attachment.id, updated_data)
+
+    assert result is not None
+
+
+def test_delete_enrollment_attachment(enrollment_attachments, created_enrollment_attachment):
+    result = created_enrollment_attachment
+
+    enrollment_attachments.delete(result.id)
+
+
+def test_download_enrollment_attachment(enrollment_attachments, attachment_id):
+    result = enrollment_attachments.download(attachment_id)
+
+    assert result.file_contents is not None