from unittest.mock import patch
from django.core.exceptions import PermissionDenied
from django.test import TransactionTestCase
from apps.organization import test_assignment_concurrency as concurrency
from apps.inventory.test_usage import setup_usage
from apps.service import test_repair as repair
from apps.service.services import cancel_service_case
from apps.parts import services as parts
from .tests import QuotationFixture
from .models import ServiceQuotation, QuotationFamily, QuotationDecision, CommercialWorkAuthorization


class QuotationConcurrencyTests(QuotationFixture,TransactionTestCase):
    run_concurrent=concurrency.AssignmentConcurrencyTests.run_concurrent

    def setUp(self): setup_usage(self)

    def test_simultaneous_first_quotation(self):
        self.run_concurrent(self.create,self.create,expected="validation")
        self.assertEqual(ServiceQuotation.objects.count(),1)

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

    def test_draft_edit_before_submission(self):
        row=self.lines(self.create())
        self.run_concurrent(lambda:self.update(row),lambda:self.submit(row),expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status,"DRAFT")

    def test_submission_before_draft_edit(self):
        row=self.lines(self.create())
        self.run_concurrent(lambda:self.submit(row),lambda:self.update(row),expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status,"SUBMITTED")

    def test_line_edit_before_submission(self):
        row=self.lines(self.create())
        self.run_concurrent(lambda:self.lines(row,[self.line(unit_price="200")]),lambda:self.submit(row),expected="validation")
        row.refresh_from_db()
        self.assertEqual((row.status,row.grand_total),("DRAFT",200))

    def test_submission_before_line_edit(self):
        row=self.lines(self.create())
        self.run_concurrent(lambda:self.submit(row),lambda:self.lines(row),expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status,"SUBMITTED")

    def test_approval_before_rejection(self):
        row=self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.decide(row),lambda:self.decide(row,outcome="REJECTED"),expected="validation")
        self.assertEqual(QuotationDecision.objects.get().outcome,"APPROVED")

    def test_rejection_before_approval(self):
        row=self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.decide(row,outcome="REJECTED"),lambda:self.decide(row),expected="validation")
        self.assertEqual(QuotationDecision.objects.get().outcome,"REJECTED")

    def test_approval_before_stale_revision(self):
        row=self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.decide(row),lambda:self.revise(row),expected="validation")
        self.assertEqual(ServiceQuotation.objects.count(),1)

    def test_revision_before_approval(self):
        row=self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.revise(row),lambda:self.decide(row),expected="validation")
        self.assertFalse(QuotationDecision.objects.exists())

    def test_two_competing_revisions(self):
        row=self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.revise(row),lambda:self.revise(row),expected="validation")
        self.assertEqual(ServiceQuotation.objects.count(),2)
        self.assertEqual(ServiceQuotation.objects.filter(is_current=True).count(),1)

    def test_case_cancellation_before_submission(self):
        row=self.lines(self.create())
        self.run_concurrent(lambda:cancel_service_case(service_case=self.case,cancelled_by=self.actor,reason="Synthetic"),lambda:self.submit(row),expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status,"DRAFT")

    def test_part_deactivation_before_submission(self):
        row=self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])
        self.run_concurrent(lambda:parts.deactivate_spare_part(spare_part=self.part),lambda:self.submit(row),expected="validation")
        row.refresh_from_db()
        self.assertEqual(row.status,"DRAFT")

    def test_submission_before_part_deactivation_preserves_snapshot(self):
        row=self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])
        self.run_concurrent(lambda:self.submit(row),lambda:parts.deactivate_spare_part(spare_part=self.part),expected="success")
        row.refresh_from_db()
        self.assertEqual(row.status,"SUBMITTED")
        self.assertEqual(row.lines.get().part_code,self.part.part_code)

    def test_permission_revocation_before_mutation(self):
        from apps.inventory.tests import grant
        from apps.access.services import set_role_permissions
        from django.contrib.auth.models import Permission
        role=grant(self.user,self.company,center=self.center)
        set_role_permissions(role=role,permissions=Permission.objects.filter(content_type__app_label="commercial"))
        row=self.create()
        def mutate():
            with self.assertRaises(PermissionDenied): self.update(row,actor=self.user)
        self.run_concurrent(lambda:set_role_permissions(role=role,permissions=[]),mutate,expected="success")
        row.refresh_from_db()
        self.assertEqual(row.customer_note,"")

    def test_approval_first_allows_waiting_performance(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.submit(self.lines(self.create()))
        self.run_concurrent(lambda:self.decide(row),lambda:repair.perform(self,action),expected="success")
        self.assertEqual(CommercialWorkAuthorization.objects.get().quotation_id,row.pk)

    def test_revision_first_blocks_performance(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.decide(self.submit(self.lines(self.create())))
        self.run_concurrent(lambda:self.revise(row),lambda:repair.perform(self,action),expected="validation")
        action.refresh_from_db()
        self.assertIsNone(action.performed_at)

    def test_performance_first_preserves_exact_old_approval_after_revision(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.decide(self.submit(self.lines(self.create())))
        self.run_concurrent(lambda:repair.perform(self,action),lambda:self.revise(row),expected="success")
        self.assertEqual(CommercialWorkAuthorization.objects.get().quotation_id,row.pk)
        self.assertEqual(ServiceQuotation.objects.get(is_current=True).status,"DRAFT")

    def test_first_customer_declaration_blocks_performance(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.create()
        self.run_concurrent(lambda:self.lines(row),lambda:repair.perform(self,action),expected="validation")
        action.refresh_from_db()
        self.assertIsNone(action.performed_at)

    def test_performance_first_rejects_retrospective_customer_declaration(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.create()
        self.run_concurrent(lambda:repair.perform(self,action),lambda:self.lines(row),expected="validation")
        self.assertFalse(QuotationFamily.objects.get().approval_obligation)

    def test_revision_before_completion_blocks(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.decide(self.submit(self.lines(self.create())))
        repair.perform(self,action)
        self.run_concurrent(lambda:self.revise(row),lambda:repair.complete(self,execution),expected="validation")
        execution.refresh_from_db()
        self.assertEqual(execution.status,"OPEN")

    def test_completion_before_revision_rejects_late_revision(self):
        execution,action=repair.prepared(self,performed=False)
        row=self.decide(self.submit(self.lines(self.create())))
        repair.perform(self,action)
        self.run_concurrent(lambda:repair.complete(self,execution),lambda:self.revise(row),expected="validation")
        execution.refresh_from_db()
        self.assertEqual(execution.status,"COMPLETED")

    def test_failed_submission_rolls_back_before_waiting_submission(self):
        row=self.lines(self.create())
        def failed():
            from django.db import transaction
            try:
                with transaction.atomic():
                    self.submit(row)
                    raise RuntimeError("Injected")
            except RuntimeError: pass
            # Hold the same case lock after rollback, allowing the harness to
            # observe the waiting retry against a real PostgreSQL row lock.
            from apps.service.models import ServiceCase
            ServiceCase.objects.select_for_update().get(pk=self.case.pk)
        self.run_concurrent(failed,lambda:self.submit(row),expected="success")
        row.refresh_from_db()
        self.assertEqual(row.status,"SUBMITTED")

    def test_failed_approval_rolls_back_before_waiting_approval(self):
        row=self.submit(self.lines(self.create()))
        def failed():
            from django.db import transaction
            from apps.service.models import ServiceCase
            try:
                with transaction.atomic():
                    self.decide(row)
                    raise RuntimeError("Injected")
            except RuntimeError: pass
            ServiceCase.objects.select_for_update().get(pk=self.case.pk)
        self.run_concurrent(failed,lambda:self.decide(row),expected="success")
        self.assertEqual(QuotationDecision.objects.count(),1)

    def test_revision_first_blocks_consumption(self):
        issue=self.prepared_issue(quantity=1)
        _,action=repair.prepared(self,performed=False)
        row=self.decide(self.submit(self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])))
        repair.perform(self,action)
        self.run_concurrent(lambda:self.revise(row),lambda:self.consume(issue,action),expected="validation")
        self.assertFalse(CommercialWorkAuthorization.objects.filter(operation="CONSUME").exists())

    def test_consumption_first_preserves_old_revision_evidence(self):
        issue=self.prepared_issue(quantity=1)
        _,action=repair.prepared(self,performed=False)
        row=self.decide(self.submit(self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])))
        repair.perform(self,action)
        self.run_concurrent(lambda:self.consume(issue,action),lambda:self.revise(row),expected="success")
        self.assertEqual(CommercialWorkAuthorization.objects.get(operation="CONSUME").quotation_id,row.pk)

    def test_part_rename_first_is_snapshotted_at_submission(self):
        row=self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])
        self.run_concurrent(lambda:parts.update_spare_part(spare_part=self.part,name="Reviewed replacement"),lambda:self.submit(row),expected="success")
        self.assertEqual(row.lines.get().part_name,"Reviewed replacement")

    def test_mutation_first_commits_before_permission_revocation(self):
        from apps.inventory.tests import grant
        from apps.access.services import set_role_permissions
        from django.contrib.auth.models import Permission
        role=grant(self.user,self.company,center=self.center)
        set_role_permissions(role=role,permissions=Permission.objects.filter(content_type__app_label="commercial"))
        row=self.create()
        self.run_concurrent(lambda:self.update(row,actor=self.user),lambda:set_role_permissions(role=role,permissions=[]),expected="success")
        row.refresh_from_db()
        self.assertEqual(row.customer_note,"Updated")
        with self.assertRaises(PermissionDenied): self.update(row,actor=self.user)