import csv
import json
from concurrent.futures import ThreadPoolExecutor
from io import BytesIO, StringIO
from threading import Barrier
from unittest.mock import patch

from django.contrib.auth.models import User
from django.core.exceptions import ValidationError
from django.core.management import call_command
from django.db import close_old_connections
from django.test import Client, TestCase, TransactionTestCase, skipUnlessDBFeature
from django.urls import reverse
from django.utils import timezone
from openpyxl import load_workbook

from .models import AuditLog, ChoiceValue, Device, Employee, InventoryAPIKey, UserPreference
from .services import coerce_inline_value


class SecurityRegressionTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.user = User.objects.create_user("security-admin", is_superuser=True)
        cls.status = ChoiceValue.objects.create(kind="status", label="Test")

    def setUp(self):
        self.client.force_login(self.user)
        self.device = Device.objects.create(status=self.status, ram_gb=4)

    def post_inline(self, payload):
        return self.client.post(reverse("inventory:inline-update", args=[self.device.pk]),
                                json.dumps(payload), content_type="application/json")

    def test_revisions_and_legacy_timestamps_change_on_a_frozen_clock(self):
        old_time = self.device.updated_at
        with patch("inventory.models.timezone.now", return_value=old_time):
            self.device.ram_gb = 8
            self.device.save(update_fields=["ram_gb"])
        self.assertEqual(self.device.revision, 2)
        self.assertGreater(self.device.updated_at, old_time)
        response = self.post_inline({"field": "ram_gb", "value": 16, "expected_revision": 1})
        self.assertEqual(response.status_code, 409)
        response = self.post_inline({"field": "ram_gb", "value": 16, "expected_updated_at": old_time.isoformat()})
        self.assertEqual(response.status_code, 409)
        self.device.refresh_from_db()
        self.assertEqual(self.device.ram_gb, 8)

    def test_detail_rejects_stale_and_missing_versions(self):
        self.device.ram_gb = 8
        self.device.save()
        for token in ({"expected_revision": 1}, {}):
            response = self.client.post(reverse("inventory:device-detail", args=[self.device.pk]), {
                "status": self.status.pk, "date": "23.09.2026", "ram_gb": 16,
                "battery": "UNKNOWN", "wifi": "UNKNOWN", "lan": "UNKNOWN", **token,
            })
            self.assertEqual(response.status_code, 302)
            self.device.refresh_from_db()
            self.assertEqual(self.device.ram_gb, 8)
            self.assertFalse(AuditLog.objects.filter(device=self.device).exists())

    def test_malformed_inline_payloads_return_400(self):
        for payload in ([], None, {"field": "ram_gb", "value": 8},
                        {"field": [], "value": 8, "expected_revision": 1},
                        {"field": "price", "value": "NaN", "expected_revision": 1},
                        {"field": "price", "value": "Infinity", "expected_revision": 1}):
            with self.subTest(payload=payload):
                self.assertEqual(self.post_inline(payload).status_code, 400)

    def test_integer_fields_do_not_silently_truncate(self):
        for value in (1.5, "NaN", "Infinity", True):
            with self.subTest(value=value), self.assertRaises(ValidationError):
                coerce_inline_value("ram_gb", value)

    def test_invalid_filters_do_not_raise_server_errors(self):
        for query in ("status=abc", "ram_min=abc", "price_min=NaN", "price_min=Infinity", "employee=9999999999999999999999999"):
            with self.subTest(query=query):
                response = self.client.get(reverse("inventory:device-list") + "?" + query)
                self.assertEqual(response.status_code, 200)
                self.assertEqual(response.context["page_obj"].paginator.count, 0)

    def test_invalid_preferences_are_rejected_without_persisting(self):
        for payload in ([], {"page_size": []}, {"column_order": "number"},
                        {"visible_columns": ["<script>"]}, {"column_widths": {"number": "wide"}}):
            response = self.client.post(reverse("inventory:save-preferences"), json.dumps(payload), content_type="application/json")
            self.assertEqual(response.status_code, 400)
        self.assertFalse(UserPreference.objects.filter(user=self.user).exists())

    def test_exports_preserve_formula_like_text_without_executing_it(self):
        self.device.comment = '=HYPERLINK("https://example.invalid", "test")'
        self.device.model = "  +1+1"
        self.device.save()
        response = self.client.get(reverse("inventory:export-devices", args=["xlsx"]))
        workbook = load_workbook(BytesIO(response.content))
        row = dict(zip([cell.value for cell in workbook.active[1]], workbook.active[2]))
        self.assertEqual(row["Kommentar"].data_type, "s")
        self.assertEqual(row["Kommentar"].value, self.device.comment)
        self.assertEqual(row["RAM"].data_type, "n")
        response = self.client.get(reverse("inventory:export-devices", args=["csv"]))
        row = next(csv.DictReader(StringIO(response.content.decode("utf-8-sig")), delimiter=";"))
        self.assertEqual(row["Kommentar"], "'" + self.device.comment)
        self.assertEqual(row["Modell"], "'" + self.device.model)

    def test_autocomplete_includes_choices_beyond_first_fifteen(self):
        Employee.objects.bulk_create([Employee(code=f"E{i:02}", name=f"Employee {i}") for i in range(25)])
        response = self.client.get(reverse("inventory:autocomplete", args=["employee"]))
        self.assertEqual(len(response.json()["results"]), 25)

    def test_seed_preserves_administrator_changes(self):
        call_command("seed_initial_data", verbosity=0)
        status = ChoiceValue.objects.get(kind="status", label="Eingang")
        status.color, status.active = "#123456", False
        status.save()
        employee = Employee.objects.get(code="JH")
        employee.name, employee.active = "Changed name", False
        employee.save()
        call_command("seed_initial_data", verbosity=0)
        status.refresh_from_db()
        employee.refresh_from_db()
        self.assertEqual(status.color, "#123456")
        self.assertFalse(status.active)
        self.assertEqual(employee.name, "Changed name")
        self.assertFalse(employee.active)

    def test_api_supports_revisions_and_rejects_stale_revision(self):
        _, key = InventoryAPIKey.objects.create_key(name="test", service_user=self.user,
            can_read=True, can_write=True, expiry_date=timezone.now() + timezone.timedelta(days=1))
        url = f"/api/v1/devices/{self.device.number}/"
        for expected_status in (200, 409):
            response = self.client.patch(url, json.dumps({"ram_gb": 8, "expected_revision": 1}),
                content_type="application/json", HTTP_AUTHORIZATION=f"Api-Key {key}")
            self.assertEqual(response.status_code, expected_status)


class PostgreSQLConcurrencyTests(TransactionTestCase):
    @skipUnlessDBFeature("has_select_for_update")
    def test_two_detail_saves_with_same_revision_only_commit_once(self):
        user = User.objects.create_user("concurrent-admin", is_superuser=True)
        status = ChoiceValue.objects.create(kind="status", label="Test")
        device = Device.objects.create(status=status, ram_gb=4)
        clients = [Client(), Client()]
        for client in clients:
            client.force_login(user)
        barrier = Barrier(2)

        def save(index):
            close_old_connections()
            try:
                barrier.wait(timeout=10)
                return clients[index].post(reverse("inventory:device-detail", args=[device.pk]), {
                    "status": status.pk, "date": device.date.isoformat(), "ram_gb": 8 + index,
                    "battery": "UNKNOWN", "wifi": "UNKNOWN", "lan": "UNKNOWN", "expected_revision": 1,
                }).status_code
            finally:
                close_old_connections()

        with ThreadPoolExecutor(max_workers=2) as executor:
            self.assertEqual(list(executor.map(save, range(2))), [302, 302])
        device.refresh_from_db()
        self.assertEqual(device.revision, 2)
        self.assertEqual(AuditLog.objects.filter(device=device, field_key="ram_gb").count(), 1)