from concurrent.futures import ThreadPoolExecutor
from datetime import time, timedelta
from io import StringIO
from threading import Event
from unittest.mock import patch
from uuid import uuid4

from django.contrib.auth.models import Permission
from django.core.exceptions import PermissionDenied, ValidationError
from django.core.management import call_command
from django.db import connections, transaction
from django.test import TransactionTestCase, Client, override_settings
from django.urls import reverse

from apps.frontdesk.tests import prepare
from apps.frontdesk import services as frontdesk
from apps.customers import services as customers
from apps.service import services as intake
from apps.service.models import ServiceCase
from apps.organization import test_assignment_concurrency as concurrency
from apps.organization.assignment_services import create_assignment
from apps.access.models import Role
from apps.access.services import create_role_assignment
from . import services as s, hooks, providers, queries
from .models import Notification, NotificationAttempt, NotificationTemplate
from .rendering import render

FAKES = {"SMS": "apps.communications.providers.FakeSmsProvider", "EMAIL": "apps.communications.providers.FakeEmailProvider"}


class RejectProvider(providers.SmsProvider):
    identifier = "test-rejected"

    def send(self, **kwargs):
        return providers.DeliveryResult("REJECTED")


class UncertainProvider(providers.SmsProvider):
    identifier = "test-uncertain"

    def send(self, **kwargs):
        raise RuntimeError("SECRET-TOKEN <script>alert(1)</script> private@example.test")


@override_settings(COMMUNICATION_PROVIDERS=FAKES)
class CommunicationTests(TransactionTestCase):
    run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent

    def setUp(self):
        prepare(self)
        self.customer = customers.update_customer(customer=self.customer, primary_mobile="+8801700000098", primary_email="synthetic@example.test")
        self.template = self.make_template()

    def make_template(self, **kwargs):
        return s.save_template(**(dict(actor=self.admin, company=self.company, code="confirmation", name="Confirmation",
            channel="SMS", subject="", body="Hello {customer_name}") | kwargs))

    def request(self, **kwargs):
        return s.request_notification(**(dict(actor=self.admin, template=self.template, customer=self.customer,
            service_center=self.center, event_key="MANUAL:test") | kwargs))

    def send(self, row):
        return s.send_notification(actor=self.admin, notification=row)

    def appointment(self):
        slot = frontdesk.create_slot(actor=self.admin, service_center=self.center, date=self.today + timedelta(days=1),
                                    start_time=time(9), end_time=time(10), capacity=2)
        return frontdesk.book_appointment(actor=self.admin, slot=slot, customer=self.customer)

    def test_literal_rendering_and_unknown_fields(self):
        self.assertEqual(render(self.template, {"customer_name": "Synthetic"}), ("", "Hello Synthetic"))
        for text in ["{unsupported}", "{customer_name.__class__}", "{customer_name[0]}", "{customer_name!r}", "{customer_name:>4}", "{"]:
            with self.subTest(text=text), self.assertRaises(ValidationError):
                self.make_template(code="invalid", body=text)

    def test_missing_context_rejected(self):
        template = self.make_template(code="specific", body="Case {service_case_reference}")
        with self.assertRaises(ValidationError):
            self.request(template=template)

    def test_email_subject_and_header_validation(self):
        for subject in ("", "Header\r\nBcc: leaked@example.test"):
            with self.subTest(subject=subject), self.assertRaises(ValidationError):
                self.make_template(code="email", channel="EMAIL", subject=subject)

    def test_snapshot_survives_template_edit(self):
        row = self.request()
        self.make_template(template=self.template, body="Changed {customer_name}")
        row.refresh_from_db()
        self.assertTrue(row.body.startswith("Hello "))
        self.assertEqual(row.destination, "+8801700000098")

    def test_company_template_code_channel_uniqueness(self):
        with self.assertRaises(ValidationError):
            self.make_template()
        self.make_template(channel="EMAIL", subject="Hello")

    def test_inactive_template_denies_new_request(self):
        self.make_template(template=self.template, is_active=False)
        with self.assertRaises(ValidationError):
            self.request()

    def test_sms_and_email_fake_providers(self):
        sms = self.send(self.request())
        email = self.make_template(code="mail", channel="EMAIL", subject="Hello {customer_name}")
        mail = self.send(self.request(template=email))
        self.assertEqual((sms.status, mail.status), ("SENT", "SENT"))
        self.assertEqual(sms.attempts.get().provider, "fake-sms")
        self.assertEqual(mail.attempts.get().provider, "fake-email")
        self.assertIsNotNone(mail.sent_at)

    def test_django_email_adapter_uses_backend_without_real_network(self):
        with patch.object(providers, "EmailMessage") as email:
            email.return_value.send.return_value = 1
            result = providers.DjangoEmailProvider().send(destination="synthetic@example.test", subject="Test", body="Test", idempotency_key="key")
        self.assertEqual(result.outcome, "ACCEPTED")
        email.return_value.send.assert_called_once_with(fail_silently=False)

    def test_definite_rejection_then_retry_preserves_attempt(self):
        row = self.request()
        with patch.object(s, "get_provider", return_value=RejectProvider()):
            self.assertEqual(self.send(row).status, "FAILED")
        self.assertEqual(self.send(row).status, "SENT")
        self.assertEqual(list(row.attempts.values_list("number", "result")), [(1, "REJECTED"), (2, "ACCEPTED")])

    def test_unknown_failure_is_sanitized_and_not_retried(self):
        row = self.request()
        with patch.object(s, "get_provider", return_value=UncertainProvider()):
            self.assertEqual(self.send(row).status, "UNKNOWN")
        self.assertEqual(self.send(row).status, "UNKNOWN")
        attempt = row.attempts.get()
        self.assertEqual(attempt.error_code, "OUTCOME_UNKNOWN")
        self.assertNotIn("SECRET", str(vars(attempt)))
        self.assertEqual(attempt.provider_reference, "")

    def test_unconfigured_provider_records_failure(self):
        with self.settings(COMMUNICATION_PROVIDERS={}):
            row = self.send(self.request())
        self.assertEqual(row.status, "FAILED")
        self.assertEqual(row.attempts.get().provider, "unconfigured")

    def test_sent_retry_is_noop(self):
        row = self.send(self.request())
        with patch.object(s, "get_provider") as provider:
            self.send(row)
        provider.assert_not_called()
        self.assertEqual(row.attempts.count(), 1)

    def test_cancellation_and_no_history_deletion(self):
        row = self.request()
        s.cancel_notification(actor=self.admin, notification=row)
        self.assertEqual(self.send(row).status, "CANCELLED")
        self.assertEqual(row.attempts.count(), 0)
        for action in (row.save, row.delete, lambda: Notification.objects.all().delete(),
                       lambda: Notification.objects.all().update(status="SENT")):
            with self.assertRaises(ValidationError):
                action()

    def test_finished_attempt_cannot_be_rewritten(self):
        attempt = self.send(self.request()).attempts.get()
        attempt.result = "REJECTED"
        with self.assertRaises(ValidationError):
            attempt._persist()

    def test_creation_idempotency_and_rebinding_rejected(self):
        first = self.request()
        self.assertEqual(self.request().pk, first.pk)
        with self.assertRaises(ValidationError):
            self.request(customer=self.presenter)
        self.assertEqual(Notification.objects.count(), 1)

    def test_concurrent_creation_uses_database_serialization(self):
        self.run_concurrent(lambda: self.request(), lambda: self.request(actor=self.second), expected="success")
        self.assertEqual(Notification.objects.count(), 1)

    def test_concurrent_sends_make_one_provider_call(self):
        row = self.request()
        entered, release = Event(), Event()
        def deliver(**kwargs):
            entered.set()
            if not release.wait(10):
                raise AssertionError("Provider release timed out")
            return providers.DeliveryResult("ACCEPTED", "synthetic-reference")
        def worker():
            try:
                return self.send(row).status
            finally:
                connections.close_all()
        provider = providers.FakeSmsProvider()
        with patch.object(provider, "send", side_effect=deliver) as calls, patch.object(s, "get_provider", return_value=provider):
            with ThreadPoolExecutor(max_workers=1) as executor:
                first = executor.submit(worker)
                try:
                    self.assertTrue(entered.wait(5))
                    self.assertEqual(self.send(row).status, "SENDING")
                finally:
                    release.set()
                self.assertEqual(first.result(timeout=10), "SENT")
            self.assertEqual(calls.call_count, 1)
        self.assertEqual(row.attempts.count(), 1)

    def test_crashed_claim_is_not_resent(self):
        row = self.request()
        with patch.object(providers.FakeSmsProvider, "send", side_effect=KeyboardInterrupt):
            with self.assertRaises(KeyboardInterrupt):
                self.send(row)
        self.assertEqual(self.send(row).status, "SENDING")
        self.assertEqual(row.attempts.get().result, "STARTED")

    def test_contact_change_prevents_send_without_retargeting(self):
        row = self.request()
        customers.update_customer(customer=self.customer, primary_mobile="+8801700000099")
        with patch.object(providers.FakeSmsProvider, "send") as send:
            result = self.send(row)
        send.assert_not_called()
        self.assertEqual(result.status, "FAILED")
        self.assertEqual(result.attempts.get().error_code, "CONTACT_CHANGED")
        self.assertEqual(result.destination, "+8801700000098")

    def test_authoritative_primary_contact_is_preferred(self):
        contact = customers.create_customer_contact(customer=self.customer, contact_type="MOBILE", value="+8801700000099")
        customers.set_primary_customer_contact(contact=contact)
        self.assertEqual(self.request().destination, "+8801700000099")

    def test_cross_company_template_and_customer_are_rejected(self):
        with self.assertRaises(ValidationError):
            self.request(customer=self.outsider)
        with self.assertRaises(ValidationError):
            self.request(service_center=self.other_center)

    def grant(self):
        assignment = create_assignment(user=self.user, company=self.company, region=self.region, service_center=self.center)
        role = Role.objects.create(code="COMMS", name="Communications")
        role.permissions.set(Permission.objects.filter(content_type__app_label="communications", codename__in=["view_notification", "send_notification"]))
        create_role_assignment(user=self.user, role=role, organization_assignment=assignment)

    def test_scope_and_staff_do_not_bypass_authorization(self):
        row = self.request()
        self.user.is_staff = True
        self.user.save()
        with self.assertRaises(PermissionDenied):
            s.send_notification(actor=self.user, notification=row)
        self.grant()
        self.assertEqual(list(queries.history(self.user)), [row])
        other = self.request(service_center=self.center2, event_key="other")
        self.assertNotIn(other, list(queries.history(self.user)))
        with self.assertRaises(PermissionDenied):
            s.send_notification(actor=self.user, notification=other)

    def test_reminders_are_explicit_upcoming_and_idempotent(self):
        a = self.appointment()
        for _ in range(2):
            rows = hooks.queue_appointment_reminders(actor=self.admin, template=self.template, date=a.slot.date, policy="day-before")
            self.assertEqual(len(rows), 1)
        self.assertEqual(Notification.objects.count(), 1)
        self.assertEqual(NotificationAttempt.objects.count(), 0)

    def test_reminder_command_only_queues(self):
        a = self.appointment()
        for _ in range(2):
            call_command("queue_appointment_reminders", actor=str(self.admin.pk), template=str(self.template.pk),
                         date=a.slot.date.isoformat(), policy="tomorrow", stdout=StringIO())
        self.assertEqual(Notification.objects.count(), 1)
        self.assertFalse(NotificationAttempt.objects.exists())

    def test_cancelled_appointment_suppresses_queued_reminder(self):
        a = self.appointment()
        row = hooks.workflow_notification(actor=self.admin, template=self.template, event="APPOINTMENT_REMINDER", record=a, policy="day-before")
        frontdesk.end_appointment(actor=self.admin, appointment=a, reason="Customer cancelled")
        self.assertEqual(self.send(row).attempts.get().error_code, "CONTEXT_CHANGED")

    def test_failure_does_not_rollback_existing_case(self):
        case = intake.create_service_case(company=self.company, service_center=self.center, customer=self.customer,
            device=self.device, created_by=self.admin, intake_channel="WALK_IN")
        row = hooks.workflow_notification(actor=self.admin, template=self.template, event="SERVICE_INTAKE", record=case)
        with patch.object(s, "get_provider", return_value=RejectProvider()):
            self.assertEqual(self.send(row).status, "FAILED")
        self.assertEqual(ServiceCase.objects.get(pk=case.pk).status, "RECEIVED")

    def test_post_commit_hook_not_run_on_rollback(self):
        a = self.appointment()
        with self.assertRaises(RuntimeError), transaction.atomic():
            hooks.after_commit_notification(actor=self.admin, template=self.template, event="APPOINTMENT_CONFIRMATION", record=a)
            raise RuntimeError("Rollback")
        self.assertFalse(Notification.objects.exists())

    def test_preparation_failure_logs_safely_without_invalidating_commit(self):
        a = self.appointment()
        with self.assertLogs("communications", level="ERROR") as logs:
            with transaction.atomic():
                hooks.after_commit_notification(actor=self.admin, template=self.template, event="NOT_AN_EVENT", record=a)
        self.assertIn("NOTIFICATION_PREPARATION_FAILED", logs.output[0])
        a.refresh_from_db()
        self.assertEqual(a.status, "SCHEDULED")

    def test_send_inside_authoritative_transaction_is_rejected_before_io(self):
        row = self.request()
        with transaction.atomic(), self.assertRaises(ValidationError):
            self.send(row)
        self.assertFalse(row.attempts.exists())

    def test_ui_escapes_content_masks_address_and_requires_csrf(self):
        self.make_template(template=self.template, body="<script>alert(1)</script>")
        row = self.request()
        self.client.force_login(self.admin)
        response = self.client.get(reverse("communications:history"))
        self.assertContains(response, "&lt;script&gt;")
        self.assertNotContains(response, row.destination)
        client = Client(enforce_csrf_checks=True)
        client.force_login(self.admin)
        url = reverse("communications:action", args=[row.pk, "send"])
        self.assertEqual(client.post(url).status_code, 403)
        self.assertEqual(client.get(url).status_code, 200)
        self.assertFalse(row.attempts.exists())

    def test_ui_template_management_and_manual_request(self):
        self.client.force_login(self.admin)
        self.assertEqual(self.client.get(reverse("communications:templates")).status_code, 200)
        url = reverse("communications:template-edit", args=[self.template.pk])
        data = dict(company=self.company.pk, code=self.template.code, channel="SMS", name="Edited", subject="", body="Edited {customer_name}", is_active="on")
        self.assertEqual(self.client.post(url, data).status_code, 302)
        self.assertEqual(self.client.post(reverse("communications:manual"), {"template": self.template.pk, "customer": self.customer.pk, "request_key": uuid4()}).status_code, 302)

    def test_ui_idor_and_anonymous_access(self):
        row = self.request(service_center=self.center2)
        url = reverse("communications:action", args=[row.pk, "send"])
        self.assertEqual(self.client.post(url).status_code, 302)
        self.grant()
        self.client.force_login(self.user)
        self.assertEqual(self.client.post(url).status_code, 404)


class CommercialHookTests(TransactionTestCase):
    def setup_actor_and_template(self, fixture, body):
        fixture.actor.is_superuser = True
        fixture.actor.save()
        customers.update_customer(customer=fixture.case.customer, primary_mobile="+8801700000088")
        return s.save_template(actor=fixture.actor, company=fixture.case.company, code="commercial", name="Commercial",
                               channel="SMS", subject="", body=body)

    def test_quotation_ready_and_approved_use_current_revision(self):
        from apps.inventory.test_usage import setup_usage
        from apps.commercial.tests import QuotationFixture
        fixture = QuotationFixture()
        setup_usage(fixture)
        template = self.setup_actor_and_template(fixture, "Quotation {quotation_reference}")
        quote = fixture.submit(fixture.lines(fixture.create()))
        row = hooks.workflow_notification(actor=fixture.actor, template=template, event="QUOTATION_READY", record=quote)
        self.assertIn(quote.family.number, row.body)
        quote = fixture.decide(quote)
        row = hooks.workflow_notification(actor=fixture.actor, template=template, event="QUOTATION_APPROVED", record=quote)
        self.assertIn("revision 1", row.body)
        with self.assertRaises(ValidationError):
            hooks.workflow_notification(actor=fixture.actor, template=template, event="QUOTATION_READY", record=quote)

    def test_payment_receipt_and_ready_for_delivery_use_authoritative_evidence(self):
        from apps.commercial.test_payment import PaymentFixture, setup_payment
        from apps.service.test_handover import ready
        fixture = PaymentFixture()
        setup_payment(fixture)
        template = self.setup_actor_and_template(fixture, "Receipt {receipt_reference}; paid {payment_amount}; due {amount_due}")
        payment = fixture.pay()
        row = hooks.workflow_notification(actor=fixture.actor, template=template, event="PAYMENT_RECEIPT", record=payment)
        self.assertIn(payment.receipt.number, row.body)
        self.assertIn("due BDT 0", row.body)
        simple = s.save_template(actor=fixture.actor, company=fixture.case.company, code="ready", name="Ready",
            channel="SMS", subject="", body="Ready: {service_case_reference}")
        with self.assertRaises(ValidationError):
            hooks.workflow_notification(actor=fixture.actor, template=simple, event="READY_FOR_DELIVERY", record=fixture.case)
        ready(fixture)
        row = hooks.workflow_notification(actor=fixture.actor, template=simple, event="READY_FOR_DELIVERY", record=fixture.case)
        self.assertIn(fixture.case.job_number, row.body)