from datetime import timedelta from decimal import Decimal from unittest.mock import patch from django.core.exceptions import PermissionDenied, ValidationError from django.db import IntegrityError, transaction, connection from django.test import TestCase from django.utils import timezone from apps.inventory.test_usage import UsageFixture, setup_usage from apps.service import test_repair as repair from apps.inventory.models import StockMovement, StockReservation, PartsDisposition, SerializedStockUnit from apps.parts import services as parts from . import services as s, queries as q from .models import ServiceQuotation, QuotationLine, QuotationFamily, QuotationDecision, CommercialWorkAuthorization from .money import calculate class QuotationFixture(UsageFixture): def create(self, **kwargs): return s.create_service_quotation(**(dict(actor=self.actor,service_case=self.case,currency="BDT")|kwargs)) def line(self, **kwargs): return dict(kind="LABOR",responsibility="CUSTOMER",description="Proposed repair labor",quantity=1,unit_price="100.00")|kwargs def lines(self, row, inputs=None, **kwargs): return s.set_quotation_lines(**(dict(actor=self.actor,quotation=row,expected_revision=s.revision(row),lines=[self.line()] if inputs is None else inputs)|kwargs)) def submit(self, row, **kwargs): return s.submit_service_quotation(**(dict(actor=self.actor,quotation=row,expected_revision=s.revision(row))|kwargs)) def decide(self, row, **kwargs): return s.record_quotation_decision(**(dict(actor=self.actor,quotation=row,expected_revision=s.revision(row),outcome="APPROVED", channel="PHONE",recipient_name="Synthetic customer",reference="Synthetic confirmation")|kwargs)) def revise(self, row, **kwargs): return s.revise_service_quotation(**(dict(actor=self.actor,quotation=row,expected_revision=s.revision(row))|kwargs)) def update(self, row, **kwargs): return s.update_service_quotation(**(dict(actor=self.actor,quotation=row,expected_revision=s.revision(row),currency="BDT",customer_note="Updated")|kwargs)) class QuotationTests(QuotationFixture, TestCase): @classmethod def setUpTestData(cls): setup_usage(cls) def test_creation_number_and_duplicate_case(self): row=self.create() self.assertEqual((row.family.number,row.revision,row.status),("EST-00000001",1,"DRAFT")) with self.assertRaises(ValidationError): self.create() def test_creation_rollback_restores_number(self): with self.assertRaises(RuntimeError), transaction.atomic(): self.create() raise RuntimeError("Injected") self.assertEqual(self.create().family.number,"EST-00000001") def test_fixed_decimal_discount_tax_rounding(self): values=calculate(quantity=3,unit_price="0.05",discount="0.01",tax_rate="5") self.assertEqual((values["subtotal"],values["taxable_amount"],values["tax"],values["total"]), (Decimal("0.15"),Decimal("0.14"),Decimal("0.01"),Decimal("0.15"))) def test_invalid_money_rejected(self): for value in (-1, 1.1, "NaN", "Infinity", "1.001", "1000000000000"): with self.subTest(value=value), self.assertRaises(ValidationError): calculate(quantity=1,unit_price=value) for qty in (0,-1,True,1.2,1000001): with self.subTest(qty=qty), self.assertRaises(ValidationError): calculate(quantity=qty,unit_price="1") with self.assertRaises(ValidationError): calculate(quantity=1,unit_price="1",discount="2") def test_responsibility_reconciles_without_zeroing_warranty(self): row=self.lines(self.create(),[self.line(unit_price="1000"),self.line(kind="PART",spare_part=self.part, unit_price="10000",responsibility="WARRANTY",coverage_reason="Recorded intake plus technical review")]) self.assertEqual((row.grand_total,row.customer_pay_total,row.covered_total),(11000,1000,10000)) self.assertTrue(row.family.approval_obligation) def test_explicit_coverage_reason_required(self): for responsibility in ("WARRANTY","COMPANY"): row=self.create() if not ServiceQuotation.objects.exists() else ServiceQuotation.objects.get() with self.assertRaises(ValidationError): self.lines(row,[self.line(responsibility=responsibility)]) def test_tax_label_required_and_snapshot_retained(self): row=self.create() with self.assertRaises(ValidationError): self.lines(row,[self.line(tax_rate="5")]) row=self.submit(self.lines(row,[self.line(tax_rate="5",tax_label="Explicit test tax")])) self.assertEqual((row.tax_total,row.grand_total),(5,105)) def test_calculated_totals_not_accepted(self): with self.assertRaises(ValidationError): self.lines(self.create(),[self.line(total="0")]) def test_all_line_types_and_company_responsibility(self): row=self.lines(self.create(),[self.line(kind="SERVICE",responsibility="COMPANY",coverage_reason="Goodwill")]) self.assertEqual((row.customer_pay_total,row.covered_total),(0,100)) self.assertFalse(row.family.approval_obligation) def test_zero_charge_does_not_establish_obligation(self): row=self.lines(self.create(),[self.line(unit_price="0")]) self.assertFalse(row.family.approval_obligation) def test_part_snapshot_survives_master_rename(self): row=self.submit(self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])) original=row.lines.get().part_name parts.update_spare_part(spare_part=self.part,name="Later label") self.assertEqual(row.lines.get().part_name,original) def test_part_deactivation_before_submission_rejects(self): row=self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)]) parts.deactivate_spare_part(spare_part=self.part) with self.assertRaises(ValidationError): self.submit(row) def test_wrong_part_kind_rejected(self): with self.assertRaises(ValidationError): self.lines(self.create(),[self.line(spare_part=self.part)]) def test_empty_submission_rejected(self): with self.assertRaises(ValidationError): self.submit(self.create()) def test_stale_draft_and_line_edits_rejected(self): row=self.create() self.update(row) for operation in (self.update,self.lines,self.submit): with self.assertRaises(ValidationError): operation(row) def test_submitted_content_requires_revision(self): row=self.submit(self.lines(self.create())) with self.assertRaises(ValidationError): self.update(row) with self.assertRaises(ValidationError): self.lines(row) newer=self.revise(row) row.refresh_from_db() self.assertEqual((row.status,row.is_current,newer.revision),("SUPERSEDED",False,2)) def test_approval_records_exact_revision_actor_and_evidence(self): row=self.decide(self.submit(self.lines(self.create()))) self.assertEqual(row.decision.recorded_by,self.actor) self.assertEqual(row.decision.channel,"PHONE") self.assertEqual(row.status,"APPROVED") def test_rejection_and_revision_preserve_decision(self): row=self.decide(self.submit(self.lines(self.create())),outcome="REJECTED",note="Customer declined") new=self.revise(row) row.refresh_from_db() self.assertEqual(row.status,"REJECTED") self.assertEqual(row.decision.note,"Customer declined") self.assertEqual(new.status,"DRAFT") def test_revision_keeps_approved_history_and_obligation(self): old=self.decide(self.submit(self.lines(self.create()))) new=self.lines(self.revise(old),[]) old.refresh_from_db() self.assertEqual(old.status,"APPROVED") self.assertFalse(old.is_current) self.assertTrue(new.family.approval_obligation) self.assertEqual(new.customer_pay_total,0) def test_revision_currency_cannot_change(self): new=self.revise(self.submit(self.lines(self.create()))) with self.assertRaises(ValidationError): self.update(new,currency="USD") def test_expired_submission_and_decision_rejected(self): from unittest.mock import patch row=self.lines(self.create(valid_until=timezone.now()+timedelta(minutes=1))) with patch("apps.commercial.services.timezone.now",return_value=timezone.now()+timedelta(hours=1)): with self.assertRaises(ValidationError): self.submit(row) row=self.submit(row) with patch("apps.commercial.services.timezone.now",return_value=timezone.now()+timedelta(hours=1)): with self.assertRaises(ValidationError): self.decide(row) def test_quotation_operations_never_touch_inventory(self): before=[model.objects.count() for model in (StockMovement,StockReservation,PartsDisposition,SerializedStockUnit)] row=self.decide(self.submit(self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)]))) self.revise(row) self.assertEqual(before,[model.objects.count() for model in (StockMovement,StockReservation,PartsDisposition,SerializedStockUnit)]) def test_warranty_snapshot_not_modified(self): before=self.case.warranty_snapshot.__dict__.copy() self.submit(self.lines(self.create(),[self.line(responsibility="WARRANTY",coverage_reason="Explicit review")])) self.case.warranty_snapshot.refresh_from_db() for field in self.case.warranty_snapshot.immutable_fields: self.assertEqual(before[field],getattr(self.case.warranty_snapshot,field)) def test_staff_and_direct_permissions_do_not_grant_business_scope(self): from django.contrib.auth.models import Permission self.user.is_staff=True self.user.save() self.user.user_permissions.set(Permission.objects.filter(content_type__app_label="commercial")) with self.assertRaises(PermissionDenied): self.create(actor=self.user) self.assertFalse(q.service_quotations(actor=self.user).exists()) def test_permission_scoped_actor_can_manage(self): from apps.inventory.tests import grant from django.contrib.auth.models import Permission from apps.access.services import set_role_permissions 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(actor=self.user) self.assertEqual(q.current_service_quotation(actor=self.user,service_case=self.case),row) def test_inactive_actor_rejected(self): self.actor.is_active=False self.actor.save() with self.assertRaises(ValidationError): self.create() def test_cancelled_case_rejects_quotation(self): from apps.service.services import cancel_service_case cancel_service_case(service_case=self.case,reason="Synthetic cancellation",cancelled_by=self.actor) with self.assertRaises(ValidationError): self.create() def test_submitted_database_edits_and_deletes_rejected(self): row=self.submit(self.lines(self.create())) for change in ({"customer_note":"tampered"},{"status":"DRAFT"},{"grand_total":0}): with self.assertRaises(IntegrityError),transaction.atomic(): ServiceQuotation.objects.filter(pk=row.pk).update(**change) with self.assertRaises(IntegrityError),transaction.atomic(): row.lines.update(description="tampered") with self.assertRaises(ValidationError): row.delete() with self.assertRaises(ValidationError): row.lines.all().delete() def test_decision_immutable(self): row=self.decide(self.submit(self.lines(self.create()))) with self.assertRaises(IntegrityError),transaction.atomic(): QuotationDecision.objects.filter(quotation=row).update(reference="tampered") def test_database_totals_reconcile(self): row=self.lines(self.create()) with self.assertRaises(IntegrityError),transaction.atomic(): ServiceQuotation.objects.filter(pk=row.pk).update(subtotal=200,grand_total=200,customer_pay_total=200) with connection.cursor() as c: c.execute("SET CONSTRAINTS ALL IMMEDIATE") def test_submission_rollback(self): row=self.lines(self.create()) original=ServiceQuotation._persist def fail(record): original(record) if record.status=="SUBMITTED": raise RuntimeError("Injected after write") with patch.object(ServiceQuotation,"_persist",fail),self.assertRaises(RuntimeError): self.submit(row) row.refresh_from_db() self.assertEqual(row.status,"DRAFT") self.assertEqual(row.context_snapshot,{}) def test_approval_rollback(self): row=self.submit(self.lines(self.create())) with patch.object(ServiceQuotation,"_persist",side_effect=RuntimeError("Injected")),self.assertRaises(RuntimeError): self.decide(row) self.assertFalse(QuotationDecision.objects.exists()) row.refresh_from_db() self.assertEqual(row.status,"SUBMITTED") def test_query_budgets_and_queues(self): row=self.submit(self.lines(self.create(),[self.line(kind="PART",spare_part=self.part)])) with self.assertNumQueries(1): self.assertEqual([str(x) for x in q.service_quotations(actor=self.actor)],[str(row)]) with self.assertNumQueries(2): detail=q.quotation_detail(actor=self.actor,quotation=row) self.assertEqual([line.part_code for line in detail.lines.all()],[self.part.part_code]) with self.assertNumQueries(1): self.assertEqual(list(q.quotation_history(actor=self.actor,service_case=self.case)),[row]) with self.assertNumQueries(1): self.assertEqual(list(q.service_cases_awaiting_customer_approval(actor=self.actor)),[self.case]) with self.assertNumQueries(1): self.assertEqual([str(line.spare_part) for line in q.quotation_lines(actor=self.actor,quotation=row)],[str(self.part)]) self.assertFalse(q.service_cases_awaiting_quotation(actor=self.actor).exists()) class GatingTests(QuotationFixture,TestCase): @classmethod def setUpTestData(cls): setup_usage(cls) def test_quotation_free_work_remains_unclassified(self): execution,action=repair.prepared(self) repair.complete(self,execution) self.assertFalse(QuotationFamily.objects.exists()) self.assertFalse(CommercialWorkAuthorization.objects.exists()) def test_planning_allowed_customer_performance_blocked_until_approval(self): row=self.lines(self.create()) execution,action=repair.prepared(self,performed=False) with self.assertRaises(ValidationError): repair.perform(self,action) row=self.decide(self.submit(row)) repair.perform(self,action) repair.complete(self,execution) self.assertEqual(list(row.work_authorizations.values_list("operation",flat=True)),["PERFORM","COMPLETE"]) def test_replacement_draft_cannot_clear_obligation(self): execution,action=repair.prepared(self,performed=False) row=self.decide(self.submit(self.lines(self.create()))) new=self.lines(self.revise(row),[self.line(responsibility="COMPANY",coverage_reason="Explicit covered revision")]) with self.assertRaises(ValidationError): repair.perform(self,action) new=self.decide(self.submit(new)) repair.perform(self,action) self.assertEqual(new.work_authorizations.get().repair_action,action) def test_new_action_not_implicitly_approved(self): execution,action=repair.prepared(self,performed=False) self.decide(self.submit(self.lines(self.create()))) second=repair.add(self,execution,repair_action=self.action_type2) with self.assertRaises(ValidationError): repair.perform(self,second) repair.perform(self,action) def test_unapproved_replacement_blocks_completion_after_old_authorized_performance(self): execution,action=repair.prepared(self,performed=False) row=self.decide(self.submit(self.lines(self.create()))) repair.perform(self,action) self.revise(row) with self.assertRaises(ValidationError): repair.complete(self,execution) self.assertEqual(row.work_authorizations.get().operation,"PERFORM") def test_late_customer_classification_cannot_rewrite_unclassified_performance(self): row=self.create() repair.prepared(self) with self.assertRaises(ValidationError): self.lines(row) row.family.refresh_from_db() self.assertFalse(row.family.approval_obligation) def test_consumption_requires_current_approved_part_quantity(self): issue=self.prepared_issue(quantity=2) execution,action=repair.prepared(self,performed=False) row=self.decide(self.submit(self.lines(self.create(),[self.line(kind="PART",spare_part=self.part,quantity=1)]))) repair.perform(self,action) self.consume(issue,action) with self.assertRaises(ValidationError): self.consume(issue,action) self.assertEqual(PartsDisposition.objects.filter(kind="CONSUMED").count(),1) self.assertEqual(row.work_authorizations.filter(operation="CONSUME").count(),1) def test_revision_blocks_consumption_and_authorization_rolls_back_on_failure(self): issue=self.prepared_issue(quantity=1) execution,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.revise(row) with self.assertRaises(ValidationError): self.consume(issue,action) self.assertFalse(PartsDisposition.objects.exists())