"""Adversarial commercial-chain checks; frozen feature tests remain unchanged.""" from decimal import Decimal from unittest.mock import patch from django.core.exceptions import ValidationError, PermissionDenied from django.db import IntegrityError, transaction, connection from django.test import TestCase, SimpleTestCase from apps.commercial.test_invoice import InvoiceFixture from apps.commercial.test_payment import PaymentFixture, shared_payment_data from apps.commercial import models as m, payment_queries as pq from apps.commercial import queries as qq, invoice_queries as iq from apps.commercial.money import calculate, decimal_value, MAX_MONEY from apps.inventory.test_usage import setup_usage from apps.service import test_handover as delivery from apps.service.models import ServiceCaseHandover class CommercialEndToEndAudit(PaymentFixture, TestCase): @classmethod def setUpTestData(cls): setup_usage(cls) def finalized(self, responsibility="CUSTOMER", **kwargs): self.invoice = self.billed(quoted=1, responsibility=responsibility, **kwargs) self.ready() self.invoice = self.finalize(self.invoice) return self.invoice def assert_chain(self, *, paid, balance): self.invoice.refresh_from_db() self.case.refresh_from_db() self.assertEqual(self.invoice.status, "FINALIZED") self.assertEqual(self.summary()["paid"], Decimal(paid)) self.assertEqual(self.summary()["balance"], Decimal(balance)) self.assertEqual(self.invoice.customer_pay_total - Decimal(paid), Decimal(balance)) line = self.invoice.lines.get(is_active=True) allocation = line.allocations.get() self.assertEqual(allocation.consumption_id, self.disposition.pk) self.assertEqual(allocation.consumption.movement.kind, "CONSUME") self.assertEqual(allocation.quantity, line.quantity) self.assertEqual(line.quotation_line.quotation.decision.outcome, "APPROVED") def test_customer_repair_to_receipt_and_delivery(self): self.finalized() payment = self.pay() self.deliver() self.assert_chain(paid=100, balance=0) self.assertEqual(self.case.status, "DELIVERED") self.assertEqual(payment.receipt.invoice_number, self.invoice.number) self.assertEqual(payment.allocation.invoice_id, self.invoice.pk) self.assertEqual(set(m.CommercialWorkAuthorization.objects.values_list("operation", flat=True)), {"PERFORM", "CONSUME", "COMPLETE"}) def test_warranty_chain_has_real_value_and_no_payment(self): self.finalized("WARRANTY") self.deliver() self.assert_chain(paid=0, balance=0) self.assertEqual(self.invoice.warranty_covered_total, 100) self.assertFalse(m.ServicePaymentReceipt.objects.exists()) def test_mixed_consumption_requires_explicit_payer_allocation(self): row = self.billed(mixed=True, consumed=2, quoted=2) self.assertEqual(row.grand_total, 0) lines = list(self.quote.lines.filter(is_active=True)) row = self.reconcile(row, [self.allocation(line=line) for line in lines]) self.ready() self.invoice = self.finalize(row) self.assertEqual((self.invoice.customer_pay_total, self.invoice.warranty_covered_total), (100, 100)) self.pay() self.deliver() self.assertEqual(self.summary()["balance"], 0) self.assertEqual(sum(m.InvoiceAllocation.objects.filter(line__is_active=True).values_list("quantity", flat=True)), 2) self.assertEqual(m.PaymentAllocation.objects.get().amount, 100) def test_partial_payment_then_final_cent_allows_delivery(self): self.finalized() self.pay(amount="99.99") with self.assertRaises(ValidationError): self.deliver() self.assertFalse(ServiceCaseHandover.objects.exists()) self.assert_chain(paid="99.99", balance="0.01") self.pay(amount="0.01") delivery.handover(self) self.assert_chain(paid=100, balance=0) self.assertEqual(m.ServicePaymentReceipt.objects.count(), 2) def test_due_release_and_terminal_closure_preserve_receivable(self): self.finalized() self.pay(amount="0.01") release = self.release_due() self.deliver() self.case.refresh_from_db() delivery.close(self) self.assert_chain(paid="0.01", balance="99.99") self.assertEqual(self.case.status, "CLOSED") self.assertEqual(release.outstanding_amount, Decimal("99.99")) self.assertEqual(pq.outstanding_customer_balance(actor=self.actor, customer=self.customer, currency="BDT"), Decimal("99.99")) self.assertEqual(pq.unsettled_invoices(actor=self.actor).get().pk, self.invoice.pk) def test_reversal_blocks_delivery_retains_complete_history(self): self.finalized() payment = self.pay() receipt = payment.receipt reversal = self.reverse_payment(payment) with self.assertRaises(ValidationError): self.deliver() self.assert_chain(paid=0, balance=100) receipt.refresh_from_db() self.assertEqual((receipt.amount, reversal.amount, payment.allocation.amount), (100, 100, 100)) self.assertFalse(ServiceCaseHandover.objects.exists()) def test_replacement_revision_rollback_restores_authoritative_approval(self): self.quoted() original = m.ServiceQuotation._persist def fail(row): original(row) if row.revision == 2: raise RuntimeError("Synthetic revision failure") with patch.object(m.ServiceQuotation, "_persist", fail), self.assertRaises(RuntimeError): self.revise(self.quote) self.quote.refresh_from_db() self.assertTrue(self.quote.is_current) self.assertEqual(self.quote.status, "APPROVED") self.assertEqual(m.ServiceQuotation.objects.count(), 1) self.assertEqual(m.QuotationDecision.objects.count(), 1) def test_invoice_allocation_failure_restores_counter_and_history(self): self.quoted(quantity=1) issue = self.prepared_issue(quantity=1) self.consume(issue, self.action) original = m.InvoiceAllocation._persist def fail(row): original(row) raise RuntimeError("Synthetic allocation failure") with patch.object(m.InvoiceAllocation, "_persist", fail), self.assertRaises(RuntimeError): self.prepare() self.assertFalse(m.ServiceInvoice.objects.exists()) self.assertFalse(m.InvoiceLine.objects.exists()) self.assertFalse(m.InvoiceSequence.objects.exists()) self.assertEqual(self.prepare().number, "INV-00000001") class SettlementIntegrityAudit(PaymentFixture, TestCase): @classmethod def setUpTestData(cls): shared_payment_data(cls) def test_stale_revision_cannot_replay_after_balance_returns_to_original(self): revision = self.summary()["revision"] payment = self.pay(amount="0.01") self.reverse_payment(payment) self.assertEqual(self.summary()["balance"], 100) with self.assertRaises(ValidationError): self.pay(expected_revision=revision) self.assertEqual(m.ServicePayment.objects.count(), 1) def test_release_is_not_settlement_and_cannot_be_replayed(self): revision = self.summary()["revision"] self.release_due() with self.assertRaises(ValidationError): self.release_due(expected_revision=revision) self.assertEqual(self.summary()["state"], "UNPAID") self.assertEqual(m.ServiceFinancialRelease.objects.count(), 1) self.assertFalse(m.ServicePayment.objects.exists()) def test_database_rejects_duplicate_and_mismatched_allocation(self): payment = self.pay() for amount in ("100.00", "0.01"): with self.subTest(amount=amount), self.assertRaises(IntegrityError), transaction.atomic(): m.PaymentAllocation.objects.bulk_create([m.PaymentAllocation(payment=payment, invoice=self.invoice, amount=amount, allocated_by=self.actor)]) self.assertEqual(m.PaymentAllocation.objects.get(payment=payment).amount, 100) def test_database_rejects_posted_financial_fact_changes(self): payment = self.pay() for change in (dict(amount=0), dict(currency="USD"), dict(method="OTHER"), dict(reference="FORGED"), dict(status="DRAFT")): with self.subTest(change=change), self.assertRaises(IntegrityError), transaction.atomic(): m.ServicePayment.objects.filter(pk=payment.pk).update(**change) self.assertEqual(self.summary()["paid"], 100) def test_queries_do_not_multiply_money_across_reversals_and_releases(self): self.pay(amount="10") payment = self.pay(amount="20") self.reverse_payment(payment) self.release_due() self.pay(amount="0.01") with self.assertNumQueries(1): summary = self.summary() self.assertEqual((summary["paid"], summary["balance"]), (Decimal("10.01"), Decimal("89.99"))) with self.assertNumQueries(1): rows = list(pq.unsettled_invoices(actor=self.actor)) self.assertEqual(len(rows), 1) with self.assertNumQueries(1): receipts = list(pq.receipt_history(actor=self.actor)) self.assertEqual(len(receipts), 3) def test_commercial_query_budgets_retain_provenance(self): with self.assertNumQueries(1): list(qq.quotation_history(actor=self.actor, service_case=self.case)) with self.assertNumQueries(2): detail = qq.quotation_detail(actor=self.actor, quotation=self.quote) list(detail.lines.all()) with self.assertNumQueries(1): list(iq.service_invoices(actor=self.actor)) with self.assertNumQueries(1): list(iq.invoice_allocations(actor=self.actor, invoice=self.invoice)) with self.assertNumQueries(4): detail = iq.invoice_detail(actor=self.actor, invoice=self.invoice) for line in detail.lines.all(): for allocation in line.allocations.all(): list(allocation.consumption.movement.unit_links.all()) def test_center_deactivation_preserves_financial_evidence(self): from apps.organization.services import deactivate_service_center payment = self.pay(amount="10") deactivate_service_center(service_center=self.center) with self.assertRaises(ValidationError): self.pay(amount="10") payment.refresh_from_db() self.assertEqual(payment.status, "POSTED") self.assertEqual(payment.receipt.amount, 10) self.assertEqual(m.PaymentAllocation.objects.get().amount, 10) def test_company_region_and_center_department_payment_scope(self): from django.contrib.auth.models import Permission from apps.organization.assignment_services import create_assignment from apps.organization.models import Department from apps.access.models import Role from apps.access.services import create_role_assignment, set_role_permissions department = Department.objects.create(company=self.company, code="AUDIT-DEPT", name="Synthetic audit department") paths = ({}, dict(region=self.region), dict(region=self.region, service_center=self.center, department=department)) for index, scope in enumerate(paths): with self.subTest(scope=scope), transaction.atomic(): path = create_assignment(user=self.user, company=self.company, **scope) role = Role.objects.create(code=f"AUDIT-SCOPE-{index}", name="Synthetic audit scope") set_role_permissions(role=role, permissions=Permission.objects.filter(content_type__app_label="commercial")) create_role_assignment(user=self.user, role=role, organization_assignment=path) payment = self.pay(actor=self.user, amount="0.01") self.assertEqual(pq.payment_detail(actor=self.user, payment=payment).pk, payment.pk) self.reverse_payment(payment, actor=self.user) self.release_due(actor=self.user) self.assertEqual(self.summary()["balance"], 100) transaction.set_rollback(True) def test_company_view_path_cannot_lend_financial_authority_to_other_center(self): from django.contrib.auth.models import Permission from apps.inventory.tests import grant from apps.access.services import set_role_permissions payment = self.pay(amount="10") view_role = grant(self.user, self.company) set_role_permissions(role=view_role, permissions=Permission.objects.filter(content_type__app_label="commercial", codename="view_servicepayment")) write_role = grant(self.user, self.company, center=self.center2) set_role_permissions(role=write_role, permissions=Permission.objects.filter(content_type__app_label="commercial")) self.assertEqual(pq.payment_detail(actor=self.user, payment=payment).pk, payment.pk) for operation in (lambda:self.pay(actor=self.user, amount="10"), lambda:self.reverse_payment(payment, actor=self.user), lambda:self.release_due(actor=self.user)): with self.assertRaises(PermissionDenied): operation() self.assertEqual(self.summary()["balance"], 90) self.assertFalse(m.PaymentReversal.objects.exists()) self.assertFalse(m.ServiceFinancialRelease.objects.exists()) class MoneyBoundaryAudit(SimpleTestCase): def test_rounding_discount_tax_and_maximum(self): values = calculate(quantity=3, unit_price="0.03", discount="0.01", tax_rate="12.50") self.assertEqual((values["subtotal"], values["tax"], values["total"]), (Decimal("0.09"), Decimal("0.01"), Decimal("0.09"))) self.assertEqual(decimal_value(MAX_MONEY), MAX_MONEY) for value in (MAX_MONEY + Decimal("0.01"), 0.01, True, "NaN", "Infinity", "0.001"): with self.subTest(value=value), self.assertRaises(ValidationError): decimal_value(value)