[MPT-14915] Added e2e tests for billing ledger, charges, and ledger attachments

e2e_config.test.json

@@ -17,6 +17,9 @@
   "billing.custom_ledger.id": "CLE-2665-3524",
   "billing.journal.attachment.id": "JOA-6425-9776",
   "billing.journal.id": "BJO-6562-0928",
+  "billing.ledger.attachment.id": "LEA-4971-4321",
+  "billing.ledger.charge.id": "CHG-2589-1434-0000-0000-0200",
+  "billing.ledger.id": "BLE-2589-1434-7310-3075",
   "catalog.authorization.id": "AUT-9288-6146",
   "catalog.listing.id": "LST-5489-0806",
   "catalog.price_list.id": "PRC-7255-3950-0245",

tests/e2e/billing/ledger/attachment/conftest.py

@@ -0,0 +1,24 @@
+import pytest
+
+
+@pytest.fixture
+def invalid_ledger_attachment_id():
+    return "LEA-0000-0000"
+
+
+@pytest.fixture
+def ledger_attachment_id(e2e_config):
+    return e2e_config["billing.ledger.attachment.id"]
+
+
+@pytest.fixture
+def ledger_attachment_factory():
+    def factory(
+        name: str = "E2E Created Ledger Attachment",
+    ):
+        return {
+            "name": name,
+            "description": name,
+        }
+
+    return factory

tests/e2e/billing/ledger/attachment/test_async_ledger_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_ledger_attachment(async_mpt_ops, ledger_attachment_factory, ledger_id, pdf_fd):
+    new_ledger_attachment_request_data = ledger_attachment_factory(
+        name="E2E Seeded Ledger Attachment",
+    )
+    ledger_attachments = async_mpt_ops.billing.ledgers.attachments(ledger_id)
+    created_ledger_attachment = await ledger_attachments.create(
+        new_ledger_attachment_request_data, file=pdf_fd
+    )
+
+    yield created_ledger_attachment
+
+    try:
+        await ledger_attachments.delete(created_ledger_attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete ledger attachment: {error.title}")
+
+
+@pytest.fixture
+def ledger_attachments(async_mpt_ops, ledger_id):
+    return async_mpt_ops.billing.ledgers.attachments(ledger_id)
+
+
+async def test_get_ledger_attachment_by_id(ledger_attachments, ledger_attachment_id):
+    result = await ledger_attachments.get(ledger_attachment_id)
+
+    assert result is not None
+
+
+async def test_get_ledger_attachment_by_id_not_found(
+    ledger_attachments, invalid_ledger_attachment_id
+):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await ledger_attachments.get(invalid_ledger_attachment_id)
+
+
+async def test_list_ledger_attachments(ledger_attachments):
+    limit = 10
+
+    result = await ledger_attachments.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_filter_ledger_attachments(ledger_attachments, ledger_attachment_id):
+    select_fields = ["-description"]
+    filtered_attachments = (
+        ledger_attachments.filter(RQLQuery(id=ledger_attachment_id))
+        .filter(RQLQuery(name="E2E Seeded Ledger Attachment"))
+        .select(*select_fields)
+    )
+
+    result = [attachment async for attachment in filtered_attachments.iterate()]
+
+    assert len(result) > 0
+
+
+def test_create_billing_ledger_attachment(created_ledger_attachment):
+    result = created_ledger_attachment
+
+    assert result is not None
+
+
+async def test_update_billing_ledger_attachment(
+    ledger_attachments, created_ledger_attachment, ledger_attachment_factory
+):
+    updated_data = ledger_attachment_factory(
+        name="E2E Updated Ledger Attachment",
+    )
+
+    result = await ledger_attachments.update(created_ledger_attachment.id, updated_data)
+
+    assert result is not None
+
+
+async def test_delete_billing_ledger_attachment(ledger_attachments, created_ledger_attachment):
+    result = created_ledger_attachment
+
+    await ledger_attachments.delete(result.id)
+
+
+async def test_download_billing_ledger_attachment(ledger_attachments, ledger_attachment_id):
+    result = await ledger_attachments.download(ledger_attachment_id)
+
+    assert result.file_contents is not None
+    assert result.filename is not None

tests/e2e/billing/ledger/attachment/test_sync_ledger_attachment.py

@@ -0,0 +1,93 @@
+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_ledger_attachment(mpt_ops, ledger_attachment_factory, ledger_id, pdf_fd):
+    new_ledger_attachment_request_data = ledger_attachment_factory(
+        name="E2E Created Ledger Attachment",
+    )
+    ledger_attachments = mpt_ops.billing.ledgers.attachments(ledger_id)
+
+    created_ledger_attachment = ledger_attachments.create(
+        new_ledger_attachment_request_data, file=pdf_fd
+    )
+
+    yield created_ledger_attachment
+
+    try:
+        ledger_attachments.delete(created_ledger_attachment.id)
+    except MPTAPIError as error:
+        print(f"TEARDOWN - Unable to delete ledger attachment: {error.title}")
+
+
+@pytest.fixture
+def ledger_attachments(mpt_ops, ledger_id):
+    return mpt_ops.billing.ledgers.attachments(ledger_id)
+
+
+def test_get_ledger_attachment_by_id(ledger_attachments, ledger_attachment_id):
+    result = ledger_attachments.get(ledger_attachment_id)
+
+    assert result is not None
+
+
+def test_get_ledger_attachment_by_id_not_found(ledger_attachments, invalid_ledger_attachment_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        ledger_attachments.get(invalid_ledger_attachment_id)
+
+
+def test_list_ledger_attachments(ledger_attachments):
+    limit = 10
+
+    result = ledger_attachments.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+def test_filter_ledger_attachments(ledger_attachments, ledger_attachment_id):
+    select_fields = ["-description"]
+    filtered_attachments = (
+        ledger_attachments.filter(RQLQuery(id=ledger_attachment_id))
+        .filter(RQLQuery(name="E2E Seeded Ledger Attachment"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_attachments.iterate())
+
+    assert len(result) > 0
+
+
+def test_create_billing_ledger_attachment(created_ledger_attachment):
+    result = created_ledger_attachment
+
+    assert result is not None
+
+
+def test_update_billing_ledger_attachment(
+    ledger_attachments, created_ledger_attachment, ledger_attachment_factory
+):
+    updated_data = ledger_attachment_factory(
+        name="E2E Updated Ledger Attachment",
+    )
+
+    result = ledger_attachments.update(created_ledger_attachment.id, updated_data)
+
+    assert result is not None
+
+
+def test_delete_billing_ledger_attachment(ledger_attachments, created_ledger_attachment):
+    result = created_ledger_attachment
+
+    ledger_attachments.delete(result.id)
+
+
+def test_download_billing_ledger_attachment(ledger_attachments, ledger_attachment_id):
+    result = ledger_attachments.download(ledger_attachment_id)
+
+    assert result.file_contents is not None
+    assert result.filename is not None

tests/e2e/billing/ledger/charge/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+
+
+@pytest.fixture
+def ledger_charge_id(e2e_config):
+    return e2e_config["billing.ledger.charge.id"]
+
+
+@pytest.fixture
+def invalid_ledger_charge_id():
+    return "CHG-0000-0000-0000-0000-0000"

tests/e2e/billing/ledger/charge/test_async_ledger_charge.py

@@ -0,0 +1,43 @@
+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 ledger_charges(async_mpt_ops, ledger_id):
+    return async_mpt_ops.billing.ledgers.charges(ledger_id)
+
+
+async def test_get_ledger_charge_by_id(ledger_charges, ledger_charge_id):
+    result = await ledger_charges.get(ledger_charge_id)
+
+    assert result is not None
+
+
+async def test_get_ledger_charge_by_id_not_found(ledger_charges, invalid_ledger_charge_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await ledger_charges.get(invalid_ledger_charge_id)
+
+
+async def test_list_ledger_charges(ledger_charges):
+    limit = 10
+
+    result = await ledger_charges.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_filter_ledger_charges(ledger_charges, ledger_charge_id):
+    select_fields = ["-journal"]
+    filtered_charges = (
+        ledger_charges.filter(RQLQuery(id=ledger_charge_id))
+        .filter(RQLQuery(externalIds__invoice="INV12345"))
+        .select(*select_fields)
+    )
+
+    result = [charges async for charges in filtered_charges.iterate()]
+
+    assert len(result) > 0

tests/e2e/billing/ledger/charge/test_sync_ledger_charge.py

@@ -0,0 +1,43 @@
+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 ledger_charges(mpt_ops, ledger_id):
+    return mpt_ops.billing.ledgers.charges(ledger_id)
+
+
+def test_get_ledger_charge_by_id(ledger_charges, ledger_charge_id):
+    result = ledger_charges.get(ledger_charge_id)
+
+    assert result is not None
+
+
+def test_get_ledger_charge_by_id_not_found(ledger_charges, invalid_ledger_charge_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        ledger_charges.get(invalid_ledger_charge_id)
+
+
+def test_list_ledger_charges(ledger_charges):
+    limit = 10
+
+    result = ledger_charges.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+def test_filter_ledger_charges(ledger_charges, ledger_charge_id):
+    select_fields = ["-journal"]
+    filtered_charges = (
+        ledger_charges.filter(RQLQuery(id=ledger_charge_id))
+        .filter(RQLQuery(externalIds__invoice="INV12345"))
+        .select(*select_fields)
+    )
+
+    result = list(filtered_charges.iterate())
+
+    assert len(result) > 0

tests/e2e/billing/ledger/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+
+
+@pytest.fixture
+def ledger_id(e2e_config):
+    return e2e_config["billing.ledger.id"]
+
+
+@pytest.fixture
+def invalid_ledger_id():
+    return "BLE-0000-0000-0000-0000"

tests/e2e/billing/ledger/test_async_ledger.py

@@ -0,0 +1,38 @@
+import pytest
+
+from mpt_api_client.exceptions import MPTAPIError
+from mpt_api_client.rql.query_builder import RQLQuery
+
+pytestmark = [pytest.mark.flaky]
+
+
+async def test_get_billing_ledger_by_id(async_mpt_ops, ledger_id):
+    result = await async_mpt_ops.billing.ledgers.get(ledger_id)
+
+    assert result is not None
+
+
+async def test_list_billing_ledgers(async_mpt_ops):
+    limit = 10
+
+    result = await async_mpt_ops.billing.ledgers.fetch_page(limit=limit)
+
+    assert len(result) > 0
+
+
+async def test_get_billing_ledger_by_id_not_found(async_mpt_ops, invalid_ledger_id):
+    with pytest.raises(MPTAPIError, match=r"404 Not Found"):
+        await async_mpt_ops.billing.ledgers.get(invalid_ledger_id)
+
+
+async def test_filter_billing_ledgers(async_mpt_ops, ledger_id, commerce_product_id):
+    select_fields = ["-authorization"]
+    filtered_ledgers = (
+        async_mpt_ops.billing.ledgers.filter(RQLQuery(id=ledger_id))
+        .filter(RQLQuery(f"product.id={commerce_product_id}"))
+        .select(*select_fields)
+    )
+
+    result = [ledger async for ledger in filtered_ledgers.iterate()]
+
+    assert len(result) > 0