"""Atomic replacement and repair namespace race tests."""

from __future__ import annotations

import fcntl
import os
import signal
from datetime import date, datetime
from pathlib import Path
from unittest import TestCase
from unittest.mock import patch

import checkin_cli.weekly_operations_repair_publish as repair_publish
from checkin_cli.weekly_operations import CanonicalPin, CustomerKey, DayState, WeeklyOperationInput, WeeklyOperationsAuthorityCompromise, WeeklyOperationsCorruption, customer_identity_digest
from checkin_cli.weekly_operations_authority import AuthorityId, issue_repair_authority
from checkin_cli.weekly_operations_layout import customer_data_name, customer_lock_name
from checkin_cli.weekly_operations_store import WeeklyOperationsStore
from tests._weekly_operations_support import initialize_at

CUSTOMER = CustomerKey("client_001")
ASSERTIONS = TestCase()


def _operation() -> WeeklyOperationInput:
    return WeeklyOperationInput.for_customer(CUSTOMER, date(2026, 8, 17), DayState.SUBMITTED, CanonicalPin(1, "a" * 64), datetime.fromisoformat("2026-08-17T20:00:00+09:00"))


def _setup(tmp_path: Path):
    path = tmp_path / "authority"
    authority = initialize_at(path, AuthorityId("1" * 64))
    store = WeeklyOperationsStore.for_authority(authority, CUSTOMER)
    _ = store.append(_operation())
    data = path / customer_data_name(customer_identity_digest(CUSTOMER))
    lock = path / customer_lock_name(customer_identity_digest(CUSTOMER))
    with data.open("ab") as handle:
        _ = handle.write(b"torn-tail")
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    return path, data, lock, repair_authority, repair_store


def test_temp_name_substitution_never_deletes_unowned_or_returns_success(tmp_path: Path) -> None:
    path, data, _, _, store = _setup(tmp_path)
    customers = path
    stolen = customers / "retained-repair-evidence"
    real_replace = os.replace

    def substitute_temp(source: str, destination: str, *, src_dir_fd: int | None = None, dst_dir_fd: int | None = None) -> None:
        real_replace(source, stolen.name, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd)
        replacement = os.open(source, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600, dir_fd=src_dir_fd)
        try:
            _ = os.write(replacement, b"unowned-substitute")
            os.fsync(replacement)
        finally:
            os.close(replacement)
        real_replace(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd)

    with patch("os.replace", side_effect=substitute_temp):
        with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
            _ = store.repair_torn_tail()
    assert stolen.is_file() and data.read_bytes() == b"unowned-substitute"


def test_temp_cleanup_name_substitution_preserves_replacement(tmp_path: Path) -> None:
    path, data, _, _, store = _setup(tmp_path)
    customers = path
    original = data.read_bytes()
    stolen = customers / "moved-repair-temp"

    def substitute_then_fail(descriptor: int, payload: bytes) -> None:
        _ = os.write(descriptor, payload)
        temp = next(customers.glob(".weekly-repair-*.tmp"))
        _ = temp.replace(stolen)
        temp.touch(mode=0o600)
        raise OSError("injected repair temp cleanup substitution")

    with patch.object(repair_publish, "_write_all", side_effect=substitute_then_fail):
        with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
            _ = store.repair_torn_tail()
    assert stolen.exists() and tuple(customers.glob(".weekly-repair-*.tmp"))
    assert data.read_bytes() == original


def test_postreplace_destination_substitution_is_typed_incident(tmp_path: Path) -> None:
    _path, data, _, _, store = _setup(tmp_path)
    original = data.read_bytes()
    stolen = tmp_path / "published-repair-evidence"
    calls = 0
    real_replace = os.replace

    def substitute_reconcile(_customers: int, _name: str, _expected: tuple[int, int], _prefix: bytes) -> None:
        nonlocal calls
        calls += 1
        if calls == 1:
            real_replace(data, stolen)
            _ = data.write_bytes(b"unowned-canonical")
            data.chmod(0o600)
        raise WeeklyOperationsAuthorityCompromise("injected postreplace destination substitution")

    with patch.object(repair_publish, "_reconcile", side_effect=substitute_reconcile):
        with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
            _ = store.repair_torn_tail()
    assert stolen.read_bytes() == original[: -len(b"torn-tail")]
    assert data.read_bytes() == b"unowned-canonical"


def test_replace_report_failure_after_commit_reconciles_success(tmp_path: Path) -> None:
    _path, data, _, repair_authority, store = _setup(tmp_path)
    binding = repair_authority.repair_binding
    assert binding is not None
    original = data.read_bytes()
    real_replace = os.replace

    def replace_then_fail(source: str, destination: str, *, src_dir_fd: int | None = None, dst_dir_fd: int | None = None) -> None:
        real_replace(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd)
        raise OSError("injected report failure after committed replacement")

    with patch("os.replace", side_effect=replace_then_fail):
        repaired = store.repair_torn_tail()
    assert data.read_bytes() == original[: binding.valid_prefix_offset]
    assert repaired.removed_bytes == binding.torn_tail_length


def test_sigterm_cleans_owned_temp_without_touching_data(tmp_path: Path) -> None:
    path, data, _, _, store = _setup(tmp_path)
    original = data.read_bytes()

    def terminate(_descriptor: int, _payload: bytes) -> None:
        os.kill(os.getpid(), signal.SIGTERM)

    with patch.object(repair_publish, "_write_all", side_effect=terminate):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "terminated"):
            _ = store.repair_torn_tail()
    assert data.read_bytes() == original
    assert not tuple((path).glob(".weekly-repair-*.tmp"))


def test_lock_bytes_mutated_after_flock_are_rejected_before_replace(tmp_path: Path) -> None:
    _, data, lock, _, store = _setup(tmp_path)
    original = data.read_bytes()
    lock_inode = lock.stat().st_ino
    real_flock = fcntl.flock

    def mutate_after_lock(descriptor: int, operation: int) -> None:
        real_flock(descriptor, operation)
        if operation & fcntl.LOCK_EX and os.fstat(descriptor).st_ino == lock_inode:
            _ = os.lseek(descriptor, 0, os.SEEK_SET)
            _ = os.write(descriptor, b"mutated-under-lock")
            os.fsync(descriptor)

    with patch("fcntl.flock", side_effect=mutate_after_lock):
        with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
            _ = store.repair_torn_tail()
    assert data.read_bytes() == original