Added e2e tests for custom ledgers endpoints

Author: Robert Segal

e2e_config.test.json

@@ -12,6 +12,9 @@
   "accounts.user.id": "USR-9673-3314",
   "accounts.user_group.id": "UGR-6822-0561",
   "audit.record.id": "AUD-3748-4760-1006-3938",
+  "billing.custom_ledger.attachment.id": "CLA-1777-7485",
+  "billing.custom_ledger.charge.id": "CHG-2665-3524-0000-0000-0020",
+  "billing.custom_ledger.id": "CLE-2665-3524",
   "billing.journal.attachment.id": "JOA-6425-9776",
   "billing.journal.id": "BJO-6562-0928",
   "catalog.authorization.id": "AUT-9288-6146",

mpt_api_client/constants.py

@@ -1 +1,2 @@
 APPLICATION_JSON = "application/json"
+MIMETYPE_EXCEL_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"

mpt_api_client/resources/billing/custom_ledger_upload.py

@@ -1,10 +1,16 @@
+import pathlib
+from typing import cast
+from urllib.parse import urljoin
+
+from mpt_api_client.constants import MIMETYPE_EXCEL_XLSX
 from mpt_api_client.http import AsyncService, Service
 from mpt_api_client.http.mixins import (
     AsyncCollectionMixin,
     AsyncManagedResourceMixin,
     CollectionMixin,
     ManagedResourceMixin,
 )
+from mpt_api_client.http.types import FileContent, FileTypes
 from mpt_api_client.models import Model
 from mpt_api_client.resources.billing.custom_ledger_attachments import (
     AsyncCustomLedgerAttachmentsService,
@@ -14,10 +20,6 @@
     AsyncCustomLedgerChargesService,
     CustomLedgerChargesService,
 )
-from mpt_api_client.resources.billing.custom_ledger_upload import (
-    AsyncCustomLedgerUploadService,
-    CustomLedgerUploadService,
-)
 from mpt_api_client.resources.billing.mixins import AcceptableMixin, AsyncAcceptableMixin
 
 
@@ -31,6 +33,8 @@ class CustomLedgersServiceConfig:
     _endpoint = "/public/v1/billing/custom-ledgers"
     _model_class = CustomLedger
     _collection_key = "data"
+    _upload_file_key = "file"
+    _upload_data_key = "id"
 
 
 class CustomLedgersService(
@@ -42,20 +46,48 @@ class CustomLedgersService(
 ):
     """Custom Ledgers service."""
 
+    def upload(self, custom_ledger_id: str, file: FileTypes) -> CustomLedger:
+        """Upload custom ledger file.
+
+        Args:
+            custom_ledger_id: Custom Ledger ID.
+            file: Custom Ledger file.
+
+        Returns:
+            CustomLedger: Created resource.
+        """
+        files: dict[str, FileTypes] = {}
+
+        filename = pathlib.Path(getattr(file, "name", "uploaded_file.xlsx")).name
+
+        # Mimetype is set to Excel XLSX to prevent 415 response from the server
+        files[self._upload_file_key] = (
+            filename,
+            cast("FileContent", file),
+            MIMETYPE_EXCEL_XLSX,
+        )  # UNUSED type: ignore[attr-defined]
+        files[self._upload_data_key] = custom_ledger_id  # UNUSED type: ignore
+
+        path = urljoin(f"{self.path}/", f"{custom_ledger_id}/upload")
+
+        response = self.http_client.request(  # UNUSED type: ignore[attr-defined]
+            "post",
+            path,  # UNUSED type: ignore[attr-defined]
+            files=files,
+            force_multipart=True,
+        )
+
+        return self._model_class.from_response(
+            response
+        )  # UNUSED type: ignore[attr-defined, no-any-return]
+
     def charges(self, custom_ledger_id: str) -> CustomLedgerChargesService:
         """Return custom ledger charges service."""
         return CustomLedgerChargesService(
             http_client=self.http_client,
             endpoint_params={"custom_ledger_id": custom_ledger_id},
         )
 
-    def upload(self, custom_ledger_id: str) -> CustomLedgerUploadService:
-        """Get the Custom Ledger Upload service."""
-        return CustomLedgerUploadService(
-            http_client=self.http_client,
-            endpoint_params={"custom_ledger_id": custom_ledger_id},
-        )
-
     def attachments(self, custom_ledger_id: str) -> CustomLedgerAttachmentsService:
         """Return custom ledger attachments service."""
         return CustomLedgerAttachmentsService(
@@ -73,20 +105,48 @@ class AsyncCustomLedgersService(
 ):
     """Async Custom Ledgers service."""
 
+    async def upload(self, custom_ledger_id: str, file: FileTypes) -> CustomLedger:
+        """Upload custom ledger file.
+
+        Args:
+            custom_ledger_id: Custom Ledger ID.
+            file: Custom Ledger file.
+
+        Returns:
+            CustomLedger: Created resource.
+        """
+        files: dict[str, FileTypes] = {}
+
+        filename = pathlib.Path(getattr(file, "name", "uploaded_file.xlsx")).name
+
+        # Mimetype is set to Excel XLSX to prevent 415 response from the server
+        files[self._upload_file_key] = (
+            filename,
+            cast("FileContent", file),
+            MIMETYPE_EXCEL_XLSX,
+        )  # UNUSED type: ignore[attr-defined]
+        files[self._upload_data_key] = custom_ledger_id  # UNUSED type: ignore
+
+        path = urljoin(f"{self.path}/", f"{custom_ledger_id}/upload")
+
+        response = await self.http_client.request(  # UNUSED type: ignore[attr-defined]
+            "post",
+            path,  # UNUSED type: ignore[attr-defined]
+            files=files,
+            force_multipart=True,
+        )
+
+        return self._model_class.from_response(
+            response
+        )  # UNUSED type: ignore[attr-defined, no-any-return]
+
     def charges(self, custom_ledger_id: str) -> AsyncCustomLedgerChargesService:
         """Return custom ledger charges service."""
         return AsyncCustomLedgerChargesService(
             http_client=self.http_client,
             endpoint_params={"custom_ledger_id": custom_ledger_id},
         )
 
-    def upload(self, custom_ledger_id: str) -> AsyncCustomLedgerUploadService:
-        """Get the Async Custom Ledger Upload service."""
-        return AsyncCustomLedgerUploadService(
-            http_client=self.http_client,
-            endpoint_params={"custom_ledger_id": custom_ledger_id},
-        )
-
     def attachments(self, custom_ledger_id: str) -> AsyncCustomLedgerAttachmentsService:
         """Return custom ledger attachments service."""
         return AsyncCustomLedgerAttachmentsService(

mpt_api_client/resources/billing/custom_ledgers.py

@@ -119,13 +119,14 @@ per-file-ignores = [
   "mpt_api_client/mpt_client.py: WPS214 WPS235",
   "mpt_api_client/resources/*: WPS215",
   "mpt_api_client/resources/accounts/*.py: WPS202 WPS215 WPS214 WPS235 WPS453",
-  "mpt_api_client/resources/billing/*.py: WPS202 WPS204 WPS214 WPS215 WPS235",
+  "mpt_api_client/resources/billing/*.py: WPS202 WPS204 WPS214 WPS215 WPS235 WPS110",
   "mpt_api_client/resources/catalog/*.py: WPS110 WPS214 WPS215 WPS235",
   "mpt_api_client/resources/catalog/mixins.py: WPS110 WPS202 WPS214 WPS215 WPS235",
   "mpt_api_client/resources/catalog/products.py: WPS204 WPS214 WPS215 WPS235",
   "mpt_api_client/resources/commerce/*.py: WPS235 WPS215",
   "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/catalog/*.py: WPS202 WPS421",
   "tests/e2e/catalog/items/*.py: WPS110 WPS202",
   "tests/e2e/commerce/*.py: WPS202 WPS204 WPS453",

pyproject.toml

@@ -0,0 +1,24 @@
+import pytest
+
+
+@pytest.fixture
+def custom_ledger_attachment_id(e2e_config):
+    return e2e_config["billing.custom_ledger.attachment.id"]
+
+
+@pytest.fixture
+def invalid_custom_ledger_attachment_id():
+    return "CLA-0000-0000"
+
+
+@pytest.fixture
+def custom_ledger_attachment_factory():
+    def factory(
+        name: str = "E2E Created Custom Ledger Attachment",
+    ):
+        return {
+            "name": name,
+            "description": name,
+        }
+
+    return factory

tests/data/test_custom_ledger.xlsx

@@ -0,0 +1,110 @@
+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_custom_ledger_attachment(
+    async_mpt_ops, custom_ledger_attachment_factory, custom_ledger_id, pdf_fd
+):
+    new_custom_ledger_attachment_request_data = custom_ledger_attachment_factory(
+        name="E2E Created Custom Ledger Attachment",
+    )
+    custom_ledger_attachments = async_mpt_ops.billing.custom_ledgers.attachments(custom_ledger_id)
+
+    created_attachment = await custom_ledger_attachments.create(
+        new_custom_ledger_attachment_request_data, file=pdf_fd
+    )
+
+    yield created_attachment
+
+    try:
+        await custom_ledger_attachments.delete(created_attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete custom ledger attachment: {error.title}")  # noqa: WPS421
+
+
+@pytest.fixture
+def custom_ledger_attachments(async_mpt_ops, custom_ledger_id):
+    return async_mpt_ops.billing.custom_ledgers.attachments(custom_ledger_id)
+
+
+async def test_get_custom_ledger_attachment_by_id(
+    custom_ledger_attachments, custom_ledger_attachment_id
+):
+    result = await custom_ledger_attachments.get(custom_ledger_attachment_id)
+
+    assert result is not None
+
+
+async def test_get_custom_ledger_attachment_by_id_not_found(
+    custom_ledger_attachments, invalid_custom_ledger_attachment_id
+):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await custom_ledger_attachments.get(invalid_custom_ledger_attachment_id)
+
+
+async def test_list_custom_ledger_attachments(custom_ledger_attachments):
+    limit = 10
+
+    result = await custom_ledger_attachments.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_filter_custom_ledger_attachments(
+    custom_ledger_attachments, custom_ledger_attachment_id
+):
+    select_fields = ["-price"]
+    filtered_attachments = (
+        custom_ledger_attachments.filter(RQLQuery(id=custom_ledger_attachment_id))
+        .filter(RQLQuery(name="test_custom_ledger.xlsx"))
+        .select(*select_fields)
+    )
+
+    result = [attachment async for attachment in filtered_attachments.iterate()]
+
+    assert len(result) == 1
+
+
+def test_create_billing_custom_ledger_attachment(created_custom_ledger_attachment):
+    result = created_custom_ledger_attachment
+
+    assert result is not None
+
+
+async def test_update_billing_custom_ledger_attachment(
+    custom_ledger_attachments, created_custom_ledger_attachment
+):
+    updated_name = "E2E Updated Custom Ledger Attachment"
+
+    update_data = {
+        "name": updated_name,
+    }
+
+    updated_attachment = await custom_ledger_attachments.update(
+        created_custom_ledger_attachment.id,
+        update_data,
+    )
+
+    assert updated_attachment.name == updated_name
+
+
+async def test_delete_billing_custom_ledger_attachment(
+    custom_ledger_attachments, created_custom_ledger_attachment
+):
+    result = created_custom_ledger_attachment
+
+    await custom_ledger_attachments.delete(result.id)
+
+
+async def test_download_billing_custom_ledger_attachment(
+    custom_ledger_attachments, created_custom_ledger_attachment
+):
+    result = await custom_ledger_attachments.download(created_custom_ledger_attachment.id)
+
+    assert result.file_contents is not None
+    assert result.filename is not None