"""Cross-module commercial audit schedules on independent PostgreSQL connections."""
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from unittest.mock import patch

from django.core.exceptions import ValidationError
from django.db import connections, connection
from django.test import TransactionTestCase

from apps.commercial.tests import QuotationFixture
from apps.commercial import services as quotation
from apps.devices.models import Device
from apps.inventory.test_usage import setup_usage
from apps.commercial.test_payment import PaymentFixture
from apps.commercial import models as m, payment_services as payments
from apps.organization import test_assignment_concurrency as harness
from apps.service import test_repair as repair, test_handover as delivery
from apps.service.models import ServiceCaseHandover
from apps.inventory.models import PartsDisposition
from apps.inventory.tests import grant
from apps.access.services import set_role_permissions
from django.contrib.auth.models import Permission


class CommercialLockAudit(QuotationFixture, TransactionTestCase):
    def setUp(self):
        setup_usage(self)

    def test_overlapping_quotation_edits_do_not_upgrade_dependency_locks(self):
        row = self.lines(self.create())
        barrier = Barrier(2)
        original = quotation.share

        def shared(model, pk):
            result = original(model, pk)
            if model is Device:
                barrier.wait(timeout=10)
            return result

        def edit():
            try:
                with connection.cursor() as cursor:
                    cursor.execute("SET lock_timeout = '8s'")
                try:
                    self.update(row)
                except ValidationError:
                    return "stale"
                return "updated"
            finally:
                connections.close_all()

        with patch.object(quotation, "share", shared), ThreadPoolExecutor(max_workers=2) as pool:
            futures = [pool.submit(edit) for _ in range(2)]
            results = [future.result(timeout=20) for future in futures]
        self.assertCountEqual(results, ["updated", "stale"])
        row.refresh_from_db()
        self.assertEqual((row.status, row.customer_note, row.customer_pay_total), ("DRAFT", "Updated", 100))


class CommercialChainRaces(PaymentFixture, TransactionTestCase):
    run_concurrent = harness.AssignmentConcurrencyTests.run_concurrent

    def setUp(self):
        setup_usage(self)
        self.received = []

    def finalized(self):
        self.invoice = self.billed(quoted=1)
        self.ready()
        self.invoice = self.finalize(self.invoice)

    def test_replacement_removing_customer_lines_still_blocks_repair(self):
        self.execution, self.action = repair.prepared(self, performed=False)
        quote = self.decide(self.submit(self.lines(self.create())))
        def replace():
            self.lines(self.revise(quote), [self.line(responsibility="WARRANTY", coverage_reason="Synthetic")])
        self.run_concurrent(replace, lambda:repair.perform(self, self.action), expected="validation")
        self.action.refresh_from_db()
        self.assertIsNone(self.action.performed_at)
        self.assertTrue(m.QuotationFamily.objects.get().approval_obligation)
        self.assertFalse(m.CommercialWorkAuthorization.objects.exists())

    def test_approval_then_performance_records_exact_revision(self):
        self.execution, self.action = repair.prepared(self, performed=False)
        quote = self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.decide(quote), lambda:repair.perform(self, self.action), expected="success")
        self.assertEqual(m.CommercialWorkAuthorization.objects.get().quotation_id, quote.pk)
        self.action.refresh_from_db()
        self.assertIsNotNone(self.action.performed_at)

    def test_replacement_approval_then_consumption_records_new_revision(self):
        self.quoted(quantity=1)
        issue = self.prepared_issue(quantity=1)
        replacement = self.submit(self.revise(self.quote))
        self.run_concurrent(lambda:self.decide(replacement), lambda:self.consume(issue, self.action), expected="success")
        self.assertEqual(m.CommercialWorkAuthorization.objects.get(operation="CONSUME").quotation_id, replacement.pk)
        self.assertEqual(PartsDisposition.objects.get(kind="CONSUMED").quantity, 1)

    def test_consumption_then_old_reconciliation_cannot_finalize_missing_quantity(self):
        row = self.billed(quoted=2, consumed=1)
        old = self.allocation()
        self.run_concurrent(lambda:self.consume(self.issue_row, self.action), lambda:self.reconcile(row, [old]), expected="success")
        row.refresh_from_db()
        self.ready()
        with self.assertRaises(ValidationError):
            self.finalize(row)
        self.assertEqual(PartsDisposition.objects.filter(kind="CONSUMED").count(), 2)
        self.assertEqual(row.status, "DRAFT")

    def test_finalization_serializes_with_reconciliation(self):
        row = self.billed(quoted=1)
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.reconcile(row, [self.allocation()]), expected="validation")
        row.refresh_from_db()
        self.assertEqual((row.status, row.generation, row.customer_pay_total), ("FINALIZED", 1, 100))

    def test_finalization_then_payment_reads_committed_liability(self):
        row = self.billed(quoted=1)
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:payments.receive_service_payment(actor=self.actor,
            invoice=row, expected_revision=f"{row.pk}:0", amount="100", method="CASH"), expected="success")
        self.invoice = row
        self.assertEqual(self.summary()["state"], "PAID")
        self.assertEqual(m.ServicePaymentReceipt.objects.count(), 1)

    def test_competing_methods_cannot_allocate_same_balance(self):
        self.finalized()
        revision = self.summary()["revision"]
        self.run_concurrent(lambda:self.pay(amount="60", expected_revision=revision),
            lambda:self.pay(amount="60", method="CARD", reference="SYNTHETIC-RACE", expected_revision=revision), expected="validation")
        self.assertEqual((self.summary()["balance"], m.PaymentAllocation.objects.count()), (40, 1))

    def test_payment_invalidates_concurrent_reversal_revision(self):
        self.finalized()
        payment = self.pay(amount="20")
        revision = self.summary()["revision"]
        self.run_concurrent(lambda:self.pay(amount="0.01"),
            lambda:self.reverse_payment(payment, expected_revision=revision), expected="validation")
        self.assertEqual(str(self.summary()["paid"]), "20.01")
        self.assertFalse(m.PaymentReversal.objects.exists())

    def test_last_cent_serializes_with_handover(self):
        self.finalized()
        self.pay(amount="99.99")
        delivery.released(self)
        self.run_concurrent(lambda:self.pay(amount="0.01"), lambda:delivery.handover(self), expected="success")
        self.assertEqual(self.summary()["balance"], 0)
        self.assertEqual(ServiceCaseHandover.objects.count(), 1)

    def test_reversal_serializes_with_handover(self):
        self.finalized()
        payment = self.pay()
        delivery.released(self)
        self.run_concurrent(lambda:self.reverse_payment(payment), lambda:delivery.handover(self), expected="validation")
        self.assertFalse(ServiceCaseHandover.objects.exists())
        self.assertEqual(self.summary()["balance"], 100)
        self.assertEqual(m.ServicePaymentReceipt.objects.count(), 1)

    def test_due_release_serializes_with_handover_without_erasing_debt(self):
        self.finalized()
        self.pay(amount="10")
        delivery.released(self)
        self.run_concurrent(self.release_due, lambda:delivery.handover(self), expected="success")
        self.assertEqual((self.summary()["balance"], ServiceCaseHandover.objects.count()), (90, 1))
        self.assertEqual(m.ServiceFinancialRelease.objects.get().outstanding_amount, 90)

    def revoke_actor_before(self, operation):
        role = grant(self.user, self.company)
        set_role_permissions(role=role, permissions=Permission.objects.filter(content_type__app_label="commercial"))
        def revoke():
            self.user.is_active = False
            self.user.save(update_fields=["is_active"])
        self.run_concurrent(revoke, operation, expected="validation")

    def test_actor_revocation_before_quotation_approval(self):
        row = self.submit(self.lines(self.create()))
        self.revoke_actor_before(lambda:self.decide(row, actor=self.user))
        self.assertFalse(m.QuotationDecision.objects.exists())
        row.refresh_from_db()
        self.assertEqual(row.status, "SUBMITTED")

    def test_actor_revocation_before_invoice_finalization(self):
        row = self.billed(quoted=1)
        self.ready()
        self.revoke_actor_before(lambda:self.finalize(row, actor=self.user))
        row.refresh_from_db()
        self.assertEqual(row.status, "DRAFT")

    def test_actor_revocation_before_payment(self):
        self.finalized()
        self.revoke_actor_before(lambda:self.pay(actor=self.user))
        self.assertFalse(m.ServicePayment.objects.exists())

    def test_actor_revocation_before_reversal(self):
        self.finalized()
        payment = self.pay()
        self.revoke_actor_before(lambda:self.reverse_payment(payment, actor=self.user))
        self.assertEqual(self.summary()["state"], "PAID")
        self.assertFalse(m.PaymentReversal.objects.exists())

    def test_actor_revocation_before_due_release(self):
        self.finalized()
        self.revoke_actor_before(lambda:self.release_due(actor=self.user))
        self.assertFalse(m.ServiceFinancialRelease.objects.exists())