"""Synthetic ownership history, database, lifecycle and administration tests."""
from datetime import timedelta, datetime
from unittest.mock import patch
import uuid

from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.db import IntegrityError, transaction
from django.test import TestCase, TransactionTestCase
from django.urls import reverse
from django.utils import timezone

from apps.catalog.tests import make_catalog
from apps.customers.tests import make_company, create
from apps.customers import services as customers
from apps.organization import services as organizations
from apps.organization import test_assignment_concurrency as concurrency
from . import services as s, queries as q
from .models import CustomerDeviceRelationship as R, Device
from .tests import policy
from .evidence_admin import revision_token


def setup(test):
    test.company = make_company("OWNERS")
    test.other_company = make_company("OTHER")
    test.a, test.b, test.c = [create(test.company) for _ in range(3)]
    test.outsider = create(test.other_company)
    test.brand, _, test.model, _ = make_catalog("OWNERS")
    policy(test.model)
    test.device = s.register_device(product_model=test.model)
    test.start = timezone.now() - timedelta(days=10)


class OwnershipTests(TestCase):
    def setUp(self):
        setup(self)

    def assign(self, **kwargs):
        return s.assign_device_owner(**dict(device=self.device, customer=self.a, started_at=self.start, **kwargs))

    def test_ownerless_device(self):
        self.assertIsNone(q.current_device_owner(self.device))
        self.assertFalse(q.device_ownership_history(self.device).exists())

    def test_assign_fields_and_queries(self):
        row = self.assign(note="Synthetic history")
        self.assertIsInstance(row.pk, uuid.UUID)
        self.assertEqual(row.relationship_type, "OWNER")
        self.assertIsNone(row.ended_at)
        self.assertIsNotNone(row.created_at)
        self.assertEqual(q.current_device_owner(self.device), self.a)
        with self.assertNumQueries(0):
            devices = q.currently_owned_devices(self.a)
        self.assertEqual(list(devices), [self.device])
        self.assertEqual(list(q.currently_owned_devices_for_company(self.company)), [self.device])
        self.assertFalse(q.currently_owned_devices_for_company(self.other_company).exists())

    def test_duplicate_assign_rejected(self):
        self.assign()
        with self.assertRaises(ValidationError):
            self.assign()

    def test_transfer_back_and_order(self):
        first = self.assign()
        second = s.transfer_device_ownership(device=self.device, new_customer=self.b, reason="Synthetic transfer")
        third = s.transfer_device_ownership(device=self.device, new_customer=self.a)
        first.refresh_from_db(); second.refresh_from_db()
        self.assertEqual(first.ended_at, second.started_at)
        self.assertEqual(second.ended_at, third.started_at)
        self.assertEqual(first.end_reason, "Synthetic transfer")
        self.assertEqual(list(q.device_ownership_history(self.device)), [third, second, first])
        self.assertEqual(q.current_device_owner(self.device), self.a)

    def test_end_preserves_history(self):
        row = self.assign()
        closed = s.end_device_ownership(device=self.device, reason="Synthetic closure")
        self.assertEqual(row.pk, closed.pk)
        self.assertIsNone(q.current_device_owner(self.device))
        self.assertEqual(R.objects.count(), 1)
        with self.assertRaises(ValidationError):
            s.end_device_ownership(device=self.device)

    def test_transfer_requires_owner_and_different_customer(self):
        with self.assertRaises(ValidationError):
            s.transfer_device_ownership(device=self.device, new_customer=self.b)
        self.assign()
        with self.assertRaises(ValidationError):
            s.transfer_device_ownership(device=self.device, new_customer=self.a)

    def test_affinity_survives_end(self):
        self.assign()
        with self.assertRaises(ValidationError):
            s.transfer_device_ownership(device=self.device, new_customer=self.outsider)
        s.end_device_ownership(device=self.device)
        with self.assertRaises(ValidationError):
            s.assign_device_owner(device=self.device, customer=self.outsider)
        s.assign_device_owner(device=self.device, customer=self.b)

    def test_backdating_append_only_and_adjacent_boundary(self):
        self.assign()
        end = self.start + timedelta(days=2)
        s.end_device_ownership(device=self.device, ended_at=end)
        with self.assertRaises(ValidationError):
            s.assign_device_owner(device=self.device, customer=self.b, started_at=end-timedelta(seconds=1))
        row = s.assign_device_owner(device=self.device, customer=self.b, started_at=end)
        self.assertEqual(row.started_at, end)

    def test_invalid_timestamps(self):
        for value in [datetime(2025, 1, 1), timezone.now()+timedelta(days=1), "2025-01-01"]:
            with self.subTest(value=value), self.assertRaises(ValidationError):
                s.assign_device_owner(device=self.device, customer=self.a, started_at=value)
        self.assign()
        with self.assertRaises(ValidationError):
            s.end_device_ownership(device=self.device, ended_at=self.start-timedelta(seconds=1))
        self.assertIsNotNone(q.current_device_owner(self.device))

    def test_stale_preconditions(self):
        old = self.assign()
        current = s.transfer_device_ownership(device=self.device, new_customer=self.b)
        for operation in [lambda: s.end_device_ownership(device=self.device, expected_current_relationship_id=old.pk),
                          lambda: s.transfer_device_ownership(device=self.device, new_customer=self.c, expected_current_relationship_id=old.pk)]:
            with self.assertRaises(ValidationError):
                operation()
        self.assertEqual(q.current_ownership_relationship(self.device), current)

    def test_unsaved_inputs(self):
        with self.assertRaises(ValidationError):
            s.assign_device_owner(device=Device(product_model=self.model), customer=self.a)
        self.assertIsNone(q.current_device_owner(Device()))
        self.assertFalse(q.currently_owned_devices(None).exists())

    def test_stale_customer_rejected_after_deactivation(self):
        customers.deactivate_customer(customer=self.a)
        with self.assertRaises(ValidationError):
            self.assign()

    def test_stale_device_rejected_after_deactivation(self):
        s.deactivate_device(device=self.device)
        with self.assertRaises(ValidationError):
            self.assign()

    def test_inactive_company_rejected(self):
        organizations.deactivate_company(company=self.company)
        with self.assertRaises(ValidationError):
            self.assign()

    def test_lifecycle_preserves_record_and_allows_closure(self):
        row = self.assign()
        customers.deactivate_customer(customer=self.a)
        s.deactivate_device(device=self.device)
        organizations.deactivate_company(company=self.company)
        self.assertEqual(q.current_ownership_relationship(self.device), row)
        self.assertFalse(q.currently_owned_devices(self.a).exists())
        s.end_device_ownership(device=self.device)
        self.assertEqual(q.device_ownership_history(self.device).count(), 1)

    def test_customer_reactivation_does_not_recreate_ended_owner(self):
        self.assign()
        s.end_device_ownership(device=self.device)
        customers.deactivate_customer(customer=self.a)
        customers.reactivate_customer(customer=self.a)
        self.assertIsNone(q.current_device_owner(self.device))

    def test_device_reactivation_preserves_record(self):
        row = self.assign()
        s.deactivate_device(device=self.device)
        s.reactivate_device(device=self.device)
        self.assertEqual(q.current_ownership_relationship(self.device), row)

    def test_catalog_inactive_does_not_rewrite_or_prevent_ownership(self):
        from apps.catalog.services import deactivate_brand
        row = self.assign()
        deactivate_brand(brand=self.brand)
        self.assertEqual(q.current_ownership_relationship(self.device), row)
        s.transfer_device_ownership(device=self.device, new_customer=self.b)

    def test_identifier_correction_preserves_ownership(self):
        row = self.assign()
        s.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-OWNERSHIP-001")
        self.assertEqual(q.current_ownership_relationship(self.device), row)
        self.assertEqual(q.find_device_by_identifier("SYNTHETIC-OWNERSHIP-001"), self.device)

    def test_purchase_and_warranty_independence(self):
        from .test_evidence import coverage
        evidence = s.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-ONLY")
        warranty = coverage(self.device)
        before = (evidence.updated_at, warranty.updated_at)
        self.assign()
        s.transfer_device_ownership(device=self.device, new_customer=self.b)
        s.end_device_ownership(device=self.device)
        evidence.refresh_from_db(); warranty.refresh_from_db()
        self.assertEqual((evidence.updated_at, warranty.updated_at), before)
        self.assertEqual(q.current_warranty_coverage(self.device), warranty)

    def test_model_save_delete_and_queryset_delete_rejected(self):
        row = self.assign()
        for operation in [row.save, row.delete, R.objects.all().delete]:
            with self.assertRaises(ValidationError):
                operation()

    def test_immutable_current_and_history(self):
        row = self.assign()
        row.customer = self.b
        with self.assertRaises(ValidationError):
            row.full_clean()
        closed = s.end_device_ownership(device=self.device)
        for field, value in [("note", "changed"), ("end_reason", "changed"), ("ended_at", None), ("started_at", self.start-timedelta(days=1))]:
            closed.refresh_from_db()
            setattr(closed, field, value)
            with self.subTest(field=field), self.assertRaises(ValidationError):
                closed.full_clean()

    def test_database_constraints(self):
        self.assign()
        invalid = [dict(), dict(relationship_type="OTHER"), dict(started_at=self.start, ended_at=self.start-timedelta(days=1)),
                   dict(started_at=self.start+timedelta(days=1), ended_at=self.start+timedelta(days=2))]
        for changes in invalid:
            with self.subTest(changes=changes), self.assertRaises(IntegrityError), transaction.atomic():
                R.objects.bulk_create([R(**dict(dict(device=self.device, customer=self.b, started_at=self.start), **changes))])

    def test_historical_database_overlap(self):
        self.assign()
        s.end_device_ownership(device=self.device, ended_at=self.start+timedelta(days=3))
        with self.assertRaises(IntegrityError), transaction.atomic():
            R.objects.bulk_create([R(device=self.device, customer=self.b, started_at=self.start+timedelta(days=1), ended_at=self.start+timedelta(days=2))])

    def test_assign_rollback(self):
        with self.assertRaises(RuntimeError), transaction.atomic():
            self.assign()
            raise RuntimeError("Synthetic failure")
        self.assertFalse(R.objects.exists())

    def test_end_rollback(self):
        row = self.assign()
        with self.assertRaises(RuntimeError), transaction.atomic():
            s.end_device_ownership(device=self.device)
            raise RuntimeError("Synthetic failure")
        self.assertEqual(q.current_ownership_relationship(self.device), row)

    def test_transfer_insert_failure_restores_old_owner(self):
        row = self.assign()
        with patch("apps.devices.relationship_services._append", side_effect=ValidationError("Synthetic failure")):
            with self.assertRaises(ValidationError):
                s.transfer_device_ownership(device=self.device, new_customer=self.b)
        self.assertEqual(q.current_ownership_relationship(self.device), row)
        self.assertEqual(R.objects.count(), 1)

    def test_history_query_has_no_n_plus_one(self):
        self.assign()
        s.transfer_device_ownership(device=self.device, new_customer=self.b)
        with self.assertNumQueries(1):
            for row in q.device_ownership_history(self.device):
                str(row.customer.company); str(row.device)

    def test_protected_customer_and_device_foreign_keys(self):
        from django.db.models.deletion import Collector, ProtectedError
        self.assign()
        for obj in (self.a, self.device):
            with self.subTest(model=type(obj).__name__), self.assertRaises(ProtectedError):
                Collector(using="default").collect([obj])

    def test_type_and_period_constraints_without_competing_owner(self):
        for changes in [dict(relationship_type="INVALID"), dict(ended_at=self.start-timedelta(seconds=1))]:
            with self.subTest(changes=changes), self.assertRaises(IntegrityError), transaction.atomic():
                R.objects.bulk_create([R(**dict(dict(device=self.device, customer=self.a, started_at=self.start), **changes))])

    def test_zero_length_period_is_valid(self):
        first = self.assign()
        s.end_device_ownership(device=self.device, ended_at=self.start)
        second = s.assign_device_owner(device=self.device, customer=self.b, started_at=self.start)
        self.assertEqual(list(q.device_ownership_history(self.device).values_list("pk", flat=True)), sorted([first.pk, second.pk]))

    def test_transfer_to_inactive_customer_and_device_rejected(self):
        self.assign()
        customers.deactivate_customer(customer=self.b)
        with self.assertRaises(ValidationError):
            s.transfer_device_ownership(device=self.device, new_customer=self.b)
        s.deactivate_device(device=self.device)
        with self.assertRaises(ValidationError):
            s.transfer_device_ownership(device=self.device, new_customer=self.c)
        self.assertEqual(q.current_device_owner(self.device), self.a)

    def test_company_reactivation_preserves_record(self):
        row = self.assign()
        organizations.deactivate_company(company=self.company)
        self.assertFalse(q.currently_owned_devices_for_company(self.company).exists())
        organizations.reactivate_company(company=self.company)
        self.assertEqual(q.current_ownership_relationship(self.device), row)
        self.assertEqual(list(q.currently_owned_devices_for_company(self.company)), [self.device])

    def test_model_invalid_type_and_naive_timestamp(self):
        for changes in [dict(relationship_type="INVALID"), dict(started_at=datetime(2025, 1, 1)), dict(ended_at=self.start-timedelta(seconds=1))]:
            row = R(**dict(dict(device=self.device, customer=self.a, started_at=self.start), **changes))
            with self.subTest(changes=changes), self.assertRaises(ValidationError):
                row.full_clean()


class OwnershipAdminTests(TestCase):
    def setUp(self):
        setup(self)
        self.user = get_user_model().objects.create_superuser(username="synthetic-ownership-admin", password="synthetic-only")
        self.client.force_login(self.user)

    def url(self, row):
        return reverse("admin:devices_customerdevicerelationship_change", args=[row.pk])

    def test_admin_add(self):
        response = self.client.post(reverse("admin:devices_customerdevicerelationship_add"), dict(device=self.device.pk, customer=self.a.pk, started_at="", note="Synthetic", _save="Save"))
        self.assertEqual(response.status_code, 302)
        self.assertEqual(q.current_device_owner(self.device), self.a)

    def test_admin_transfer_and_historical_readonly(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        response = self.client.post(self.url(row), dict(revision=revision_token(row), operation="transfer", new_customer=self.b.pk, _save="Save"))
        self.assertEqual(response.status_code, 302)
        self.assertEqual(q.current_device_owner(self.device), self.b)
        self.assertEqual(self.client.get(self.url(row)).status_code, 200)
        self.assertEqual(self.client.post(self.url(row), dict(operation="end")).status_code, 403)

    def test_admin_end(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        response = self.client.post(self.url(row), dict(revision=revision_token(row), operation="end", reason="Synthetic", _save="Save"))
        self.assertEqual(response.status_code, 302)
        self.assertIsNone(q.current_device_owner(self.device))

    def test_stale_admin_page_cannot_end_new_owner(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        page = self.client.get(self.url(row))
        self.assertEqual(page.status_code, 200)
        token = page.context["adminform"].form.initial["revision"]
        new = s.transfer_device_ownership(device=self.device, new_customer=self.b)
        self.client.post(self.url(row), dict(revision=token, operation="end", _save="Save"))
        self.assertEqual(q.current_ownership_relationship(self.device), new)

    def test_tampered_revision_and_forged_owner(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        response = self.client.post(self.url(row), dict(revision="bad", operation="end", customer=self.b.pk, _save="Save"))
        self.assertEqual(response.status_code, 200)
        self.assertEqual(q.current_device_owner(self.device), self.a)

    def test_staff_without_permissions_denied(self):
        user = get_user_model().objects.create_user(username="synthetic-no-access", is_staff=True)
        self.client.force_login(user)
        self.assertEqual(self.client.get(reverse("admin:devices_customerdevicerelationship_add")).status_code, 403)

    def test_admin_cross_company_transfer_rejected(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        self.client.post(self.url(row), dict(revision=revision_token(row), operation="transfer", new_customer=self.outsider.pk, _save="Save"))
        self.assertEqual(q.current_device_owner(self.device), self.a)
        self.assertEqual(R.objects.count(), 1)

    def test_admin_delete_denied(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        response = self.client.post(reverse("admin:devices_customerdevicerelationship_delete", args=[row.pk]), {"post": "yes"})
        self.assertEqual(response.status_code, 403)
        self.assertTrue(R.objects.filter(pk=row.pk).exists())

    def test_admin_precondition_handles_transfer_after_form_validation(self):
        row = s.assign_device_owner(device=self.device, customer=self.a)
        original = s.end_device_ownership

        def changed_before_lock(**kwargs):
            s.transfer_device_ownership(device=self.device, new_customer=self.b)
            return original(**kwargs)

        with patch("apps.devices.relationship_admin.services.end_device_ownership", side_effect=changed_before_lock):
            response = self.client.post(self.url(row), dict(revision=revision_token(row), operation="end", _save="Save"))
        self.assertEqual(response.status_code, 302)
        # Both writes are in the admin transaction in this injected scenario;
        # stale precondition rejection rolls the whole transaction back.
        self.assertEqual(q.current_device_owner(self.device), self.a)
        self.assertEqual(R.objects.count(), 1)


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

    def setUp(self):
        setup(self)

    def assign(self):
        return s.assign_device_owner(device=self.device, customer=self.a, started_at=self.start)

    def test_two_initial_assignments(self):
        self.run_concurrent(self.assign, lambda: s.assign_device_owner(device=self.device, customer=self.b), expected="validation")
        self.assertEqual(R.objects.filter(ended_at__isnull=True).count(), 1)

    def test_cross_company_initial_assignments(self):
        self.run_concurrent(self.assign, lambda: s.assign_device_owner(device=self.device, customer=self.outsider), expected="validation")

    def test_competing_transfers_with_precondition(self):
        row = self.assign()
        self.run_concurrent(lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b, expected_current_relationship_id=row.pk),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.c, expected_current_relationship_id=row.pk), expected="validation")
        self.assertEqual(q.current_device_owner(self.device), self.b)
        self.assertEqual(R.objects.count(), 2)

    def test_transfers_without_precondition_serialize(self):
        self.assign()
        self.run_concurrent(lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.c), expected="success")
        self.assertEqual(q.current_device_owner(self.device), self.c)
        self.assertEqual(R.objects.count(), 3)

    def test_end_then_transfer(self):
        self.assign()
        self.run_concurrent(lambda: s.end_device_ownership(device=self.device),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b), expected="validation")

    def test_transfer_then_stale_end(self):
        row = self.assign()
        self.run_concurrent(lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b),
                            lambda: s.end_device_ownership(device=self.device, expected_current_relationship_id=row.pk), expected="validation")
        self.assertEqual(q.current_device_owner(self.device), self.b)

    def test_customer_deactivate_then_assignment(self):
        self.run_concurrent(lambda: customers.deactivate_customer(customer=self.a), self.assign, expected="validation")

    def test_company_deactivate_then_assignment(self):
        self.run_concurrent(lambda: organizations.deactivate_company(company=self.company), self.assign, expected="validation")

    def test_device_deactivate_then_assignment(self):
        self.run_concurrent(lambda: s.deactivate_device(device=self.device), self.assign, expected="validation")

    def test_target_customer_deactivate_then_transfer(self):
        self.assign()
        self.run_concurrent(lambda: customers.deactivate_customer(customer=self.b),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b), expected="validation")

    def test_assignment_then_customer_deactivation_preserves_history(self):
        self.run_concurrent(self.assign, lambda: customers.deactivate_customer(customer=self.a), expected="success")
        self.assertEqual(q.current_device_owner(self.device), self.a)
        self.assertFalse(q.currently_owned_devices(self.a).exists())

    def test_end_then_backdated_assignment(self):
        self.assign()
        self.run_concurrent(lambda: s.end_device_ownership(device=self.device),
                            lambda: s.assign_device_owner(device=self.device, customer=self.b, started_at=self.start), expected="validation")

    def test_company_deactivate_then_transfer(self):
        self.assign()
        self.run_concurrent(lambda: organizations.deactivate_company(company=self.company),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b), expected="validation")

    def test_device_deactivate_then_transfer(self):
        self.assign()
        self.run_concurrent(lambda: s.deactivate_device(device=self.device),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b), expected="validation")

    def test_cross_company_assignment_after_end_race(self):
        self.assign()
        self.run_concurrent(lambda: s.end_device_ownership(device=self.device),
                            lambda: s.assign_device_owner(device=self.device, customer=self.outsider), expected="validation")

    def test_backdated_transfers_cannot_reverse_timeline(self):
        self.assign()
        self.run_concurrent(lambda: s.transfer_device_ownership(device=self.device, new_customer=self.b, transferred_at=self.start+timedelta(days=2)),
                            lambda: s.transfer_device_ownership(device=self.device, new_customer=self.c, transferred_at=self.start+timedelta(days=1)), expected="validation")
        self.assertEqual(q.current_device_owner(self.device), self.b)

    def test_assignment_then_device_deactivation_preserves_history(self):
        self.run_concurrent(self.assign, lambda: s.deactivate_device(device=self.device), expected="success")
        self.assertEqual(q.current_device_owner(self.device), self.a)
        self.assertFalse(q.currently_owned_devices(self.a).exists())

    def test_assignment_then_company_deactivation_preserves_history(self):
        self.run_concurrent(self.assign, lambda: organizations.deactivate_company(company=self.company), expected="success")
        self.assertEqual(q.current_device_owner(self.device), self.a)
        self.assertFalse(q.currently_owned_devices(self.a).exists())