"""Synthetic address/contact integrity, admin and PostgreSQL concurrency checks."""
import uuid
from unittest.mock import patch

from django.contrib import admin
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.core.exceptions import ValidationError
from django.db import IntegrityError, transaction
from django.db.models.deletion import ProtectedError
from django.test import Client, TestCase, TransactionTestCase
from django.urls import reverse

from apps.access.authorization import is_authorized, authorized_queryset
from apps.organization import services as organization_services
from apps.organization import test_assignment_concurrency as concurrency
from . import services, queries
from .models import Customer, CustomerAddress, CustomerContact
from .tests import make_company, create


def address(customer, **kwargs):
    return services.create_customer_address(**{
        "customer": customer, "address_type": "HOME", "address_line_1": "123 Example Street",
        "city": "Example City", "country_code": "US", **kwargs,
    })


def contact(customer, **kwargs):
    return services.create_customer_contact(**{
        "customer": customer, "contact_type": "MOBILE", "value": "+1 (202) 555-0123", **kwargs,
    })


KINDS = (("address", CustomerAddress, address), ("contact", CustomerContact, contact))


def operate(verb, kind, record, **kwargs):
    return getattr(services, f"{verb}_customer_{kind}")(**{kind: record}, **kwargs)


class CustomerDetailTests(TestCase):
    def setUp(self):
        self.company = make_company()
        self.customer = create(self.company, primary_mobile="+12025550999", primary_email="quick@example.com")
        self.other_customer = create(make_company("OTHER"))

    def test_uuid_timestamps_active_default_and_explicit_nonprimary(self):
        for _, model, make in KINDS:
            record = make(self.customer)
            self.assertIsInstance(record.pk, uuid.UUID)
            self.assertIsNotNone(record.created_at)
            self.assertIsNotNone(record.updated_at)
            self.assertTrue(record.is_active)
            self.assertFalse(record.is_primary)
            self.assertEqual(record.customer_id, self.customer.pk)
            self.assertNotIn("Example Street", str(record))
            self.assertEqual(model.objects.count(), 1)

    def test_all_address_types_and_country_normalization(self):
        for value in CustomerAddress.Type.values:
            record = address(self.customer, address_type=value, country_code=" gb ", city=" London ", label=" Office ")
            self.assertEqual(record.country_code, "GB")
            self.assertEqual(record.city, "London")
            self.assertEqual(record.label, "Office")

    def test_required_address_fields_reject_empty_whitespace_and_invalid_formats(self):
        for field in ("address_line_1", "city", "country_code"):
            for value in ("", " \t\n "):
                with self.subTest(field=field), self.assertRaises(ValidationError):
                    address(self.customer, **{field: value})
        for changes in ({"country_code": "USA"}, {"country_code": "1A"}, {"country_code": "éA"},
                        {"address_type": "INVALID"}, {"country_code": "U\nS"}):
            with self.subTest(changes=changes), self.assertRaises(ValidationError):
                address(self.customer, **changes)
        self.assertFalse(CustomerAddress.objects.exists())

    def test_contact_normalization_by_type_preserves_entered_representation(self):
        for kind in ("MOBILE", "PHONE", "WHATSAPP"):
            record = contact(self.customer, contact_type=kind, value=" +1 (202) 555-0123 ")
            self.assertEqual(record.value, "+1 (202) 555-0123")
            self.assertEqual(record.normalized_value, "+12025550123")
        email = contact(self.customer, contact_type="EMAIL", value=" Contact@EXAMPLE.COM ")
        self.assertEqual(email.value, "Contact@EXAMPLE.COM")
        self.assertEqual(email.normalized_value, "Contact@example.com")
        other = contact(self.customer, contact_type="OTHER", value=" Example channel ")
        self.assertEqual(other.normalized_value, "Example channel")

    def test_contact_invalid_types_blank_values_and_email_phone_syntax(self):
        for kind, value in (("INVALID", "anything"), ("OTHER", " \t "), ("EMAIL", "invalid"),
                            ("PHONE", "123"), ("MOBILE", "+0123456789"), ("WHATSAPP", "letters")):
            with self.subTest(kind=kind), self.assertRaises(ValidationError):
                contact(self.customer, contact_type=kind, value=value)
        self.assertFalse(CustomerContact.objects.exists())

    def test_active_duplicates_rejected_but_shared_values_across_customers_allowed(self):
        first = contact(self.customer)
        with self.assertRaises(ValidationError):
            contact(self.customer, value="+12025550123")
        contact(self.other_customer, value="+12025550123")
        same_company = create(self.company)
        contact(same_company, value="+12025550123")
        services.deactivate_customer_contact(contact=first)
        replacement = contact(self.customer, value="+12025550123")
        with self.assertRaises(ValidationError):
            services.reactivate_customer_contact(contact=first)
        first.refresh_from_db()
        self.assertFalse(first.is_active)
        services.deactivate_customer_contact(contact=replacement)
        self.assertTrue(services.reactivate_customer_contact(contact=first).is_active)

    def test_immutable_customer_and_contact_type_through_model_and_service(self):
        for kind, model, make in KINDS:
            record = make(self.customer)
            record.customer = self.other_customer
            with self.assertRaises(ValidationError):
                record.save()
            with self.assertRaises(ValidationError):
                record.save(update_fields=["customer"])
            with self.assertRaises(TypeError):
                operate("update", kind, record, customer=self.other_customer)
            record.refresh_from_db()
            self.assertEqual(record.customer_id, self.customer.pk)
        record = CustomerContact.objects.get()
        record.contact_type = "EMAIL"
        record.value = "test@example.com"
        with self.assertRaises(ValidationError):
            record.save()
        with self.assertRaises(TypeError):
            services.update_customer_contact(contact=record, contact_type="EMAIL")

    def test_primary_creation_switches_deliberately_and_contacts_are_per_type(self):
        first_address = address(self.customer, is_primary=True)
        second_address = address(self.customer, is_primary=True)
        first = contact(self.customer, is_primary=True)
        email = contact(self.customer, contact_type="EMAIL", value="test@example.com", is_primary=True)
        second = contact(self.customer, value="+12025550124", is_primary=True)
        first_address.refresh_from_db()
        first.refresh_from_db()
        self.assertFalse(first_address.is_primary)
        self.assertFalse(first.is_primary)
        self.assertEqual(queries.primary_address_for_customer(self.customer), second_address)
        self.assertEqual(queries.primary_contact_for_customer(self.customer, contact_type="MOBILE"), second)
        self.assertEqual(queries.primary_contact_for_customer(self.customer, contact_type="EMAIL"), email)
        services.set_primary_customer_contact(contact=first)
        self.assertEqual(queries.primary_contact_for_customer(self.customer, contact_type="EMAIL"), email)
        self.assertEqual(queries.primary_address_for_customer(self.customer), second_address)

    def test_primary_switch_is_idempotent_and_rejects_inactive_target(self):
        for kind, _, make in KINDS:
            record = make(self.customer, is_primary=True)
            self.assertEqual(operate("set_primary", kind, record).updated_at, record.updated_at)
            operate("deactivate", kind, record)
            with self.assertRaises(ValidationError):
                operate("set_primary", kind, record)

    def test_primary_switch_failure_restores_previous_primary(self):
        for kind, model, make in KINDS:
            first = make(self.customer, is_primary=True)
            second = make(self.customer, **({"value": "+12025550124"} if kind == "contact" else {}))
            with patch.object(model, "save", side_effect=ValidationError("Synthetic failed write")):
                with self.assertRaises(ValidationError):
                    operate("set_primary", kind, second)
            first.refresh_from_db()
            second.refresh_from_db()
            self.assertTrue(first.is_primary)
            self.assertFalse(second.is_primary)

    def test_primary_creation_failure_rolls_back_new_row_and_previous_primary(self):
        for kind, model, make in KINDS:
            first = make(self.customer, is_primary=True)
            original = model.save
            def fail_primary(record, **kwargs):
                if record.is_primary:
                    raise ValidationError("Synthetic failure after clearing primary")
                return original(record, **kwargs)
            with patch.object(model, "save", fail_primary):
                with self.assertRaises(ValidationError):
                    make(self.customer, is_primary=True, **({"value": "+12025550124"} if kind == "contact" else {}))
            first.refresh_from_db()
            self.assertTrue(first.is_primary)
            self.assertEqual(model.objects.count(), 1)

    def test_lifecycle_clears_primary_does_not_choose_replacement_or_restore_primary(self):
        for kind, model, make in KINDS:
            first = make(self.customer, is_primary=True)
            make(self.customer, **({"value": "+12025550124"} if kind == "contact" else {}))
            inactive = operate("deactivate", kind, first)
            self.assertFalse(inactive.is_primary)
            self.assertFalse(inactive.is_active)
            self.assertEqual(operate("deactivate", kind, first).updated_at, inactive.updated_at)
            self.assertFalse(model.objects.filter(is_primary=True).exists())
            active = operate("reactivate", kind, first)
            self.assertTrue(active.is_active)
            self.assertFalse(active.is_primary)
            self.assertEqual(operate("reactivate", kind, first).updated_at, active.updated_at)

    def test_stale_updates_preserve_current_lifecycle_and_primary(self):
        for kind, model, make in KINDS:
            first = make(self.customer, is_primary=True)
            second = make(self.customer, **({"value": "+12025550124"} if kind == "contact" else {}))
            operate("set_primary", kind, second)
            # Even a caller with forged ownership is just an identifier in a
            # service. Ownership is freshly read; it is never copied from input.
            first.customer = self.other_customer
            current = operate("update", kind, first, label="Corrected")
            self.assertEqual(current.customer_id, self.customer.pk)
            self.assertFalse(current.is_primary)
            operate("deactivate", kind, second)
            current = operate("update", kind, second, label="Still inactive")
            self.assertFalse(current.is_active)
            self.assertFalse(current.is_primary)

    def test_partial_model_save_validates_effective_state_and_recomputes_normalization(self):
        record = contact(self.customer, is_primary=True)
        services.deactivate_customer_contact(contact=record)
        record.value = "+1 (202) 555-0124"
        record.normalized_value = "forged"
        record.save(update_fields=["value"])
        record.refresh_from_db()
        self.assertFalse(record.is_active)
        self.assertFalse(record.is_primary)
        self.assertEqual(record.normalized_value, "+12025550124")
        home = address(self.customer, is_primary=True)
        services.deactivate_customer_address(address=home)
        home.city = "Corrected City"
        home.save(update_fields=["city"])
        home.refresh_from_db()
        self.assertFalse(home.is_active)
        self.assertFalse(home.is_primary)

    def test_customer_deactivation_preserves_records_flags_and_history(self):
        records = [address(self.customer, is_primary=True), contact(self.customer, is_primary=True)]
        before = [(record.pk, record.updated_at) for record in records]
        services.deactivate_customer(customer=self.customer)
        for record, original in zip(records, before):
            record.refresh_from_db()
            self.assertEqual((record.pk, record.updated_at), original)
            self.assertTrue(record.is_active)
            self.assertTrue(record.is_primary)
        self.assertFalse(queries.active_addresses_for_customer(self.customer).exists())
        self.assertFalse(queries.active_contacts_for_customer(self.customer).exists())
        self.assertIsNone(queries.primary_address_for_customer(self.customer))
        self.assertIsNone(queries.primary_contact_for_customer(self.customer, contact_type="MOBILE"))
        services.reactivate_customer(customer=self.customer)
        self.assertEqual(queries.primary_address_for_customer(self.customer), records[0])

    def test_inactive_company_or_customer_rejects_create_reactivate_and_primary(self):
        for kind, _, make in KINDS:
            active = make(self.customer)
            inactive = make(self.customer, **({"value": "+12025550124"} if kind == "contact" else {}))
            operate("deactivate", kind, inactive)
            for deactivate, reactivate in (
                (lambda: services.deactivate_customer(customer=self.customer), lambda: services.reactivate_customer(customer=self.customer)),
                (lambda: organization_services.deactivate_company(company=self.company), lambda: organization_services.reactivate_company(company=self.company)),
            ):
                deactivate()
                with self.assertRaises(ValidationError):
                    make(self.customer)
                with self.assertRaises(ValidationError):
                    operate("reactivate", kind, inactive)
                with self.assertRaises(ValidationError):
                    operate("set_primary", kind, active)
                self.assertEqual(operate("update", kind, active, label="Historical correction").label, "Historical correction")
                reactivate()

    def test_normal_model_creation_and_reactivation_validate_operational_parents(self):
        home = address(self.customer)
        services.deactivate_customer_address(address=home)
        services.deactivate_customer(customer=self.customer)
        home.is_active = True
        with self.assertRaises(ValidationError):
            home.save()
        with self.assertRaises(ValidationError):
            CustomerContact.objects.create(customer=self.customer, contact_type="EMAIL", value="synthetic@example.com")

    def test_database_address_constraints(self):
        first = address(self.customer, is_primary=True)
        second = address(self.customer)
        for changes in ({"address_type": "INVALID"}, {"country_code": "uS"}, {"country_code": "1A"},
                        {"address_line_1": ""}, {"city": ""}, {"is_primary": True}):
            with self.subTest(changes=changes), self.assertRaises(IntegrityError), transaction.atomic():
                CustomerAddress.objects.filter(pk=second.pk).update(**changes)
        with self.assertRaises(IntegrityError), transaction.atomic():
            CustomerAddress.objects.filter(pk=first.pk).update(is_active=False)

    def test_database_contact_constraints_and_inactive_history(self):
        first = contact(self.customer, is_primary=True)
        second = contact(self.customer, value="+12025550124")
        for changes in ({"contact_type": "INVALID"}, {"value": ""}, {"normalized_value": ""},
                        {"normalized_value": first.normalized_value}, {"is_primary": True}):
            with self.subTest(changes=changes), self.assertRaises(IntegrityError), transaction.atomic():
                CustomerContact.objects.filter(pk=second.pk).update(**changes)
        with self.assertRaises(IntegrityError), transaction.atomic():
            CustomerContact.objects.filter(pk=first.pk).update(is_active=False)
        services.deactivate_customer_contact(contact=second)
        CustomerContact.objects.filter(pk=second.pk).update(normalized_value=first.normalized_value)
        self.assertEqual(CustomerContact.objects.count(), 2)

    def test_contact_update_duplicate_rejection_is_atomic(self):
        first = contact(self.customer)
        second = contact(self.customer, value="+12025550124")
        with self.assertRaises(ValidationError):
            services.update_customer_contact(contact=second, label="Should roll back", value=first.value)
        second.refresh_from_db()
        self.assertEqual(second.label, "")
        self.assertEqual(second.normalized_value, "+12025550124")

    def test_customer_delete_protection_for_each_detail_type(self):
        for _, _, make in KINDS:
            record = make(self.customer)
            with self.assertRaises(ProtectedError):
                self.customer.delete()
            record.delete()

    def test_customer_primary_quick_fields_are_never_synchronized(self):
        record = contact(self.customer, is_primary=True)
        email = contact(self.customer, contact_type="EMAIL", value="additional@example.com", is_primary=True)
        services.update_customer_contact(contact=record, value="+12025550124")
        services.deactivate_customer_contact(contact=email)
        services.reactivate_customer_contact(contact=email)
        self.customer.refresh_from_db()
        self.assertEqual(self.customer.primary_mobile, "+12025550999")
        self.assertEqual(self.customer.primary_email, "quick@example.com")

    def test_queries_are_isolated_lazy_and_respect_current_company_state(self):
        home = address(self.customer, is_primary=True)
        mobile = contact(self.customer, is_primary=True)
        address(self.other_customer, is_primary=True)
        contact(self.other_customer, is_primary=True)
        with self.assertNumQueries(0):
            homes = queries.active_addresses_for_customer(self.customer)
            contacts = queries.active_contacts_for_customer(self.customer)
        with self.assertNumQueries(1):
            self.assertEqual(list(homes), [home])
        with self.assertNumQueries(1):
            self.assertEqual(list(contacts), [mobile])
        with self.assertNumQueries(1):
            self.assertEqual(queries.primary_contact_for_customer(self.customer, contact_type="MOBILE"), mobile)
        with self.assertNumQueries(1):
            self.assertEqual(queries.primary_address_for_customer(self.customer), home)
        pending = queries.active_contacts_for_customer(self.customer)
        organization_services.deactivate_company(company=self.company)
        self.assertEqual(list(pending), [])
        self.assertEqual(queries.active_contacts_for_customer(self.other_customer).count(), 1)
        self.assertFalse(queries.active_addresses_for_customer(self.customer).exists())

    def test_queries_handle_invalid_unsaved_and_missing_inputs(self):
        for customer in (None, Customer()):
            self.assertFalse(queries.active_addresses_for_customer(customer).exists())
            self.assertIsNone(queries.primary_address_for_customer(customer))
            self.assertFalse(queries.active_contacts_for_customer(customer).exists())
        self.assertFalse(queries.active_contacts_for_customer(self.customer, contact_type="INVALID").exists())
        self.assertIsNone(queries.primary_contact_for_customer(self.customer, contact_type=None))
        for _, _, make in KINDS:
            with self.assertRaises(ValidationError):
                make(Customer())

    def test_frozen_authorization_does_not_gain_detail_adapters(self):
        user = get_user_model().objects.create_superuser(username="detail-admin", password="test-only-password")
        for _, model, make in KINDS:
            record = make(self.customer)
            permission = f"customers.view_{model._meta.model_name}"
            self.assertTrue(user.has_perm(permission))
            self.assertFalse(is_authorized(user=user, permission=permission, target=record))
            self.assertFalse(authorized_queryset(user=user, permission=permission, queryset=model.objects.all()).exists())


class CustomerDetailAdminTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.customer = create(make_company())
        cls.other_customer = create(make_company("OTHER"))
        cls.user = get_user_model().objects.create_superuser(username="detail-admin", password="test-only-password")

    def setUp(self):
        self.client.force_login(self.user)

    def payload(self, kind, **kwargs):
        fields = ({"address_type": "HOME", "address_line_1": "123 Example Street", "address_line_2": "",
                   "city": "Example City", "country_code": "us", "area": "", "district": "", "postal_code": ""}
                  if kind == "address" else {"contact_type": "MOBILE", "value": "+1 (202) 555-0123"})
        return {"customer": str(self.customer.pk), "label": "", "_save": "Save", **fields, **kwargs}

    def url(self, model, action, record=None):
        return reverse(f"admin:customers_{model._meta.model_name}_{action}", args=[record.pk] if record else None)

    def test_admin_create_uses_normalization_and_ignores_forged_flags(self):
        for kind, model, _ in KINDS:
            response = self.client.post(self.url(model, "add"), self.payload(kind, is_primary="on", is_active="", normalized_value="forged"))
            self.assertEqual(response.status_code, 302)
            record = model.objects.get()
            self.assertTrue(record.is_active)
            self.assertFalse(record.is_primary)
            self.assertEqual(record.country_code if kind == "address" else record.normalized_value,
                             "US" if kind == "address" else "+12025550123")

    def test_admin_invalid_required_contact_and_duplicate_input_rejected(self):
        for kind, model, _ in KINDS:
            invalid = {"city": " "} if kind == "address" else {"value": "invalid"}
            self.assertEqual(self.client.post(self.url(model, "add"), self.payload(kind, **invalid)).status_code, 200)
            self.assertFalse(model.objects.exists())
        contact(self.customer)
        self.assertEqual(self.client.post(self.url(CustomerContact, "add"), self.payload("contact", value="+12025550123")).status_code, 200)
        self.assertEqual(CustomerContact.objects.count(), 1)

    def test_admin_ownership_type_derived_and_lifecycle_fields_are_protected(self):
        for kind, model, make in KINDS:
            record = make(self.customer)
            response = self.client.post(self.url(model, "change", record), self.payload(kind,
                customer=str(self.other_customer.pk), contact_type="EMAIL", normalized_value="forged", is_primary="on", is_active="", label="Corrected"))
            self.assertEqual(response.status_code, 302)
            record.refresh_from_db()
            self.assertEqual(record.customer_id, self.customer.pk)
            self.assertTrue(record.is_active)
            self.assertFalse(record.is_primary)
            self.assertEqual(record.label, "Corrected")
            if kind == "contact":
                self.assertEqual(record.contact_type, "MOBILE")
                self.assertEqual(record.normalized_value, "+12025550123")

    def test_stale_admin_edit_cannot_restore_deactivated_primary(self):
        for kind, model, make in KINDS:
            record = make(self.customer, is_primary=True)
            registered = admin.site._registry[model]
            original = registered.save_model
            def interleave(request, obj, form, change):
                operate("deactivate", kind, record)
                return original(request, obj, form, change)
            with patch.object(registered, "save_model", side_effect=interleave):
                response = self.client.post(self.url(model, "change", record), self.payload(kind, label="Corrected"))
            self.assertEqual(response.status_code, 302)
            record.refresh_from_db()
            self.assertFalse(record.is_active)
            self.assertFalse(record.is_primary)
            self.assertEqual(record.label, "Corrected")

    def test_stale_admin_edit_cannot_undo_primary_switch(self):
        for kind, model, make in KINDS:
            first = make(self.customer, is_primary=True)
            second = make(self.customer, **({"value": "+12025550124"} if kind == "contact" else {}))
            registered = admin.site._registry[model]
            original = registered.save_model
            def interleave(request, obj, form, change):
                operate("set_primary", kind, second)
                return original(request, obj, form, change)
            with patch.object(registered, "save_model", side_effect=interleave):
                response = self.client.post(self.url(model, "change", first), self.payload(kind, label="Corrected"))
            self.assertEqual(response.status_code, 302)
            first.refresh_from_db()
            second.refresh_from_db()
            self.assertFalse(first.is_primary)
            self.assertTrue(second.is_primary)

    def test_admin_create_lifecycle_race_rolls_back_and_reports_retry(self):
        # The interleaving is in the same outer admin transaction; rejecting
        # creation must leave no partial child or lifecycle write committed.
        for kind, model, _ in KINDS:
            registered = admin.site._registry[model]
            original = registered.save_model
            def interleave(request, obj, form, change):
                services.deactivate_customer(customer=self.customer)
                return original(request, obj, form, change)
            with patch.object(registered, "save_model", side_effect=interleave):
                response = self.client.post(self.url(model, "add"), self.payload(kind))
            self.assertEqual(response.status_code, 302)
            self.assertFalse(model.objects.exists())
            self.customer.refresh_from_db()
            self.assertTrue(self.customer.is_active)

    def test_admin_lifecycle_primary_and_delete_actions(self):
        for kind, model, make in KINDS:
            record = make(self.customer)
            url = self.url(model, "changelist")
            for action, active, primary in (("make_primary", True, True), ("deactivate_selected", False, False), ("reactivate_selected", True, False)):
                self.assertEqual(self.client.post(url, {"action": action, "_selected_action": str(record.pk)}).status_code, 302)
                record.refresh_from_db()
                self.assertEqual((record.is_active, record.is_primary), (active, primary))
            self.assertEqual(self.client.post(self.url(model, "delete", record), {"post": "yes"}).status_code, 403)

    def test_admin_view_only_and_csrf_protect_details(self):
        reader = get_user_model().objects.create_user(username="detail-reader", is_staff=True)
        for kind, model, make in KINDS:
            record = make(self.customer)
            reader.user_permissions.add(Permission.objects.get(content_type__app_label="customers", codename=f"view_{model._meta.model_name}"))
            self.client.force_login(reader)
            self.assertEqual(self.client.post(self.url(model, "change", record), self.payload(kind, label="Forbidden")).status_code, 403)
            client = Client(enforce_csrf_checks=True)
            client.force_login(self.user)
            self.assertEqual(client.post(self.url(model, "add"), self.payload(kind)).status_code, 403)


class CustomerDetailConcurrencyTests(TransactionTestCase):
    run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent

    def setUp(self):
        self.company = make_company()
        self.customer = create(self.company)
        self.other_customer = create(self.company)

    def test_primary_address_switches_serialize(self):
        first, second = address(self.customer), address(self.customer)
        self.run_concurrent(lambda: services.set_primary_customer_address(address=first),
                            lambda: services.set_primary_customer_address(address=second), expected="success")
        self.assertEqual(list(CustomerAddress.objects.filter(is_primary=True)), [second])

    def test_primary_contact_same_type_switches_serialize(self):
        first, second = contact(self.customer), contact(self.customer, value="+12025550124")
        self.run_concurrent(lambda: services.set_primary_customer_contact(contact=first),
                            lambda: services.set_primary_customer_contact(contact=second), expected="success")
        self.assertEqual(list(CustomerContact.objects.filter(is_primary=True)), [second])

    def test_primary_contact_different_types_preserve_each_other(self):
        mobile, email = contact(self.customer), contact(self.customer, contact_type="EMAIL", value="test@example.com")
        self.run_concurrent(lambda: services.set_primary_customer_contact(contact=mobile),
                            lambda: services.set_primary_customer_contact(contact=email), expected="success")
        self.assertEqual(set(CustomerContact.objects.filter(is_primary=True)), {mobile, email})

    def test_duplicate_normalized_contact_creations_serialize_then_reject(self):
        self.run_concurrent(lambda: contact(self.customer), lambda: contact(self.customer, value="+12025550123"), expected="validation")
        self.assertEqual(CustomerContact.objects.count(), 1)
        self.assertTrue(contact(self.customer, value="+12025550124").is_active)

    def test_database_duplicate_constraint_rejects_bypassing_race(self):
        def insert():
            CustomerContact.objects.bulk_create([CustomerContact(customer=self.customer, contact_type="EMAIL",
                value="test@example.com", normalized_value="test@example.com")])
        self.run_concurrent(insert, insert, expected="integrity")
        self.assertEqual(CustomerContact.objects.count(), 1)

    def test_deactivate_then_update_preserves_inactive_and_nonprimary(self):
        for kind, model, make in KINDS:
            record = make(self.customer, is_primary=True)
            self.run_concurrent(lambda: operate("deactivate", kind, record),
                                lambda: operate("update", kind, record, label="Corrected"), expected="success")
            record.refresh_from_db()
            self.assertEqual((record.is_active, record.is_primary, record.label), (False, False, "Corrected"))

    def test_deactivate_then_set_primary_rejects(self):
        for kind, _, make in KINDS:
            record = make(self.customer)
            self.run_concurrent(lambda: operate("deactivate", kind, record),
                                lambda: operate("set_primary", kind, record), expected="validation")
            record.refresh_from_db()
            self.assertFalse(record.is_primary)

    def test_customer_deactivation_then_child_creation_rejects(self):
        for kind, model, make in KINDS:
            services.reactivate_customer(customer=self.customer)
            self.run_concurrent(lambda: services.deactivate_customer(customer=self.customer),
                                lambda: make(self.customer), expected="validation")
            self.assertFalse(model.objects.exists())

    def test_child_creation_then_customer_deactivation_preserves_but_hides_children(self):
        for kind, model, make in KINDS:
            services.reactivate_customer(customer=self.customer)
            self.run_concurrent(lambda: make(self.customer),
                                lambda: services.deactivate_customer(customer=self.customer), expected="success")
            self.assertTrue(model.objects.get().is_active)
            self.assertFalse(queries.active_addresses_for_customer(self.customer).exists())
            self.assertFalse(queries.active_contacts_for_customer(self.customer).exists())

    def test_company_deactivation_then_child_creation_rejects(self):
        for kind, model, make in KINDS:
            organization_services.reactivate_company(company=self.company)
            self.run_concurrent(lambda: organization_services.deactivate_company(company=self.company),
                                lambda: make(self.customer), expected="validation")
            self.assertFalse(model.objects.exists())

    def test_different_customers_in_same_company_do_not_serialize_child_writes(self):
        self.run_concurrent(lambda: address(self.customer), lambda: contact(self.other_customer),
                            expected="success", should_block=False)
        self.assertEqual(CustomerAddress.objects.get().customer_id, self.customer.pk)
        self.assertEqual(CustomerContact.objects.get().customer_id, self.other_customer.pk)

    def test_primary_switch_then_stale_admin_edit_preserves_new_primary(self):
        user = get_user_model().objects.create_superuser(username="detail-race-admin", password="test-only-password")
        for kind, model, make in KINDS:
            first = make(self.customer, is_primary=True)
            second = make(self.customer, **({"value": "+12025550124"} if kind == "contact" else {}))
            def edit():
                client = Client()
                client.force_login(user)
                data = ({"address_type": "HOME", "address_line_1": "123 Example Street", "city": "Example City", "country_code": "US"}
                        if kind == "address" else {"value": "+1 (202) 555-0123"})
                response = client.post(reverse(f"admin:customers_{model._meta.model_name}_change", args=[first.pk]),
                                       {"label": "Corrected", "_save": "Save", **data})
                self.assertEqual(response.status_code, 302)
            self.run_concurrent(lambda: operate("set_primary", kind, second), edit, expected="success")
            first.refresh_from_db()
            second.refresh_from_db()
            self.assertFalse(first.is_primary)
            self.assertTrue(second.is_primary)
            self.assertEqual(first.label, "Corrected")