MPT-20436: Added endpoints and e2e tests for program certificates

e2e_config.test.json

@@ -70,6 +70,7 @@
   "notifications.subscriber.id": "NTS-0829-7123-7123",
   "integration.extension.id": "EXT-6587-4477",
   "integration.term.id": "ETC-6587-4477-0062",
+  "program.certificate.id": "CER-9646-2171-8417",
   "program.document.file.id": "PDM-9643-3741-0001",
   "program.enrollment.assignee.id": "USR-6337-1324",
   "program.enrollment.id": "ENR-3965-5056-7966",

mpt_api_client/resources/program/certificates.py

@@ -0,0 +1,93 @@
+from mpt_api_client.http import AsyncService, Service
+from mpt_api_client.http.mixins import (
+    AsyncCollectionMixin,
+    AsyncManagedResourceMixin,
+    CollectionMixin,
+    ManagedResourceMixin,
+)
+from mpt_api_client.models import Model
+from mpt_api_client.models.model import BaseModel, ResourceData
+
+
+class Certificate(Model):
+    """Program certificate resource.
+
+    Attributes:
+        name: Certificate name.
+        program: Reference to the program.
+        vendor: Reference to the vendor.
+        external_ids: External identifiers.
+        client: Reference to the client.
+        applicable_to: Applicable to which entities.
+        licensee: Reference to the licensee.
+        eligibility: Eligibility criteria.
+        status: Certificate status.
+        status_notes: Additional notes on the certificate status.
+        parameters: Certificate parameters.
+        audit: Audit information.
+    """
+
+    name: str | None
+    program: BaseModel | None
+    vendor: BaseModel | None
+    external_ids: BaseModel | None
+    client: BaseModel | None
+    applicable_to: str | None
+    licensee: BaseModel | None
+    eligibility: BaseModel | None
+    status: str | None
+    status_notes: str | None
+    parameters: BaseModel | None  # noqa: WPS110
+    audit: BaseModel | None
+
+
+class CertificateServiceConfig:
+    """Program certificate service config."""
+
+    _endpoint = "/public/v1/program/certificates"
+    _model_class = Certificate
+    _collection_key = "data"
+
+
+class CertificateService(
+    ManagedResourceMixin[Certificate],
+    CollectionMixin[Certificate],
+    Service[Certificate],
+    CertificateServiceConfig,
+):
+    """Program certificate service."""
+
+    def terminate(self, resource_id: str, resource_data: ResourceData | None = None) -> Certificate:
+        """Terminate a certificate.
+
+        Args:
+            resource_id: Certificate ID.
+            resource_data: Additional data for termination.
+
+        Returns:
+            Terminated certificate.
+        """
+        return self._resource(resource_id).post("/terminate", json=resource_data)
+
+
+class AsyncCertificateService(
+    AsyncManagedResourceMixin[Certificate],
+    AsyncCollectionMixin[Certificate],
+    AsyncService[Certificate],
+    CertificateServiceConfig,
+):
+    """Asynchronous program certificate service."""
+
+    async def terminate(
+        self, resource_id: str, resource_data: ResourceData | None = None
+    ) -> Certificate:
+        """Asynchronously terminate a certificate.
+
+        Args:
+            resource_id: Certificate ID.
+            resource_data: Additional data for termination.
+
+        Returns:
+            Terminated certificate.
+        """
+        return await self._resource(resource_id).post("/terminate", json=resource_data)

mpt_api_client/resources/program/program.py

@@ -1,4 +1,8 @@
 from mpt_api_client.http import AsyncHTTPClient, HTTPClient
+from mpt_api_client.resources.program.certificates import (
+    AsyncCertificateService,
+    CertificateService,
+)
 from mpt_api_client.resources.program.enrollments import AsyncEnrollmentService, EnrollmentService
 from mpt_api_client.resources.program.programs import AsyncProgramsService, ProgramsService
 
@@ -19,6 +23,11 @@ def enrollments(self) -> EnrollmentService:
         """Enrollments service."""
         return EnrollmentService(http_client=self.http_client)
 
+    @property
+    def certificates(self) -> CertificateService:
+        """Certificates service."""
+        return CertificateService(http_client=self.http_client)
+
 
 class AsyncProgram:
     """Program MPT API Module."""
@@ -35,3 +44,8 @@ def programs(self) -> AsyncProgramsService:
     def enrollments(self) -> AsyncEnrollmentService:
         """Enrollments service."""
         return AsyncEnrollmentService(http_client=self.http_client)
+
+    @property
+    def certificates(self) -> AsyncCertificateService:
+        """Certificates service."""
+        return AsyncCertificateService(http_client=self.http_client)

tests/e2e/program/certificate/conftest.py

@@ -0,0 +1,36 @@
+import pytest
+
+
+@pytest.fixture
+def certificate_id(e2e_config):
+    return e2e_config["program.certificate.id"]
+
+
+@pytest.fixture
+def invalid_certificate_id():
+    return "CER-0000-0000-0000"
+
+
+@pytest.fixture
+def certificate_data(program_id, licensee_id, buyer_account_id):
+    return {
+        "name": "E2E Created Program Certificate",
+        "program": {"id": program_id},
+        "licensee": {"id": licensee_id},
+        "client": {"id": buyer_account_id},
+        "parameters": {"ordering": [], "fulfillment": []},
+    }
+
+
+@pytest.fixture
+def terminated_certificate_data_factory():
+    def factory(certificate_id: str):
+        return {
+            "id": certificate_id,
+            "status": "terminated",
+            "statusNotes": {
+                "message": "Terminating certificate for E2E test",
+            },
+        }
+
+    return factory

tests/e2e/program/certificate/test_async_certificate.py

@@ -0,0 +1,83 @@
+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_certificate(
+    async_mpt_ops, async_mpt_vendor, certificate_data, terminated_certificate_data_factory
+):
+    certificate = await async_mpt_ops.program.certificates.create(certificate_data)
+    terminated_certificate_data = terminated_certificate_data_factory(certificate.id)
+
+    yield certificate
+
+    try:
+        await async_mpt_vendor.program.certificates.terminate(
+            certificate.id, terminated_certificate_data
+        )
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to terminate certificate {certificate.id}: {error.title}")  # noqa: WPS421
+
+
+async def test_get_certificate_by_id(async_mpt_vendor, certificate_id):
+    result = await async_mpt_vendor.program.certificates.get(certificate_id)
+
+    assert result is not None
+
+
+async def test_get_certificate_by_invalid_id(async_mpt_vendor, invalid_certificate_id):
+    with pytest.raises(MPTAPIError):
+        await async_mpt_vendor.program.certificates.get(invalid_certificate_id)
+
+
+async def test_list_certificates(async_mpt_vendor):
+    limit = 10
+
+    result = await async_mpt_vendor.program.certificates.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_filter_certificates(async_mpt_vendor, certificate_id):
+    select_fields = ["-audit", "-parameters"]
+    filtered_certificates = (
+        async_mpt_vendor.program.certificates
+        .filter(RQLQuery(id=certificate_id))
+        .filter(RQLQuery(status="Active"))
+        .select(*select_fields)
+    )
+
+    result = [certificate async for certificate in filtered_certificates.iterate()]
+
+    assert len(result) == 1
+
+
+def test_create_certificate(created_certificate):
+    result = created_certificate
+
+    assert result is not None
+
+
+async def test_update_certificate(async_mpt_client, created_certificate):
+    updated_name = "E2E Updated Certificate Name"
+    update_data = {"id": created_certificate.id, "name": updated_name}
+
+    result = await async_mpt_client.program.certificates.update(created_certificate.id, update_data)
+
+    assert result is not None
+
+
+async def test_terminate_certificate(
+    async_mpt_vendor, created_certificate, terminated_certificate_data_factory
+):
+    terminated_certificate_data = terminated_certificate_data_factory(created_certificate.id)
+
+    result = await async_mpt_vendor.program.certificates.terminate(
+        created_certificate.id, terminated_certificate_data
+    )
+
+    assert result is not None

tests/e2e/program/certificate/test_sync_certificate.py

@@ -0,0 +1,79 @@
+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_certificate(mpt_ops, mpt_vendor, certificate_data, terminated_certificate_data_factory):
+    certificate = mpt_ops.program.certificates.create(certificate_data)
+    terminated_certificate_data = terminated_certificate_data_factory(certificate.id)
+
+    yield certificate
+
+    try:
+        mpt_vendor.program.certificates.terminate(certificate.id, terminated_certificate_data)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to terminate certificate {certificate.id}: {error.title}")  # noqa: WPS421
+
+
+def test_get_certificate_by_id(mpt_vendor, certificate_id):
+    result = mpt_vendor.program.certificates.get(certificate_id)
+
+    assert result is not None
+
+
+def test_get_certificate_by_invalid_id(mpt_vendor, invalid_certificate_id):
+    with pytest.raises(MPTAPIError):
+        mpt_vendor.program.certificates.get(invalid_certificate_id)
+
+
+def test_list_certificates(mpt_vendor):
+    limit = 10
+
+    result = mpt_vendor.program.certificates.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+def test_filter_certificates(mpt_vendor, certificate_id):
+    select_fields = ["-audit", "-parameters"]
+    filtered_certificates = (
+        mpt_vendor.program.certificates
+        .filter(RQLQuery(id=certificate_id))
+        .filter(RQLQuery(status="Active"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_certificates.iterate())
+
+    assert len(result) == 1
+
+
+def test_create_certificate(created_certificate):
+    result = created_certificate
+
+    assert result is not None
+
+
+def test_update_certificate(mpt_client, created_certificate):
+    updated_name = "E2E Updated Certificate Name"
+    update_data = {"id": created_certificate.id, "name": updated_name}
+
+    result = mpt_client.program.certificates.update(created_certificate.id, update_data)
+
+    assert result is not None
+
+
+def test_terminate_certificate(
+    mpt_vendor, created_certificate, terminated_certificate_data_factory
+):
+    terminated_certificate_data = terminated_certificate_data_factory(created_certificate.id)
+
+    result = mpt_vendor.program.certificates.terminate(
+        created_certificate.id, terminated_certificate_data
+    )
+
+    assert result is not None