MPT-20330: Added endpoints and e2e tests for program terms variants

e2e_config.test.json

@@ -76,5 +76,6 @@
   "program.parameter.id": "PPM-9643-3741-0001",
   "program.program.id": "PRG-9643-3741",
   "program.template.id": "PTM-9643-3741-0004",
-  "program.terms.id": "PTC-9643-3741-0001"
+  "program.terms.id": "PTC-9643-3741-0001",
+  "program.terms.variant.id": "PTV-9643-3741-0001-0001"
 }

mpt_api_client/resources/program/programs_terms.py

@@ -11,18 +11,22 @@
     AsyncPublishableMixin,
     PublishableMixin,
 )
+from mpt_api_client.resources.program.programs_terms_variant import (
+    AsyncTermVariantService,
+    TermVariantService,
+)
 
 
 class Term(Model):
     """Program term resource.
 
     Attributes:
-         name: Program term name.
-         description: Program term description.
-         display_order: Display order of the program term.
-         status: Program term status.
-         program: Reference to the program.
-         audit: Audit information (created, updated events).
+        name: Program term name.
+        description: Program term description.
+        display_order: Display order of the program term.
+        status: Program term status.
+        program: Reference to the program.
+        audit: Audit information (created, updated events).
     """
 
     name: str | None
@@ -50,6 +54,13 @@ class TermService(
 ):
     """Program term service."""
 
+    def variants(self, term_id: str) -> TermVariantService:
+        """Access program term variants service."""
+        return TermVariantService(
+            http_client=self.http_client,
+            endpoint_params={"program_id": self.endpoint_params["program_id"], "term_id": term_id},
+        )
+
 
 class AsyncTermService(
     AsyncPublishableMixin[Term],
@@ -59,3 +70,10 @@ class AsyncTermService(
     TermServiceConfig,
 ):
     """Async program term service."""
+
+    def variants(self, term_id: str) -> AsyncTermVariantService:
+        """Access async program term variants service."""
+        return AsyncTermVariantService(
+            http_client=self.http_client,
+            endpoint_params={"program_id": self.endpoint_params["program_id"], "term_id": term_id},
+        )

mpt_api_client/resources/program/programs_terms_variant.py

@@ -0,0 +1,83 @@
+from mpt_api_client.http import AsyncService, Service
+from mpt_api_client.http.mixins import (
+    AsyncCollectionMixin,
+    AsyncCreateFileMixin,
+    AsyncDownloadFileMixin,
+    AsyncModifiableResourceMixin,
+    CollectionMixin,
+    CreateFileMixin,
+    DownloadFileMixin,
+    ModifiableResourceMixin,
+)
+from mpt_api_client.models import Model
+from mpt_api_client.models.model import BaseModel
+from mpt_api_client.resources.program.mixins import (
+    AsyncPublishableMixin,
+    PublishableMixin,
+)
+
+
+class TermVariant(Model):
+    """Term variant resource.
+
+    Attributes:
+        name: The name of the term variant.
+        type: The type of the term variant.
+        asset_url: The URL of the asset.
+        language_code: The language code of the term variant.
+        description: The description of the term variant.
+        status: The status of the term variant.
+        filename: The filename of the term variant.
+        size: The size of the term variant.
+        content_type: The content type of the term variant.
+        program_terms_and_conditions: The associated program terms and conditions.
+        file_id: The ID of the file.
+        audit: The audit information.
+    """
+
+    name: str | None
+    type: str | None
+    asset_url: str | None
+    language_code: str | None
+    description: str | None
+    status: str | None
+    filename: str | None
+    size: int | None
+    content_type: str | None
+    program_terms_and_conditions: BaseModel | None
+    file_id: str | None
+    audit: BaseModel | None
+
+
+class TermVariantServiceConfig:
+    """Term variant service configuration."""
+
+    _endpoint = "/public/v1/program/programs/{program_id}/terms/{term_id}/variants"
+    _model_class = TermVariant
+    _collection_key = "data"
+    _upload_file_key = "file"
+    _upload_data_key = "variant"
+
+
+class TermVariantService(
+    CreateFileMixin[TermVariant],
+    DownloadFileMixin[TermVariant],
+    ModifiableResourceMixin[TermVariant],
+    PublishableMixin[TermVariant],
+    CollectionMixin[TermVariant],
+    Service[TermVariant],
+    TermVariantServiceConfig,
+):
+    """Term variant service."""
+
+
+class AsyncTermVariantService(
+    AsyncCreateFileMixin[TermVariant],
+    AsyncDownloadFileMixin[TermVariant],
+    AsyncModifiableResourceMixin[TermVariant],
+    AsyncPublishableMixin[TermVariant],
+    AsyncCollectionMixin[TermVariant],
+    AsyncService[TermVariant],
+    TermVariantServiceConfig,
+):
+    """Async term variant service."""

tests/e2e/program/program/term/variant/conftest.py

@@ -0,0 +1,25 @@
+import pytest
+
+
+@pytest.fixture
+def variant_id(e2e_config):
+    return e2e_config["program.terms.variant.id"]
+
+
+@pytest.fixture
+def invalid_variant_id():
+    return "PTV-0000-0000-0000-0000"
+
+
+@pytest.fixture
+def variant_data_factory():
+    def factory(variant_type: str = "File", asset_url: str = "") -> dict:
+        return {
+            "name": "E2E Created Program Term Variant",
+            "description": "E2E Created Program Term Variant",
+            "languageCode": "en-us",
+            "type": variant_type,
+            "assetUrl": asset_url,
+        }
+
+    return factory

tests/e2e/program/program/term/variant/test_async_variant.py

@@ -0,0 +1,106 @@
+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 vendor_variant_service(async_mpt_vendor, program_id, term_id):
+    return async_mpt_vendor.program.programs.terms(program_id).variants(term_id)
+
+
+@pytest.fixture
+async def created_variant(vendor_variant_service, variant_data_factory, pdf_fd):
+    variant_data = variant_data_factory()
+    variant = await vendor_variant_service.create(variant_data, pdf_fd)
+    yield variant
+    try:
+        await vendor_variant_service.delete(variant.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete variant {variant.id}: {error.title}")  # noqa: WPS421
+
+
+@pytest.fixture
+async def created_variant_from_url(vendor_variant_service, variant_data_factory, pdf_url):
+    variant_data = variant_data_factory(variant_type="Online", asset_url=pdf_url)
+    variant = await vendor_variant_service.create(variant_data)
+    yield variant
+    try:
+        await vendor_variant_service.delete(variant.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete variant {variant.id}: {error.title}")  # noqa: WPS421
+
+
+def test_create_variant(created_variant, variant_data_factory):
+    variant_data = variant_data_factory()
+
+    result = created_variant.name == variant_data["name"]
+
+    assert result is True
+
+
+def test_create_variant_from_url(created_variant_from_url, variant_data_factory):
+    variant_data = variant_data_factory(
+        variant_type="Online", asset_url="https://example.com/file.pdf"
+    )
+
+    result = created_variant_from_url.name == variant_data["name"]
+
+    assert result is True
+
+
+async def test_update_variant(vendor_variant_service, created_variant):
+    update_data = {"name": "E2E Updated Program Term Variant"}
+
+    result = await vendor_variant_service.update(created_variant.id, update_data)
+
+    assert result.name == "E2E Updated Program Term Variant"
+
+
+async def test_delete_variant(vendor_variant_service, created_variant):
+    variant_data = created_variant
+
+    result = await vendor_variant_service.delete(variant_data.id)
+
+    assert result is None
+
+
+async def test_get_variant(vendor_variant_service, variant_id):
+    result = await vendor_variant_service.get(variant_id)
+
+    assert result.id == variant_id
+
+
+async def test_get_invalid_variant(vendor_variant_service, invalid_variant_id):
+    with pytest.raises(MPTAPIError):
+        await vendor_variant_service.get(invalid_variant_id)
+
+
+async def test_filter_and_select_variants(vendor_variant_service, variant_id):
+    select_fields = ["-description", "-audit"]
+    filtered_variants = (
+        vendor_variant_service
+        .filter(RQLQuery(id=variant_id))
+        .filter(RQLQuery(name="E2E Seeded Program Terms Variant"))
+        .select(*select_fields)
+    )
+
+    result = [variant async for variant in filtered_variants.iterate()]
+
+    assert len(result) == 1
+
+
+async def test_publish_variant(vendor_variant_service, created_variant):
+    result = await vendor_variant_service.publish(created_variant.id)
+
+    assert result.status == "Published"
+
+
+async def test_unpublish_variant(vendor_variant_service, created_variant):
+    await vendor_variant_service.publish(created_variant.id)
+
+    result = await vendor_variant_service.unpublish(created_variant.id)
+
+    assert result.status == "Unpublished"

tests/e2e/program/program/term/variant/test_sync_variant.py

@@ -0,0 +1,106 @@
+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 vendor_variant_service(mpt_vendor, program_id, term_id):
+    return mpt_vendor.program.programs.terms(program_id).variants(term_id)
+
+
+@pytest.fixture
+def created_variant(vendor_variant_service, variant_data_factory, pdf_fd):
+    variant_data = variant_data_factory()
+    variant = vendor_variant_service.create(variant_data, pdf_fd)
+    yield variant
+    try:
+        vendor_variant_service.delete(variant.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete variant {variant.id}: {error.title}")  # noqa: WPS421
+
+
+@pytest.fixture
+def created_variant_from_url(vendor_variant_service, variant_data_factory, pdf_url):
+    variant_data = variant_data_factory(variant_type="Online", asset_url=pdf_url)
+    variant = vendor_variant_service.create(variant_data)
+    yield variant
+    try:
+        vendor_variant_service.delete(variant.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete variant {variant.id}: {error.title}")  # noqa: WPS421
+
+
+def test_create_variant(created_variant, variant_data_factory):
+    variant_data = variant_data_factory()
+
+    result = created_variant.name == variant_data["name"]
+
+    assert result is True
+
+
+def test_create_variant_from_url(created_variant_from_url, variant_data_factory):
+    variant_data = variant_data_factory(
+        variant_type="Online", asset_url="https://example.com/file.pdf"
+    )
+
+    result = created_variant_from_url.name == variant_data["name"]
+
+    assert result is True
+
+
+def test_update_variant(vendor_variant_service, created_variant):
+    update_data = {"name": "E2E Updated Program Term Variant"}
+
+    result = vendor_variant_service.update(created_variant.id, update_data)
+
+    assert result.name == "E2E Updated Program Term Variant"
+
+
+def test_delete_variant(vendor_variant_service, created_variant):
+    variant_data = created_variant
+
+    result = vendor_variant_service.delete(variant_data.id)
+
+    assert result is None
+
+
+def test_get_variant(vendor_variant_service, variant_id):
+    result = vendor_variant_service.get(variant_id)
+
+    assert result.id == variant_id
+
+
+def test_get_invalid_variant(vendor_variant_service, invalid_variant_id):
+    with pytest.raises(MPTAPIError):
+        vendor_variant_service.get(invalid_variant_id)
+
+
+def test_filter_and_select_variants(vendor_variant_service, variant_id):
+    select_fields = ["-description", "-audit"]
+    filtered_variants = (
+        vendor_variant_service
+        .filter(RQLQuery(id=variant_id))
+        .filter(RQLQuery(name="E2E Seeded Program Terms Variant"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_variants.iterate())
+
+    assert len(result) == 1
+
+
+def test_publish_variant(vendor_variant_service, created_variant):
+    result = vendor_variant_service.publish(created_variant.id)
+
+    assert result.status == "Published"
+
+
+def test_unpublish_variant(vendor_variant_service, created_variant):
+    vendor_variant_service.publish(created_variant.id)
+
+    result = vendor_variant_service.unpublish(created_variant.id)
+
+    assert result.status == "Unpublished"