"""Cross-domain UAT: real services, non-superuser personas, synthetic data only."""
from datetime import timedelta, time
from uuid import uuid4
import os
import re
from pathlib import Path
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.core.exceptions import ValidationError, PermissionDenied
from django.test import TransactionTestCase
from django.test import Client
from django.urls import reverse
from django.utils import timezone
from apps.service import tests as intake_fixture
from apps.catalog.tests import make_catalog
from apps.devices.tests import policy
from apps.operations.tests import grant
from apps.devices import services as devices
from apps.customers.services import update_customer
from apps.frontdesk import services as desk
from apps.frontdesk.models import QueueEntry
from apps.service import services as intake, engineer_services as assignments
from apps.service import diagnostic_services as diagnosis, repair_services as repair
from apps.service import quality_control_services as qc, handover_services as handover
from apps.service.handover_queries import device_identity_fingerprint
from apps.service.quality_control_checklist import CHECKLIST
from apps.service.models import ServiceCase
from apps.service_catalog.models import FaultDiagnosis, RepairAction
from apps.parts import services as parts
from apps.inventory import services as stock, request_services as requests, usage_services as usage, queries as inventory
from apps.inventory.models import StockLedgerEntry, PartsDisposition
from apps.commercial import services as quotes, invoice_services as invoices, payment_services as payments, payment_queries
from apps.sla import services as sla, queries as sla_queries
from apps.communications import hooks, services as communications
from apps.communications.models import Notification
from apps.reporting import service_dashboard, commercial_analytics, inventory_analytics, service_analytics


class JourneyTests(TransactionTestCase):
    def persona(self, name, permissions, *, company_scope=False):
        user = get_user_model().objects.create_user(username="uat-"+name)
        grant(user, self.company, permissions, None if company_scope else self.center)
        return user

    def setUp(self):
        intake_fixture.setup(self)
        self.customer = update_customer(customer=self.customer, primary_email="uat@example.invalid")
        devices.assign_device_owner(device=self.device, customer=self.customer)
        intake_fixture.warranty(self)
        self.front = self.persona("front", ["service.view_servicecase", "service.add_servicecase",
            "service.change_servicecase", "service.handover_servicecase", "service.close_servicecase",
            "frontdesk.view_appointment", "frontdesk.create_appointment", "frontdesk.check_in_appointment",
            "frontdesk.manage_queue", "frontdesk.view_queue", "devices.view_device", "devices.add_device",
            "devices.add_customerdevicerelationship", "customers.view_customer"], company_scope=True)
        self.engineer = self.persona("engineer", ["service.view_servicecase", "service.handle_servicecase",
            "inventory.request_parts", "inventory.consume_parts", "inventory.view_stock"])
        self.engineer2 = self.persona("engineer2", ["service.view_servicecase", "service.handle_servicecase"])
        self.inspector = self.persona("qc", ["service.view_servicecase", "service.perform_quality_control"])
        self.warehouse = self.persona("warehouse", ["service.view_servicecase", "inventory.view_stock",
            "inventory.manage_inventory", "inventory.receive_stock", "inventory.approve_parts",
            "inventory.reserve_parts", "inventory.issue_parts", "inventory.return_parts"])
        self.cashier = self.persona("cashier", ["service.view_servicecase", "commercial.view_servicequotation",
            "commercial.manage_servicequotation", "commercial.record_quotationdecision", "commercial.view_serviceinvoice",
            "commercial.manage_serviceinvoice", "commercial.finalize_serviceinvoice",
            "commercial.view_servicepayment", "commercial.receive_servicepayment"])
        self.finance = self.persona("finance", ["service.view_servicecase", "commercial.view_servicequotation",
            "commercial.view_serviceinvoice", "commercial.view_servicepayment", "commercial.receive_servicepayment",
            "commercial.reverse_servicepayment", "commercial.authorize_due_release"])
        report_permissions = ["reporting."+p.codename for p in Permission.objects.filter(content_type__app_label="reporting")]
        self.supervisor = self.persona("supervisor", report_permissions+["service.view_servicecase", "sla.view_sla",
            "sla.monitor_sla", "sla.manage_slapolicy", "communications.manage_templates", "communications.send_notification",
            "communications.view_notification", "commercial.view_servicequotation", "commercial.view_servicepayment"], company_scope=True)
        self.admin = get_user_model().objects.create_superuser(username="uat-config")
        _,_,self.model2,_ = make_catalog("UATALT")
        policy(self.model2)
        self.fault = FaultDiagnosis.objects.create(code="UAT", name="Power connector fault", applies_to_all_product_categories=True)
        self.action_type = RepairAction.objects.create(code="UAT", name="Connector repair", applies_to_all_product_categories=True)
        category = parts.create_part_category(code="UAT", name="Connectors")
        self.part = parts.create_spare_part(category=category, part_code="UAT", name="Connector", serialization_policy="NOT_SERIALIZED")
        parts.set_spare_part_compatibility(spare_part=self.part, product_models=[self.model], product_variants=[])
        self.location = stock.create_location(actor=self.warehouse, company=self.company, service_center=self.center,
            code="UAT", name="UAT store", location_type="STORE")
        today = timezone.localdate()
        sla.save_policy(actor=self.supervisor, company=self.company, code="uat", name="UAT elapsed",
            effective_from=today-timedelta(days=1), target_minutes=60, warning_minutes=10)
        self.template = communications.save_template(actor=self.supervisor, company=self.company, code="uat", name="UAT",
            channel="EMAIL", subject="Service update", body="Case {service_case_reference} update for {customer_name}")

    def arrive(self, appointment=True):
        if appointment:
            slot = desk.create_slot(actor=self.admin, service_center=self.center, date=timezone.localdate(),
                start_time=time(0), end_time=time(23,59,59), capacity=2)
            booking = desk.book_appointment(actor=self.front, slot=slot, customer=self.customer, device=self.device)
            queue = desk.check_in(actor=self.front, appointment=booking)
        else:
            queue = desk.add_walk_in(actor=self.front, service_center=self.center, customer=self.customer, device=self.device)
        desk.transition_queue(actor=self.front, queue_entry=queue, target="CALLED")
        desk.transition_queue(actor=self.front, queue_entry=queue, target="SERVING")
        self.client.force_login(self.front)
        url = reverse("frontdesk:record-operation", args=[self.center.pk,"intake",queue.pk])
        self.assertEqual(self.client.post(url, {"device":self.device.pk}).status_code,302)
        queue.refresh_from_db()
        self.case = queue.service_case
        page=self.client.get(reverse("frontdesk:desk",args=[self.center.pk]))
        self.assertContains(page,reverse("operations:detail",args=["cases",self.case.pk]))
        self.capture(page,"front-desk")
        self.assertEqual(desk.start_intake(actor=self.front, queue_entry=queue).pk, self.case.pk)
        self.assertTrue(self.case.warranty_snapshot.recorded_coverage)
        self.complaint = intake.add_service_case_complaint(service_case=self.case, complaint_symptom=self.symptom)
        self.sla = sla.enroll(actor=self.supervisor, service_case=self.case)
        self.assertEqual(self.sla.started_at,self.case.received_at)
        note = hooks.workflow_notification(actor=self.supervisor, template=self.template, event="SERVICE_INTAKE", record=self.case)
        self.assertEqual(hooks.workflow_notification(actor=self.supervisor, template=self.template, event="SERVICE_INTAKE", record=self.case).pk,note.pk)
        # No provider is configured: failure must leave intake intact.
        self.assertEqual(communications.send_notification(actor=self.supervisor, notification=note).status,"FAILED")
        self.case.refresh_from_db()
        self.assertEqual(self.case.status,"RECEIVED")
        desk.transition_queue(actor=self.front, queue_entry=queue, target="COMPLETED")
        return queue

    def ui(self,actor,route,action,data=None):
        self.client.force_login(actor)
        url=reverse("operations:"+route,args=[self.case.pk,action])
        detail=self.client.get(reverse("operations:detail",args=["cases",self.case.pk]))
        self.assertContains(detail,url)
        self.capture(detail,"job-card")
        page=self.client.get(url)
        self.assertEqual(page.status_code,200)
        self.capture(page,route+"-"+action)
        form=page.context["form"]
        values=dict(form.initial)
        if "revision" in page.context: values["revision"]=page.context["revision"]
        values.update(data or {})
        response=self.client.post(url,values)
        self.assertEqual(response.status_code,302, response.content.decode()[:6000])
        self.case.refresh_from_db()
        return response

    def capture(self,response,name):
        folder=os.environ.get("CCARE_UAT_CAPTURE_DIR")
        if folder:
            from django.conf import settings
            target=Path(folder)
            target.mkdir(parents=True,exist_ok=True)
            content=response.content.decode()
            for asset,tag in (("workspace.css","style"),("workspace.js","script")):
                text=(settings.BASE_DIR/"apps/operations/static/operations"/asset).read_text(encoding="utf-8")
                source='<link rel="stylesheet" href="/static/operations/workspace.css">' if tag=="style" else '<script src="/static/operations/workspace.js" defer></script>'
                content=content.replace(source,f"<{tag}>{text}</{tag}>" if tag=="style" else f'<script>document.addEventListener("DOMContentLoaded",function(){{{text}}});</script>')
            (target/(name+".html")).write_text(content,encoding="utf-8")

    def test_operational_technical_and_payment_forms(self):
        self.arrive(False)
        self.ui(self.front,"workflow","assignment",dict(operation="assign",engineer=self.engineer.pk))
        self.ui(self.engineer,"workflow","diagnosis",dict(operation="begin"))
        self.ui(self.engineer,"workflow","diagnosis",dict(operation="add",fault_diagnosis=self.fault.pk))
        self.ui(self.engineer,"workflow","diagnosis",dict(operation="complete"))
        self.ui(self.engineer,"workflow","repair",dict(operation="begin"))
        self.ui(self.engineer,"workflow","repair",dict(operation="add",repair_action=self.action_type.pk))
        from apps.service.models import ServiceRepairExecution, ServiceRepairAction
        self.execution=ServiceRepairExecution.objects.get(service_case=self.case)
        self.action=ServiceRepairAction.objects.get(repair_execution=self.execution)
        self.ui(self.cashier,"commercial_workflow","quote-create",dict(currency="BDT"))
        self.ui(self.cashier,"commercial_workflow","quote-lines",{
            "lines-TOTAL_FORMS":"1","lines-INITIAL_FORMS":"0","lines-0-kind":"PART","lines-0-responsibility":"CUSTOMER",
            "lines-0-description":"UAT connector","lines-0-quantity":"2","lines-0-unit_price":"100",
            "lines-0-discount":"0","lines-0-tax_rate":"0","lines-0-spare_part":self.part.pk})
        self.ui(self.cashier,"commercial_workflow","quote-submit")
        self.ui(self.cashier,"commercial_workflow","quote-decision",dict(outcome="APPROVED",channel="IN_PERSON",recipient_name="Synthetic",reference="UAT"))
        self.ui(self.warehouse,"parts_workflow","receive",dict(spare_part=self.part.pk,location=self.location.pk,quantity=2,reference="UAT UI receipt"))
        self.ui(self.engineer,"parts_workflow","request",dict(spare_part=self.part.pk,quantity=2))
        from apps.inventory.models import PartsRequest,StockReservation,PartsIssue
        request=PartsRequest.objects.get(service_case=self.case)
        self.ui(self.warehouse,"parts_workflow","approve",dict(request=request.pk))
        self.ui(self.warehouse,"parts_workflow","reserve",dict(line=request.lines.get().pk,location=self.location.pk,quantity=2))
        reservation=StockReservation.objects.get(service_case=self.case)
        self.ui(self.warehouse,"parts_workflow","issue",dict(reservation=reservation.pk))
        issue=PartsIssue.objects.get(service_case=self.case)
        self.ui(self.engineer,"workflow","repair",dict(operation="perform",action=self.action.pk))
        self.ui(self.engineer,"parts_workflow","consume",dict(issue=issue.pk,repair_action=self.action.pk,quantity=1))
        self.ui(self.warehouse,"parts_workflow","return",dict(issue=issue.pk,location=self.location.pk,quantity=1))
        self.ui(self.engineer,"workflow","repair",dict(operation="complete",outcome="REPAIRED"))
        self.ui(self.inspector,"workflow","qc",dict(operation="submit"))
        self.ui(self.inspector,"workflow","qc",dict(operation="begin"))
        from apps.service import quality_control_queries as qq
        quality=qq.current_quality_control(self.case)
        for check in qq.quality_control_checks(quality):
            self.ui(self.inspector,"workflow","qc",dict(operation="check",check=check.pk,result="PASS"))
        complaint_check=qq.quality_control_complaint_checks(quality).get()
        self.ui(self.inspector,"workflow","qc",dict(operation="complaint",complaint_check=complaint_check.pk,complaint_result="RESOLVED"))
        self.ui(self.inspector,"workflow","qc",dict(operation="pass"))
        self.ui(self.front,"workflow","delivery",dict(operation="ready",device_confirmed="on"))
        self.ui(self.cashier,"commercial_workflow","invoice-prepare")
        self.ui(self.cashier,"commercial_workflow","invoice-finalize")
        self.ui(self.cashier,"commercial_workflow","payment-receive",dict(amount="100",method="CASH"))
        self.ui(self.front,"workflow","delivery",dict(operation="handover",recipient_type="CUSTOMER",verification_method="CUSTOMER_RECORD",
            device_confirmed="on",verification_confirmed="on",physical_handover_confirmed="on"))
        self.ui(self.front,"workflow","delivery",dict(operation="close"))
        self.assertEqual(self.case.status,"CLOSED")
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.location,spare_part=self.part),1)
        self.reconcile()

    def reconcile(self):
        """Cross-domain reconciliation after the closed operational journey."""
        from apps.commercial import invoice_queries
        from apps.commercial import queries as quotation_queries
        from apps.commercial.models import PaymentAllocation, PaymentReversal, QuotationDecision, ServicePayment, ServicePaymentReceipt
        from apps.inventory.models import PartsIssue
        # Inventory: one receipt, one issue into custody, one consumption, one
        # return. Entries are grouped per movement and sorted, because two
        # movements can share a posted_at and the movement id is a UUID.
        movements={}
        for entry in StockLedgerEntry.objects.filter(spare_part=self.part).select_related("movement","location"):
            movements.setdefault(entry.movement_id,[]).append(
                (entry.movement.kind, entry.quantity_delta, entry.location.location_type))
        self.assertEqual(sorted(tuple(sorted(group)) for group in movements.values()),sorted([
            (("RECEIPT",2,"STORE"),),
            (("MOVE",-2,"STORE"),("MOVE",2,"CUSTODY")),
            (("CONSUME",-1,"CUSTODY"),),
            (("MOVE",-1,"CUSTODY"),("MOVE",1,"STORE"))]))
        issue=PartsIssue.objects.get(service_case=self.case)
        self.assertEqual((issue.custody_location.company_id,self.location.company_id),(self.company.pk,self.company.pk))
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.location,spare_part=self.part)
            +inventory.stock_on_hand(actor=self.warehouse,location=issue.custody_location,spare_part=self.part),1)
        self.assertEqual(PartsDisposition.objects.filter(issue=issue).count(),2)
        # Commercial: one approved quotation, one decision, one settled finalized invoice.
        quotation=quotation_queries.service_quotations(actor=self.cashier).filter(family__service_case=self.case,is_current=True).get()
        self.assertEqual((quotation.status,quotation.customer_pay_total),("APPROVED",200))
        self.assertEqual(QuotationDecision.objects.count(),1)
        invoice=invoice_queries.finalized_service_invoice(actor=self.cashier,service_case=self.case)
        self.assertEqual((invoice.status,invoice.customer_pay_total,invoice.currency),("FINALIZED",100,quotation.currency))
        settlement=payment_queries.invoice_settlement_summary(actor=self.cashier,invoice=invoice)
        self.assertEqual((settlement["balance"],settlement["financially_clear"]),(0,True))
        self.assertEqual(PaymentAllocation.objects.count(),1)
        self.assertEqual((ServicePayment.objects.get().status,ServicePaymentReceipt.objects.count()),("POSTED",1))
        self.assertFalse(PaymentReversal.objects.exists())
        self.assertEqual(commercial_analytics.settlements(self.supervisor,{}).get().balance_due,0)
        # SLA and communications: stopped at the authoritative delivery release.
        observation=sla_queries.observation(actor=self.supervisor,sla=self.sla)
        self.assertEqual((observation["state"],observation["completed_at"]),("COMPLETED_ON_TIME",self.case.delivery_release.readied_at))
        self.assertEqual(Notification.objects.count(),1)
        # Reporting: no open work, one consumption/return pair, one complaint dimension.
        self.assertEqual(service_dashboard.dashboard(self.supervisor,{})["current_open"],0)
        usage=next(t for t in inventory_analytics.inventory_tables(self.supervisor,{}) if t.key=="usage_parts").rows.get()
        self.assertEqual((usage["consumed"],usage["returned_unused"]),(1,1))
        self.assertEqual(service_analytics.complaint_frequency(self.supervisor,{},dimension="product_category").get()["dimension_id"],self.category.pk)

    def diagnose(self):
        assignments.assign_service_case_engineer(service_case=self.case, engineer=self.engineer, assigned_by=self.front)
        self.assessment = diagnosis.begin_service_case_diagnosis(service_case=self.case, actor=self.engineer)
        diagnosis.add_diagnostic_finding(assessment=self.assessment, actor=self.engineer, fault_diagnosis=self.fault)
        self.assessment.refresh_from_db()
        diagnosis.complete_service_case_diagnosis(assessment=self.assessment, actor=self.engineer)
        self.case.refresh_from_db()

    def plan(self, responsibility="CUSTOMER", decision="APPROVED"):
        self.execution = repair.begin_service_case_repair(service_case=self.case, actor=self.engineer)
        self.action = repair.add_repair_action(repair_execution=self.execution, actor=self.engineer, repair_action=self.action_type)
        row = quotes.create_service_quotation(actor=self.cashier, service_case=self.case, currency="BDT")
        row = quotes.set_quotation_lines(actor=self.cashier, quotation=row, expected_revision=quotes.revision(row), lines=[dict(
            kind="PART", responsibility=responsibility, coverage_reason="Explicit reviewed coverage", spare_part=self.part,
            description="Connector", quantity=2, unit_price="100.00")])
        row = quotes.submit_service_quotation(actor=self.cashier, quotation=row, expected_revision=quotes.revision(row))
        if responsibility == "CUSTOMER":
            with self.assertRaises(ValidationError):
                repair.mark_repair_action_performed(action=self.action,actor=self.engineer)
        self.quote = quotes.record_quotation_decision(actor=self.cashier, quotation=row, expected_revision=quotes.revision(row),
            outcome=decision, channel="IN_PERSON", recipient_name="Synthetic customer", reference="UAT-confirmation")

    def use_parts(self):
        stock.receive_stock(actor=self.warehouse, destination=self.location, spare_part=self.part, quantity=2,
            reference="UAT receipt", idempotency_key=uuid4())
        request = requests.create_parts_request(actor=self.engineer, service_case=self.case, lines=[dict(spare_part=self.part,quantity=2)])
        request = requests.approve_parts_request(actor=self.warehouse,request=request,expected_revision=request.updated_at.isoformat())
        reservation = requests.reserve_parts(actor=self.warehouse,line=request.lines.get(),location=self.location,quantity=2,
            expected_request_revision=request.updated_at.isoformat())
        self.issue = usage.issue_reserved_parts(actor=self.warehouse,reservation=reservation,expected_revision=reservation.updated_at.isoformat())
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.location,spare_part=self.part),0)
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.issue.custody_location,spare_part=self.part),2)
        repair.mark_repair_action_performed(action=self.action,actor=self.engineer)
        command = uuid4()
        self.consumption = usage.consume_issued_parts(actor=self.engineer,issue=self.issue,repair_action=self.action,quantity=1,command_key=command)
        with self.assertRaises(ValidationError):
            usage.consume_issued_parts(actor=self.engineer,issue=self.issue,repair_action=self.action,quantity=1,command_key=command)
        usage.return_unused_parts(actor=self.warehouse,issue=self.issue,destination=self.location,quantity=1,command_key=uuid4())
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.location,spare_part=self.part),1)
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.issue.custody_location,spare_part=self.part),0)

    def quality(self, fail=False):
        self.execution.refresh_from_db()
        repair.complete_service_case_repair(repair_execution=self.execution,actor=self.engineer,outcome="REPAIRED")
        self.case.refresh_from_db()
        qc.submit_for_quality_control(service_case=self.case,inspector=self.inspector)
        self.case.refresh_from_db()
        row = qc.begin_quality_control(service_case=self.case,inspector=self.inspector)
        for code,_,_ in CHECKLIST:
            row.refresh_from_db()
            qc.set_quality_control_check(quality_control=row,inspector=self.inspector,check_code=code,result="FAIL" if fail and code=="BOOT" else "PASS")
        row.refresh_from_db()
        qc.set_quality_control_complaint_check(quality_control=row,inspector=self.inspector,complaint=self.complaint,result="RESOLVED")
        row.refresh_from_db()
        if fail:
            qc.complete_quality_control_fail(quality_control=row,inspector=self.inspector,failure_note="Boot failed UAT")
        else:
            qc.complete_quality_control_pass(quality_control=row,inspector=self.inspector)
        self.case.refresh_from_db()

    def release(self):
        handover.mark_service_case_ready_for_delivery(service_case=self.case,actor=self.front,expected_device=self.device,
            expected_device_identity=device_identity_fingerprint(self.device))
        self.case.refresh_from_db()

    def deliver(self):
        result=handover.handover_service_case(service_case=self.case,actor=self.front,expected_device=self.device,
            expected_device_identity=device_identity_fingerprint(self.device),recipient_type="CUSTOMER",verification_method="CUSTOMER_RECORD",accessories=[])
        self.case.refresh_from_db()
        return result

    def test_appointment_parts_paid_closed_reconciles(self):
        self.arrive(); self.diagnose(); self.plan(); self.use_parts(); self.quality(); self.release()
        row=invoices.prepare_service_invoice(actor=self.cashier,service_case=self.case)
        self.invoice=invoices.finalize_service_invoice(actor=self.cashier,invoice=row,expected_revision=invoices.revision(row))
        self.assertEqual(self.invoice.customer_pay_total,100)
        with self.assertRaises(ValidationError): self.deliver()
        summary=payment_queries.invoice_settlement_summary(actor=self.cashier,invoice=self.invoice)
        payment=payments.receive_service_payment(actor=self.cashier,invoice=self.invoice,expected_revision=summary["revision"],amount="100",method="CASH")
        self.assertEqual(payment.allocation.amount,100)
        self.assertTrue(payment.receipt.number)
        self.deliver()
        handover.close_service_case(service_case=self.case,actor=self.front)
        self.case.refresh_from_db()
        self.assertEqual(self.case.status,"CLOSED")
        self.assertEqual(payment_queries.invoice_settlement_summary(actor=self.cashier,invoice=self.invoice)["balance"],0)
        self.assertEqual(service_dashboard.dashboard(self.supervisor,{})["current_open"],0)
        debt=commercial_analytics.settlements(self.supervisor,{}).get()
        self.assertEqual(debt.balance_due,0)
        report=next(t for t in inventory_analytics.inventory_tables(self.supervisor,{}) if t.key=="usage_parts")
        self.assertEqual((report.rows.get()["consumed"],report.rows.get()["returned_unused"]),(1,1))
        self.assertEqual(Notification.objects.count(),1)
        self.assertEqual(PartsDisposition.objects.count(),2)
        observation=sla_queries.observation(actor=self.supervisor,sla=self.sla)
        self.assertEqual(observation["completed_at"],self.case.delivery_release.readied_at)
        self.assertEqual(observation["state"],"COMPLETED_ON_TIME")
        complaint=service_analytics.complaint_frequency(self.supervisor,{},dimension="product_category").get()
        self.assertEqual(complaint["complaint_symptom_id"],self.symptom.pk)
        self.assertEqual(complaint["dimension_id"],self.category.pk)
        cooccurrence=service_analytics.repair_action_diagnosis_cooccurrence(self.supervisor,{}).get()
        self.assertEqual(cooccurrence["root_cause_label"],"Unknown / Unconfirmed")
        self.assertEqual(cooccurrence["performed_actions"],1)

    def test_walk_in_no_parts_qc_rework_and_invoice_free_closure(self):
        self.arrive(appointment=False); self.diagnose()
        self.execution=repair.begin_service_case_repair(service_case=self.case,actor=self.engineer)
        action=repair.add_repair_action(repair_execution=self.execution,actor=self.engineer,repair_action=self.action_type)
        repair.mark_repair_action_performed(action=action,actor=self.engineer)
        self.quality(fail=True)
        self.assertEqual(self.case.status,"DIAGNOSED")
        self.execution=repair.begin_service_case_repair(service_case=self.case,actor=self.engineer)
        action=repair.add_repair_action(repair_execution=self.execution,actor=self.engineer,repair_action=self.action_type)
        repair.mark_repair_action_performed(action=action,actor=self.engineer)
        self.quality(); self.release(); self.deliver()
        handover.close_service_case(service_case=self.case,actor=self.front)
        self.assertFalse(StockLedgerEntry.objects.exists())

    def test_rejected_customer_quote_cannot_perform_work(self):
        self.arrive(False); self.diagnose(); self.plan(decision="REJECTED")
        with self.assertRaises(ValidationError): repair.mark_repair_action_performed(action=self.action,actor=self.engineer)
        self.case.refresh_from_db()
        self.assertEqual(self.case.status,"REPAIRING")
        self.assertFalse(PartsDisposition.objects.exists())

    def test_warranty_responsibility_needs_no_customer_payment(self):
        self.arrive(False); self.diagnose(); self.plan(responsibility="WARRANTY"); self.use_parts(); self.quality(); self.release()
        row=invoices.prepare_service_invoice(actor=self.cashier,service_case=self.case)
        row=invoices.finalize_service_invoice(actor=self.cashier,invoice=row,expected_revision=invoices.revision(row))
        self.assertEqual((row.customer_pay_total,row.warranty_covered_total),(0,100))
        self.deliver()

    def test_scope_and_persona_denials(self):
        self.arrive(False)
        self.client.force_login(self.engineer)
        self.assertEqual(self.client.get(reverse("operations:detail",args=["cases",self.case.pk])).status_code,200)
        with self.assertRaises(PermissionDenied): quotes.create_service_quotation(actor=self.engineer,service_case=self.case,currency="BDT")
        foreign=intake_fixture.intake(self,company=self.other_company,service_center=self.other_center,customer=self.outsider,device=self.device2)
        for user in (self.front,self.engineer,self.inspector,self.cashier,self.warehouse):
            self.client.force_login(user)
            self.assertEqual(self.client.get(reverse("operations:detail",args=["cases",foreign.pk])).status_code,404)
            self.assertEqual(self.client.post(reverse("frontdesk:operation",args=[self.other_center.pk,"walk-in"]),{}).status_code,403)
        with self.assertRaises(PermissionDenied):
            quotes.create_service_quotation(actor=self.cashier,service_case=foreign,currency="BDT")

    def test_new_actions_csrf_idor_stale_get_and_persona_isolation(self):
        self.arrive(False)
        url=reverse("operations:workflow",args=[self.case.pk,"assignment"])
        self.client.force_login(self.front)
        before=self.client.get(url)
        self.assertEqual(self.case.engineer_assignments.count(),0)
        guarded=Client(enforce_csrf_checks=True)
        guarded.force_login(self.front)
        self.assertEqual(guarded.post(url,{"operation":"assign"}).status_code,403)
        self.assertEqual(self.client.put(url).status_code,405)
        token=before.context["form"].initial["revision"]
        self.ui(self.front,"workflow","assignment",dict(operation="assign",engineer=self.engineer.pk))
        self.assertEqual(self.client.post(url,dict(operation="reassign",engineer=self.engineer.pk,revision=token)).status_code,400)
        self.client.force_login(self.cashier)
        self.assertEqual(self.client.get(reverse("operations:workflow",args=[self.case.pk,"diagnosis"])).status_code,403)
        foreign=intake_fixture.intake(self,company=self.other_company,service_center=self.other_center,customer=self.outsider,device=self.device2)
        for route,operation in (("workflow","assignment"),("commercial_workflow","quote-create"),("parts_workflow","receive")):
            target=reverse("operations:"+route,args=[foreign.pk,operation])
            self.assertEqual(self.client.get(target).status_code,404)
            self.assertEqual(self.client.post(target,{}).status_code,404)
        self.client.logout()
        self.assertEqual(self.client.get(url).status_code,302)

    def test_sla_exact_due_overdue_then_cancellation(self):
        self.arrive(False)
        self.assertEqual(sla_queries.observation(actor=self.supervisor,sla=self.sla,at=self.sla.due_at)["state"],"DUE_SOON")
        at=self.sla.due_at+timedelta(microseconds=1)
        self.assertEqual(sla_queries.observation(actor=self.supervisor,sla=self.sla,at=at)["state"],"OVERDUE")
        with patch("django.utils.timezone.now",return_value=at):
            sla.record_escalation(actor=self.supervisor,sla=self.sla,at=at)
            intake.cancel_service_case(service_case=self.case,cancelled_by=self.front,reason="Customer withdrew before assignment")
            self.case.refresh_from_db()
            self.assertEqual(sla_queries.observation(actor=self.supervisor,sla=self.sla,at=at+timedelta(seconds=1))["state"],"CANCELLED")
            prior=self.sla.escalations.count()
            sla.monitor(actor=self.supervisor,case_id=self.case.pk)
            self.assertEqual(self.sla.escalations.count(),prior)

    def test_customer_owned_device_registration_is_scoped_and_atomic(self):
        from apps.catalog.identification import set_identification_policy
        from apps.devices.models import Device,CustomerDeviceRelationship
        set_identification_policy(product_model=self.model,imei1_requirement="NOT_APPLICABLE",imei2_requirement="NOT_APPLICABLE",serial_requirement="REQUIRED")
        self.client.force_login(self.front)
        url=reverse("operations:register_device",args=[self.customer.pk])
        self.assertContains(self.client.get(reverse("operations:detail",args=["customers",self.customer.pk])),url)
        self.capture(self.client.get(url),"device-registration")
        before=Device.objects.count()
        self.assertEqual(self.client.post(url,dict(product_model=self.model.pk,ownership_confirmed="on")).status_code,400)
        self.assertEqual(Device.objects.count(),before)
        response=self.client.post(url,dict(product_model=self.model.pk,serial="UAT-DEVICE-REGISTER-1",ownership_confirmed="on"))
        self.assertEqual(response.status_code,302)
        self.assertEqual(Device.objects.count(),before+1)
        self.assertTrue(CustomerDeviceRelationship.objects.filter(customer=self.customer,device__identifiers__value="UAT-DEVICE-REGISTER-1").exists())
        self.assertEqual(self.client.post(url,dict(product_model=self.model.pk,serial="UAT-DEVICE-REGISTER-1",ownership_confirmed="on")).status_code,400)
        self.assertEqual(Device.objects.count(),before+1)
        self.assertEqual(self.client.get(reverse("operations:register_device",args=[self.outsider.pk])).status_code,404)
        self.client.force_login(self.engineer)
        self.assertEqual(self.client.get(url).status_code,404)

    def test_region_center_and_department_scope_at_new_boundaries(self):
        self.arrive(False)
        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
        permissions=Permission.objects.filter(content_type__app_label="service",codename__in=["view_servicecase","change_servicecase"])
        role=Role.objects.create(code="UAT-SCOPES",name="Arbitrary scope test")
        set_role_permissions(role=role,permissions=permissions)
        department=Department.objects.create(company=self.company,code="UAT",name="UAT department")
        for kind,scope,expected in (("region",dict(region=self.region),200),("other-center",dict(region=self.region,service_center=self.center2),404),
                ("department",dict(department=department),404)):
            actor=get_user_model().objects.create_user(username="uat-scope-"+kind)
            assignment=create_assignment(user=actor,company=self.company,**scope)
            create_role_assignment(user=actor,role=role,organization_assignment=assignment)
            self.client.force_login(actor)
            self.assertEqual(self.client.get(reverse("operations:workflow",args=[self.case.pk,"assignment"])).status_code,expected)

    def labour_lines(self,description="UAT labour",unit_price="60"):
        return {"lines-TOTAL_FORMS":"1","lines-INITIAL_FORMS":"0","lines-0-kind":"LABOR","lines-0-responsibility":"CUSTOMER",
            "lines-0-description":description,"lines-0-quantity":"1","lines-0-unit_price":unit_price,
            "lines-0-discount":"0","lines-0-tax_rate":"0"}

    def test_engineer_reassignment_and_unassignment_through_operational_ui(self):
        self.arrive(False)
        self.ui(self.front,"workflow","assignment",dict(operation="assign",engineer=self.engineer.pk))
        self.ui(self.front,"workflow","assignment",dict(operation="reassign",engineer=self.engineer2.pk,reason="Original engineer on approved leave"))
        from apps.service import engineer_queries as eq
        self.assertEqual(eq.current_engineer_assignment(self.case).engineer_id,self.engineer2.pk)
        self.assertEqual(self.case.engineer_assignments.count(),2)
        self.ui(self.front,"workflow","assignment",dict(operation="unassign",reason="Reassigned to the field engineering team"))
        self.assertIsNone(eq.current_engineer_assignment(self.case))
        self.assertEqual(self.case.status,"RECEIVED")
        history=self.case.engineer_assignments.order_by("assigned_at","pk")
        self.assertEqual([row.engineer_id for row in history],[self.engineer.pk,self.engineer2.pk])
        self.assertEqual(history.last().end_reason,"Reassigned to the field engineering team")
        # The unassigned engineer can no longer act on the case.
        self.client.force_login(self.engineer)
        self.assertEqual(self.client.get(reverse("operations:workflow",args=[self.case.pk,"diagnosis"])).status_code,403)

    def test_qc_rework_loop_through_operational_ui(self):
        from apps.service import quality_control_queries as qq
        from apps.service.models import ServiceQualityControl, ServiceRepairExecution
        self.arrive(False)
        self.ui(self.front,"workflow","assignment",dict(operation="assign",engineer=self.engineer.pk))
        self.ui(self.engineer,"workflow","diagnosis",dict(operation="begin"))
        self.ui(self.engineer,"workflow","diagnosis",dict(operation="add",fault_diagnosis=self.fault.pk))
        self.ui(self.engineer,"workflow","diagnosis",dict(operation="complete"))
        self.ui(self.engineer,"workflow","repair",dict(operation="begin"))
        self.ui(self.engineer,"workflow","repair",dict(operation="add",repair_action=self.action_type.pk))
        self.ui(self.cashier,"commercial_workflow","quote-create",dict(currency="BDT"))
        self.ui(self.cashier,"commercial_workflow","quote-lines",self.labour_lines())
        self.ui(self.cashier,"commercial_workflow","quote-submit")
        self.ui(self.cashier,"commercial_workflow","quote-decision",dict(outcome="APPROVED",channel="IN_PERSON",recipient_name="Synthetic",reference="UAT rework approval"))
        from apps.service.models import ServiceRepairAction
        action=ServiceRepairAction.objects.get(repair_execution__service_case=self.case)
        self.ui(self.engineer,"workflow","repair",dict(operation="perform",action=action.pk))
        self.ui(self.engineer,"workflow","repair",dict(operation="complete",outcome="REPAIRED"))
        self.ui(self.inspector,"workflow","qc",dict(operation="submit"))
        self.ui(self.inspector,"workflow","qc",dict(operation="begin"))
        quality=qq.current_quality_control(self.case)
        for check in qq.quality_control_checks(quality):
            self.ui(self.inspector,"workflow","qc",dict(operation="check",check=check.pk,result="FAIL" if check.check_code=="BOOT" else "PASS"))
        complaint=qq.quality_control_complaint_checks(quality).get()
        self.ui(self.inspector,"workflow","qc",dict(operation="complaint",complaint_check=complaint.pk,complaint_result="RESOLVED"))
        self.ui(self.inspector,"workflow","qc",dict(operation="fail",reason="Device failed the boot check on the first pass"))
        self.assertEqual(self.case.status,"DIAGNOSED")
        # Rework reopens the technical stages. New performed scope is not covered
        # by the approved quotation, so commercial must re-approve it first.
        self.ui(self.engineer,"workflow","repair",dict(operation="begin"))
        self.ui(self.engineer,"workflow","repair",dict(operation="add",repair_action=self.action_type.pk))
        second=ServiceRepairAction.objects.get(repair_execution=ServiceRepairExecution.objects.filter(
            service_case=self.case).order_by("created_at","pk").last())
        blocked_url=reverse("operations:workflow",args=[self.case.pk,"repair"])
        self.client.force_login(self.engineer)
        page=self.client.get(blocked_url)
        values=dict(page.context["form"].initial)
        values.update(operation="perform",action=second.pk)
        refused=self.client.post(blocked_url,values)
        self.assertEqual(refused.status_code,400)
        self.assertContains(refused,"Repair scope changed",status_code=400)
        self.assertFalse(ServiceRepairAction.objects.get(pk=second.pk).performed_at)
        self.ui(self.cashier,"commercial_workflow","quote-revise")
        lines_url=reverse("operations:commercial_workflow",args=[self.case.pk,"quote-lines"])
        revised_page=self.client.get(lines_url)
        revised_values=dict(revised_page.context["form"].initial)
        revised_values["revision"]=revised_page.context["revision"]
        revised_values.update({"lines-TOTAL_FORMS":"2","lines-INITIAL_FORMS":"1",
            "lines-0-kind":"LABOR","lines-0-responsibility":"CUSTOMER","lines-0-description":"UAT rework labour",
            "lines-0-quantity":"1","lines-0-unit_price":"25","lines-0-discount":"0","lines-0-tax_rate":"0",
            "lines-0-repair_action":second.pk})
        revised_response=self.client.post(lines_url,revised_values)
        self.assertEqual(revised_response.status_code,302,revised_response.content.decode()[:6000])
        self.ui(self.cashier,"commercial_workflow","quote-submit")
        self.ui(self.cashier,"commercial_workflow","quote-decision",dict(outcome="APPROVED",channel="IN_PERSON",recipient_name="Synthetic",reference="UAT rework re-approval"))
        self.ui(self.engineer,"workflow","repair",dict(operation="perform",action=second.pk))
        self.ui(self.engineer,"workflow","repair",dict(operation="complete",outcome="REPAIRED"))
        self.ui(self.inspector,"workflow","qc",dict(operation="submit"))
        self.ui(self.inspector,"workflow","qc",dict(operation="begin"))
        quality=qq.current_quality_control(self.case)
        for check in qq.quality_control_checks(quality):
            self.ui(self.inspector,"workflow","qc",dict(operation="check",check=check.pk,result="PASS"))
        complaint=qq.quality_control_complaint_checks(quality).get()
        self.ui(self.inspector,"workflow","qc",dict(operation="complaint",complaint_check=complaint.pk,complaint_result="RESOLVED"))
        self.ui(self.inspector,"workflow","qc",dict(operation="pass"))
        self.assertEqual(self.case.status,"QC_PASSED")
        self.assertEqual(ServiceQualityControl.objects.filter(service_case=self.case).count(),2)
        self.assertEqual(ServiceQualityControl.objects.filter(service_case=self.case,outcome="FAILED").count(),1)
        self.assertEqual(ServiceRepairExecution.objects.filter(service_case=self.case).count(),2)

    def test_quotation_revision_after_rejection_reaches_an_approved_quote(self):
        from apps.commercial import invoice_queries
        from apps.commercial import queries as quotation_queries
        self.arrive(False); self.diagnose()
        self.execution=repair.begin_service_case_repair(service_case=self.case,actor=self.engineer)
        repair.add_repair_action(repair_execution=self.execution,actor=self.engineer,repair_action=self.action_type)
        self.ui(self.cashier,"commercial_workflow","quote-create",dict(currency="BDT"))
        self.ui(self.cashier,"commercial_workflow","quote-lines",self.labour_lines("UAT first proposal"))
        self.ui(self.cashier,"commercial_workflow","quote-submit")
        self.ui(self.cashier,"commercial_workflow","quote-decision",dict(outcome="REJECTED",channel="IN_PERSON",recipient_name="Synthetic",reference="UAT customer declined"))
        rejected=quotation_queries.service_quotations(actor=self.cashier).filter(family__service_case=self.case,is_current=True).get()
        self.assertEqual((rejected.revision,rejected.status),(1,"REJECTED"))
        self.ui(self.cashier,"commercial_workflow","quote-revise")
        revised=quotation_queries.service_quotations(actor=self.cashier).filter(family__service_case=self.case,is_current=True).get()
        self.assertEqual((revised.revision,revised.status),(2,"DRAFT"))
        # The revision screen must prefill the copied line, not an empty formset.
        url=reverse("operations:commercial_workflow",args=[self.case.pk,"quote-lines"])
        page=self.client.get(url)
        group=page.context["groups"][0][1]
        self.assertEqual((group.total_form_count(),group.forms[0].initial["description"]),(2,"UAT first proposal"))
        values=dict(page.context["form"].initial)
        values["revision"]=page.context["revision"]
        values.update(self.labour_lines("UAT revised proposal","45"))
        values.update({"lines-TOTAL_FORMS":"2","lines-INITIAL_FORMS":"1"})
        response=self.client.post(url,values)
        self.assertEqual(response.status_code,302,response.content.decode()[:6000])
        revised.refresh_from_db()
        self.assertEqual([l.description for l in revised.lines.filter(is_active=True)],["UAT revised proposal"])
        # A completed spare row adds a line; discount and tax stay explicit.
        page=self.client.get(url)
        added=dict(page.context["form"].initial)
        added["revision"]=page.context["revision"]
        added.update({"lines-TOTAL_FORMS":"2","lines-INITIAL_FORMS":"1",
            "lines-0-kind":"LABOR","lines-0-responsibility":"CUSTOMER","lines-0-description":"UAT revised proposal",
            "lines-0-quantity":"1","lines-0-unit_price":"45","lines-0-discount":"0","lines-0-tax_rate":"0",
            "lines-1-kind":"LABOR","lines-1-responsibility":"CUSTOMER","lines-1-description":"UAT second line",
            "lines-1-quantity":"1","lines-1-unit_price":"5","lines-1-discount":"0","lines-1-tax_rate":"0"})
        second_post=self.client.post(url,added)
        self.assertEqual(second_post.status_code,302,second_post.content.decode()[:4000])
        revised.refresh_from_db()
        self.assertEqual([(l.description,l.responsibility,str(l.total)) for l in revised.lines.filter(is_active=True)],
            [("UAT revised proposal","CUSTOMER","45.00"),("UAT second line","CUSTOMER","5.00")])
        self.assertEqual((revised.grand_total,revised.customer_pay_total),(50,50))
        self.ui(self.cashier,"commercial_workflow","quote-submit")
        self.ui(self.cashier,"commercial_workflow","quote-decision",dict(outcome="APPROVED",channel="IN_PERSON",recipient_name="Synthetic",reference="UAT customer approved"))
        self.ui(self.cashier,"commercial_workflow","invoice-prepare")
        draft=invoice_queries.current_service_invoice(actor=self.cashier,service_case=self.case)
        # Only consumption is prepared automatically; labour needs confirmation.
        self.assertEqual((draft.status,draft.lines.filter(is_active=True).count()),("DRAFT",0))
        active=list(revised.lines.filter(is_active=True))
        reconcile_url=reverse("operations:commercial_workflow",args=[self.case.pk,"invoice-reconcile"])
        page=self.client.get(reconcile_url)
        self.assertEqual({group.prefix:group.initial_form_count() for _,group in page.context["groups"]},
            {"allocations":0,"confirmations":0,"adjustments":0})
        values=dict(page.context["form"].initial)
        values["revision"]=page.context["revision"]
        values["note"]="UAT confirmed the performed labour"
        values.update({"allocations-TOTAL_FORMS":"0","allocations-INITIAL_FORMS":"0",
            "adjustments-TOTAL_FORMS":"0","adjustments-INITIAL_FORMS":"0",
            "confirmations-TOTAL_FORMS":str(len(active)),"confirmations-INITIAL_FORMS":"0"})
        for index,line in enumerate(active):
            values[f"confirmations-{index}-quotation_line"]=line.pk
            values[f"confirmations-{index}-quantity"]=line.quantity
            values[f"confirmations-{index}-reason"]="UAT confirmed performed labour"
        reconciled=self.client.post(reconcile_url,values)
        self.assertEqual(reconciled.status_code,302,reconciled.content.decode()[:6000])
        draft.refresh_from_db()
        self.assertEqual((draft.customer_pay_total,draft.lines.filter(is_active=True).count()),(50,2))

    def test_invoice_reconcile_prefills_and_preserves_prepared_allocation(self):
        from apps.commercial import invoice_queries
        self.arrive(False); self.diagnose(); self.plan(); self.use_parts(); self.quality(); self.release()
        self.ui(self.cashier,"commercial_workflow","invoice-prepare")
        draft=invoice_queries.current_service_invoice(actor=self.cashier,service_case=self.case)
        allocation=invoice_queries.invoice_allocations(actor=self.cashier,invoice=draft).get()
        self.assertEqual(allocation.mode,"AUTO")
        url=reverse("operations:commercial_workflow",args=[self.case.pk,"invoice-reconcile"])
        page=self.client.get(url)
        groups={group.prefix:group for _,group in page.context["groups"]}
        self.assertEqual(groups["allocations"].initial_form_count(),1)
        self.assertEqual(groups["confirmations"].initial_form_count(),0)
        values=dict(page.context["form"].initial)
        values["revision"]=page.context["revision"]
        values["note"]="UAT confirmed the prepared allocation"
        for prefix in ("confirmations","adjustments"):
            values[f"{prefix}-TOTAL_FORMS"]="0"
            values[f"{prefix}-INITIAL_FORMS"]="0"
        values.update({f"allocations-TOTAL_FORMS":"1","allocations-INITIAL_FORMS":"1",
            "allocations-0-consumption":allocation.consumption_id,
            "allocations-0-quotation_line":allocation.line.quotation_line_id,
            "allocations-0-quantity":allocation.quantity,
            "allocations-0-reason":"Confirmed previously automatic allocation"})
        response=self.client.post(url,values)
        self.assertEqual(response.status_code,302,response.content.decode()[:6000])
        draft.refresh_from_db()
        reconciled=invoice_queries.invoice_allocations(actor=self.cashier,invoice=draft).get()
        self.assertEqual((reconciled.mode,reconciled.quantity),("MANUAL",allocation.quantity))
        self.assertEqual(draft.lines.filter(is_active=True).count(),1)
        self.ui(self.cashier,"commercial_workflow","invoice-finalize")
        draft.refresh_from_db()
        self.assertEqual((draft.status,draft.customer_pay_total),("FINALIZED",100))

    def test_handover_needs_settlement_or_an_authorized_financial_release(self):
        from apps.commercial.models import ServiceFinancialRelease
        self.arrive(False); self.diagnose(); self.plan(); self.use_parts(); self.quality(); self.release()
        self.ui(self.cashier,"commercial_workflow","invoice-prepare")
        self.ui(self.cashier,"commercial_workflow","invoice-finalize")
        url=reverse("operations:workflow",args=[self.case.pk,"delivery"])
        self.client.force_login(self.front)
        page=self.client.get(url)
        values=dict(page.context["form"].initial)
        values.update(operation="handover",recipient_type="CUSTOMER",verification_method="CUSTOMER_RECORD",
            device_confirmed="on",verification_confirmed="on",physical_handover_confirmed="on")
        blocked=self.client.post(url,values)
        self.assertEqual(blocked.status_code,400)
        self.assertContains(blocked,"Outstanding customer liability",status_code=400)
        self.case.refresh_from_db()
        self.assertEqual(self.case.status,"READY_FOR_DELIVERY")
        # Repeated attempts do not create a release; only authorization can.
        self.assertEqual(self.client.post(url,values).status_code,400)
        self.assertFalse(ServiceFinancialRelease.objects.exists())
        self.ui(self.finance,"commercial_workflow","payment-release",dict(reason="Customer goodwill waiver approved by the service manager",reference="UAT-WAIVER"))
        self.assertEqual(ServiceFinancialRelease.objects.count(),1)
        self.ui(self.front,"workflow","delivery",dict(operation="handover",recipient_type="CUSTOMER",verification_method="CUSTOMER_RECORD",
            device_confirmed="on",verification_confirmed="on",physical_handover_confirmed="on"))
        self.assertEqual(self.case.status,"DELIVERED")
        self.ui(self.front,"workflow","delivery",dict(operation="close"))
        self.assertEqual(self.case.status,"CLOSED")
        self.assertEqual(self.client.post(url,values).status_code,403)

    def test_payment_reversal_restores_the_outstanding_balance(self):
        from apps.commercial.models import PaymentReversal, ServicePayment, ServicePaymentReceipt
        self.arrive(False); self.diagnose(); self.plan(); self.use_parts(); self.quality(); self.release()
        row=invoices.prepare_service_invoice(actor=self.cashier,service_case=self.case)
        self.invoice=invoices.finalize_service_invoice(actor=self.cashier,invoice=row,expected_revision=invoices.revision(row))
        self.ui(self.cashier,"commercial_workflow","payment-receive",dict(amount="40",method="CASH"))
        payment=ServicePayment.objects.get()
        self.assertEqual(payment_queries.invoice_settlement_summary(actor=self.cashier,invoice=self.invoice)["balance"],60)
        self.ui(self.finance,"commercial_workflow","payment-reverse",dict(payment=payment.pk,reason="Cash drawer reconciliation error"))
        self.assertEqual(PaymentReversal.objects.count(),1)
        self.assertEqual((ServicePayment.objects.get().status,ServicePaymentReceipt.objects.count()),("VOIDED",1))
        self.assertEqual(ServicePaymentReceipt.objects.count(),1)
        self.assertEqual(payment_queries.invoice_settlement_summary(actor=self.cashier,invoice=self.invoice)["balance"],100)
        # Cashier cannot reverse; reversal is not on the payment-intake boundary.
        self.ui(self.cashier,"commercial_workflow","payment-receive",dict(amount="100",method="CASH"))
        self.client.force_login(self.cashier)
        reversal_url=reverse("operations:commercial_workflow",args=[self.case.pk,"payment-reverse"])
        self.assertEqual(self.client.get(reversal_url).status_code,403)
        self.assertEqual(self.client.post(reversal_url,{}).status_code,403)

    def test_parts_workflow_evidence_and_signed_command_replay(self):
        from apps.inventory.models import PartsRequest, StockReservation
        self.arrive(False); self.diagnose()
        url=reverse("operations:parts_workflow",args=[self.case.pk,"receive"])
        self.client.force_login(self.warehouse)
        page=self.client.get(url)
        self.assertContains(page,"No parts requested for this job yet.")
        values=dict(page.context["form"].initial)
        values["revision"]=page.context["revision"]
        values.update(spare_part=self.part.pk,location=self.location.pk,quantity=2,reference="UAT idempotent receipt")
        self.assertEqual(self.client.post(url,values).status_code,302)
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.location,spare_part=self.part),2)
        # Replaying the identical signed command must not post the receipt twice.
        self.assertEqual(self.client.post(url,values).status_code,400)
        self.assertEqual(inventory.stock_on_hand(actor=self.warehouse,location=self.location,spare_part=self.part),2)
        self.assertEqual(StockLedgerEntry.objects.filter(spare_part=self.part).count(),1)
        self.ui(self.engineer,"parts_workflow","request",dict(spare_part=self.part.pk,quantity=2))
        request=PartsRequest.objects.get(service_case=self.case)
        self.ui(self.warehouse,"parts_workflow","approve",dict(request=request.pk))
        self.ui(self.warehouse,"parts_workflow","reserve",dict(line=request.lines.get().pk,location=self.location.pk,quantity=2))
        # The evidence screen shows this job's demand, reservation and visible stock.
        evidence=self.client.get(url)
        self.assertContains(evidence,self.part.part_code)
        self.assertContains(evidence,self.location.code)
        self.assertContains(evidence,"ACTIVE")
        self.assertContains(evidence,str(self.engineer.username))

    def test_reservation_release_then_reserve_again(self):
        from apps.inventory.models import PartsIssue, PartsRequest, StockReservation
        self.arrive(False); self.diagnose()
        self.ui(self.warehouse,"parts_workflow","receive",dict(spare_part=self.part.pk,location=self.location.pk,quantity=2,reference="UAT store receipt"))
        self.ui(self.engineer,"parts_workflow","request",dict(spare_part=self.part.pk,quantity=2))
        request=PartsRequest.objects.get(service_case=self.case)
        self.ui(self.warehouse,"parts_workflow","approve",dict(request=request.pk))
        self.ui(self.warehouse,"parts_workflow","reserve",dict(line=request.lines.get().pk,location=self.location.pk,quantity=2))
        reservation=StockReservation.objects.get(service_case=self.case)
        self.assertEqual(inventory.available_stock(actor=self.warehouse,location=self.location,spare_part=self.part),0)
        self.ui(self.warehouse,"parts_workflow","release",dict(reservation=reservation.pk,reason="Store relocated the pick bucket"))
        reservation.refresh_from_db()
        self.assertEqual((reservation.status,reservation.reason),("RELEASED","Store relocated the pick bucket"))
        self.assertEqual(inventory.available_stock(actor=self.warehouse,location=self.location,spare_part=self.part),2)
        self.assertFalse(PartsIssue.objects.filter(service_case=self.case).exists())
        # Released stock can be reserved again for the same approved demand.
        self.ui(self.warehouse,"parts_workflow","reserve",dict(line=request.lines.get().pk,location=self.location.pk,quantity=1))
        self.assertEqual(StockReservation.objects.filter(service_case=self.case,status="ACTIVE").get().quantity,1)

    def test_quotation_lines_offer_only_compatible_parts(self):
        self.arrive(False); self.diagnose()
        category=parts.create_part_category(code="UATFOREIGN",name="Foreign parts")
        foreign=parts.create_spare_part(category=category,part_code="UATFOREIGN",name="Foreign part",serialization_policy="NOT_SERIALIZED")
        self.ui(self.cashier,"commercial_workflow","quote-create",dict(currency="BDT"))
        page=self.client.get(reverse("operations:commercial_workflow",args=[self.case.pk,"quote-lines"]))
        self.assertContains(page,self.part.part_code)
        self.assertNotContains(page,foreign.part_code)

    def test_device_registration_rejects_a_variant_of_another_model(self):
        from apps.catalog.models import ProductVariant
        from apps.devices.models import Device
        variant=ProductVariant.objects.get(product_model=self.model)
        foreign_variant=ProductVariant.objects.create(product_model=self.model2,code="UATFOREIGN",name="Foreign variant")
        self.client.force_login(self.front)
        url=reverse("operations:register_device",args=[self.customer.pk])
        before=Device.objects.count()
        response=self.client.post(url,dict(product_model=self.model.pk,product_variant=foreign_variant.pk,serial="UAT-MISMATCH",ownership_confirmed="on"))
        self.assertEqual(response.status_code,400)
        self.assertContains(response,"Choose a variant of the selected product model.",status_code=400)
        self.assertEqual(Device.objects.count(),before)
        accepted=self.client.post(url,dict(product_model=self.model.pk,product_variant=variant.pk,serial="UAT-MATCH",ownership_confirmed="on"))
        self.assertEqual(accepted.status_code,302)
        self.assertEqual(Device.objects.count(),before+1)

    def test_new_endpoint_persona_denial_and_cross_company_idor(self):
        self.arrive(False); self.diagnose()
        foreign=intake_fixture.intake(self,company=self.other_company,service_center=self.other_center,customer=self.outsider,device=self.device2)
        for actor,route,operation in ((self.cashier,"parts_workflow","receive"),(self.cashier,"parts_workflow","reserve"),
                (self.engineer,"commercial_workflow","quote-create"),(self.inspector,"workflow","assignment"),
                (self.inspector,"parts_workflow","request"),(self.engineer,"workflow","qc")):
            with self.subTest(persona=actor.username,route=route,operation=operation):
                self.client.force_login(actor)
                self.assertEqual(self.client.get(reverse("operations:"+route,args=[self.case.pk,operation])).status_code,403)
                target=reverse("operations:"+route,args=[foreign.pk,operation])
                self.assertEqual(self.client.get(target).status_code,404)
                self.assertEqual(self.client.post(target,{}).status_code,404)
        # Commercial and inventory personas cannot reach each other's boundaries.
        self.client.force_login(self.engineer)
        self.assertEqual(self.client.get(reverse("operations:commercial_workflow",args=[self.case.pk,"payment-receive"])).status_code,403)
        self.assertEqual(self.client.get(reverse("operations:parts_workflow",args=[self.case.pk,"receive"])).status_code,403)
        self.client.force_login(self.cashier)
        self.assertEqual(self.client.get(reverse("operations:parts_workflow",args=[self.case.pk,"reserve"])).status_code,403)
        self.client.logout()
        self.assertEqual(self.client.get(reverse("operations:parts_workflow",args=[self.case.pk,"receive"])).status_code,302)
        self.assertEqual(self.client.get(reverse("operations:register_device",args=[self.customer.pk])).status_code,302)

    def test_operational_screens_mobile_and_desktop_contract(self):
        """Every operational screen honours the shared 390px/1366px shell contract."""
        from django.conf import settings
        from pathlib import Path
        self.arrive()

        def check(screens):
            for actor,url in screens:
                with self.subTest(screen=url):
                    self.client.force_login(actor)
                    response=self.client.get(url)
                    self.assertEqual(response.status_code,200)
                    self.assertTemplateUsed(response,"operations/base.html")
                    for text in ('name="viewport"','Skip to content','operations/workspace.css','operations/workspace.js',
                            'aria-label="Application navigation"'):
                        self.assertContains(response,text)
                    content=response.content.decode()
                    # A 390px viewport needs reflow, never a fixed desktop width.
                    self.assertNotRegex(content,r"(?<!max-)width:\s*[4-9]\d\dpx")
                    self.assertNotIn("<table",content)
                    for form in re.findall(r"<form\b.*?</form>",content,flags=re.S):
                        if 'method="post"' in form:
                            self.assertIn("csrfmiddlewaretoken",form)
                    self.capture(response,"screen-"+url.strip("/").replace("/","-"))

        # Engineer assignment is only offered before diagnosis begins.
        check([(self.front,reverse("operations:workflow",args=[self.case.pk,"assignment"]))])
        self.diagnose()
        self.ui(self.cashier,"commercial_workflow","quote-create",dict(currency="BDT"))
        check([
            (self.front,reverse("operations:dashboard")),
            (self.front,reverse("operations:search")),
            (self.front,reverse("operations:list",args=["cases"])),
            (self.front,reverse("operations:list",args=["appointments"])),
            (self.front,reverse("operations:detail",args=["cases",self.case.pk])),
            (self.front,reverse("operations:detail",args=["customers",self.customer.pk])),
            (self.front,reverse("operations:detail",args=["devices",self.device.pk])),
            (self.front,reverse("operations:case_communications",args=[self.case.pk])),
            (self.front,reverse("operations:case_parts",args=[self.case.pk])),
            (self.front,reverse("operations:register_device",args=[self.customer.pk])),
            (self.engineer,reverse("operations:workflow",args=[self.case.pk,"repair"])),
            (self.cashier,reverse("operations:commercial_workflow",args=[self.case.pk,"quote-lines"])),
            (self.warehouse,reverse("operations:parts_workflow",args=[self.case.pk,"receive"])),
            (self.supervisor,reverse("operations:reservations")),
        ])
        css=(Path(settings.BASE_DIR)/"apps/operations/static/operations/workspace.css").read_text()
        js=(Path(settings.BASE_DIR)/"apps/operations/static/operations/workspace.js").read_text()
        for rule in ("@media(max-width:1000px)","@media(max-width:760px)","table,tbody,tr,td{display:block;width:100%;min-width:0}",
                "td[data-label]::before",":focus-visible"):
            self.assertIn(rule,css)
        self.assertIn("cell.dataset.label",js)
        self.assertNotIn("innerHTML",js)