"""Real PostgreSQL waits followed by persisted-state assertions."""
from django.contrib.auth.models import Permission
from django.core.exceptions import PermissionDenied, ValidationError
from django.db import transaction
from django.test import TransactionTestCase, Client
from django.urls import reverse
from apps.organization import test_assignment_concurrency as concurrency
from apps.inventory.test_usage import setup_usage
from apps.inventory.tests import grant
from apps.access.services import set_role_permissions
from apps.service import test_repair as repair
from apps.service.models import ServiceCase
from .test_invoice import InvoiceFixture
from . import invoice_services as s
from .models import ServiceInvoice, InvoiceAllocation, InvoiceSequence


class InvoiceConcurrencyTests(InvoiceFixture, TransactionTestCase):
    run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent

    def setUp(self): setup_usage(self)

    def assert_draft(self, row):
        row.refresh_from_db()
        self.assertEqual(row.status, "DRAFT")
        self.assertIsNone(row.finalized_at)

    def test_two_first_drafts(self):
        self.quoted()
        self.run_concurrent(self.prepare, self.prepare, expected="validation")
        self.assertEqual(ServiceInvoice.objects.count(), 1)
        self.assertEqual(InvoiceSequence.objects.get().next_value, 2)

    def test_simultaneous_number_allocation(self):
        from apps.service.tests import intake
        from apps.service.test_engineer_assignment import assign
        from apps.service import test_diagnosis as diagnosis
        self.quoted()
        original = self.case
        self.case = intake(self)
        assign(self)
        assessment = diagnosis.begin(self)
        diagnosis.add(self, assessment)
        diagnosis.complete(self, assessment)
        self.decide(self.submit(self.lines(self.create())))
        second = self.case
        self.case = original
        self.run_concurrent(self.prepare, lambda:self.prepare(service_case=second), expected="success")
        self.assertEqual(set(ServiceInvoice.objects.values_list("number", flat=True)), {"INV-00000001", "INV-00000002"})

    def test_simultaneous_finalization(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.finalize(row), expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status, "FINALIZED")
        self.assertEqual(ServiceInvoice.objects.count(), 1)

    def test_edit_before_finalization(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.reconcile(row, [self.allocation()]), lambda:self.finalize(row), expected="validation")
        self.assert_draft(row)
        self.assertEqual(row.lines.filter(is_active=True).count(), 1)

    def test_finalization_before_edit(self):
        row = self.billed()
        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, "FINALIZED")
        self.assertEqual(row.generation, 1)

    def test_finalization_before_regeneration(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.prepare(invoice=row, expected_revision=s.revision(row)), expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status, "FINALIZED")

    def test_regeneration_before_finalization(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.prepare(invoice=row, expected_revision=s.revision(row)), lambda:self.finalize(row), expected="validation")
        self.assert_draft(row)
        self.assertEqual(row.generation, 2)

    def test_two_manual_allocators_cannot_double_bill_source(self):
        row = self.billed(mixed=True)
        customer = self.quote.lines.get(responsibility="CUSTOMER")
        warranty = self.quote.lines.get(responsibility="WARRANTY")
        role = grant(self.user, self.company)
        set_role_permissions(role=role, permissions=Permission.objects.filter(codename="manage_serviceinvoice"))
        self.run_concurrent(lambda:self.reconcile(row, [self.allocation(line=warranty)]),
            lambda:self.reconcile(row, [self.allocation(line=customer)], actor=self.user), expected="validation")
        row.refresh_from_db()
        self.assertEqual((row.customer_pay_total, row.warranty_covered_total), (0, 100))
        self.assertEqual(InvoiceAllocation.objects.filter(line__is_active=True).count(), 1)

    def test_competing_allocations_for_approved_quantity(self):
        self.quoted(quantity=1, responsibility="WARRANTY")
        self.issue_row = self.prepared_issue(quantity=2)
        self.disposition = self.consume(self.issue_row, self.action)
        second = self.consume(self.issue_row, self.action)
        row = self.prepare()
        source = self.quote.lines.get(responsibility="WARRANTY")
        # Both writers replace the complete reconciliation at one reviewed
        # revision. The loser cannot silently overwrite the first selection.
        self.run_concurrent(lambda:self.reconcile(row, [self.allocation(line=source)]),
            lambda:self.reconcile(row, [self.allocation(line=source, source=second)]), expected="validation")
        self.assertEqual(list(InvoiceAllocation.objects.filter(line__is_active=True).values_list("consumption_id", flat=True)), [self.disposition.pk])

    def test_manual_resolution_before_finalization(self):
        row = self.billed(mixed=True)
        self.ready()
        self.run_concurrent(lambda:self.reconcile(row, [self.allocation()]), lambda:self.finalize(row), expected="validation")
        self.assert_draft(row)
        self.assertEqual(row.customer_pay_total, 100)

    def test_unresolved_ambiguity_stays_draft_during_competing_finalization(self):
        row = self.billed(mixed=True)
        self.ready()
        def rejected():
            with self.assertRaises(ValidationError): self.finalize(row)
            ServiceCase.objects.select_for_update().get(pk=self.case.pk)
        self.run_concurrent(rejected, lambda:self.finalize(row), expected="validation")
        self.assert_draft(row)
        self.assertFalse(row.lines.exists())

    def test_revision_before_allocation(self):
        row = self.billed()
        self.run_concurrent(lambda:self.revise(self.quote), lambda:self.reconcile(row, [self.allocation()]), expected="validation")
        self.assert_draft(row)
        self.assertEqual(row.generation, 1)

    def test_allocation_before_revision_leaves_stale_draft(self):
        row = self.billed()
        self.run_concurrent(lambda:self.reconcile(row, [self.allocation()]), lambda:self.revise(self.quote), expected="success")
        row.refresh_from_db()
        with self.assertRaises(ValidationError): self.reconcile(row, [self.allocation()])
        self.assert_draft(row)

    def test_finalization_before_quotation_revision(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.revise(self.quote), expected="validation")
        row.refresh_from_db()
        self.quote.refresh_from_db()
        self.assertEqual((row.status, self.quote.status, self.quote.is_current), ("FINALIZED", "APPROVED", True))

    def test_approval_change_cannot_race_finalization(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.decide(self.quote, outcome="REJECTED"), expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status, "FINALIZED")
        self.assertEqual(self.quote.decision.outcome, "APPROVED")

    def test_finalization_before_consumption(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.consume(self.issue_row, self.action), expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.grand_total, 100)
        self.assertEqual(self.issue_row.dispositions.filter(kind="CONSUMED").count(), 1)

    def test_consumption_before_draft_reconcile_requires_full_final_reconciliation(self):
        row = self.billed()
        self.run_concurrent(lambda:self.consume(self.issue_row, self.action), lambda:self.reconcile(row, [self.allocation()]), expected="success")
        row.refresh_from_db()
        self.ready()
        with self.assertRaises(ValidationError): self.finalize(row)
        self.assert_draft(row)

    def test_unused_return_before_finalization(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.return_unused(self.issue_row), lambda:self.finalize(row), expected="success")
        row.refresh_from_db()
        self.assertEqual((row.status, row.grand_total), ("FINALIZED", 100))

    def test_finalization_before_unused_return(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:self.return_unused(self.issue_row), expected="success")
        row.refresh_from_db()
        self.assertEqual((row.status, row.grand_total), ("FINALIZED", 100))

    def test_unused_return_before_manual_allocation(self):
        row = self.billed(mixed=True)
        self.run_concurrent(lambda:self.return_unused(self.issue_row), lambda:self.reconcile(row, [self.allocation()]), expected="success")
        row.refresh_from_db()
        self.assertEqual(row.grand_total, 100)
        self.assertEqual(InvoiceAllocation.objects.get().consumption, self.disposition)

    def test_finalization_before_repair_edit(self):
        row = self.billed()
        self.ready()
        self.run_concurrent(lambda:self.finalize(row), lambda:repair.add(self, self.execution), expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status, "FINALIZED")

    def test_authorization_revocation_before_finalization(self):
        row = self.billed()
        self.ready()
        role = grant(self.user, self.company, center=self.center)
        set_role_permissions(role=role, permissions=Permission.objects.filter(codename="finalize_serviceinvoice"))
        def rejected():
            with self.assertRaises(PermissionDenied): self.finalize(row, actor=self.user)
        self.run_concurrent(lambda:set_role_permissions(role=role, permissions=[]), rejected, expected="success")
        self.assert_draft(row)

    def test_handover_before_finalization(self):
        from apps.service.test_handover import released, handover
        row = self.billed()
        self.ready()
        self.received = []
        released(self)
        self.run_concurrent(lambda:handover(self), lambda:self.finalize(row), expected="validation")
        self.assert_draft(row)
        self.case.refresh_from_db()
        self.assertEqual(self.case.status, "DELIVERED")

    def test_finalization_before_handover(self):
        from apps.service.test_handover import released, handover
        row = self.billed(responsibility="WARRANTY")
        self.ready()
        self.received = []
        released(self)
        self.run_concurrent(lambda:self.finalize(row), lambda:handover(self), expected="success")
        row.refresh_from_db()
        self.case.refresh_from_db()
        self.assertEqual((row.status, self.case.status), ("FINALIZED", "DELIVERED"))

    def test_creation_rollback_racing_creation(self):
        self.quoted()
        def rollback():
            with self.assertRaises(RuntimeError), transaction.atomic():
                self.prepare()
                raise RuntimeError("Synthetic abort")
            ServiceCase.objects.select_for_update().get(pk=self.case.pk)
        self.run_concurrent(rollback, self.prepare, expected="success")
        self.assertEqual(ServiceInvoice.objects.get().number, "INV-00000001")
        self.assertEqual(InvoiceSequence.objects.get().next_value, 2)

    def test_finalization_rollback_racing_finalization(self):
        row = self.billed()
        self.ready()
        def rollback():
            with self.assertRaises(RuntimeError), transaction.atomic():
                self.finalize(row)
                raise RuntimeError("Synthetic abort")
            ServiceCase.objects.select_for_update().get(pk=self.case.pk)
        self.run_concurrent(rollback, lambda:self.finalize(row), expected="success")
        row.refresh_from_db()
        self.assertEqual(row.status, "FINALIZED")

    def test_stale_admin_finalization_after_competing_allocation(self):
        row = self.billed()
        self.ready()
        client = Client()
        client.force_login(self.actor)
        url = reverse("admin:commercial_invoice_workflow", args=[row.pk])
        token = client.get(url).context_data["revision_token"]
        def post():
            self.assertEqual(client.post(url, dict(operation="finalize", revision_token=token)).status_code, 400)
        self.run_concurrent(lambda:self.reconcile(row, [self.allocation()]), post, expected="success")
        self.assert_draft(row)