"""Unnamed marker publication races and removed-stage proof."""

from __future__ import annotations

import os
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Barrier
from unittest import TestCase
from unittest.mock import patch

import checkin_cli.weekly_operations_publish as publisher
from checkin_cli.weekly_operations import WeeklyOperationsAlreadyInitialized, WeeklyOperationsAuthorityCompromise, WeeklyOperationsCorruption
from checkin_cli.weekly_operations_authority import AuthorityId, WeeklyOperationsAuthorityRoot
from checkin_cli.weekly_operations_canonical_registry_history import REGISTRY_NAME
from checkin_cli.weekly_operations_layout import MARKER_NAME
from checkin_cli.weekly_operations_link import link_unnamed_file
from tests._weekly_operations_support import acquire_for_child, initialize_at, initialize_for_parent

ASSERTIONS = TestCase()
InitOutcome = WeeklyOperationsAuthorityRoot | OSError | WeeklyOperationsCorruption


def test_concurrent_empty_root_initializers_race_at_link_one_typed_winner(tmp_path: Path) -> None:
    for run in range(4):
        root = tmp_path / f"authority-{run}"
        root.mkdir(mode=0o700)
        start = Barrier(2)

        def initialize(identity: AuthorityId) -> InitOutcome:
            parent = acquire_for_child(root)
            _ = start.wait(timeout=5)
            try:
                return initialize_for_parent(parent, identity)
            except (OSError, WeeklyOperationsCorruption) as error:
                return error

        with ThreadPoolExecutor(max_workers=2) as executor:
            outcomes = tuple(executor.map(initialize, (AuthorityId("1" * 64), AuthorityId("2" * 64))))
        successes = tuple(value for value in outcomes if isinstance(value, WeeklyOperationsAuthorityRoot))
        failures = tuple(value for value in outcomes if isinstance(value, WeeklyOperationsAlreadyInitialized))
        assert len(successes) == len(failures) == 1
        assert {path.name for path in root.iterdir()} == {MARKER_NAME, REGISTRY_NAME}
        successes[0].close()


def test_r10_stage_mkdir_substitution_seam_is_structurally_unreachable(tmp_path: Path) -> None:
    root = tmp_path / "authority"
    root.mkdir(mode=0o700)
    calls: list[str | os.PathLike[str]] = []
    real_mkdir = os.mkdir

    def forbidden_mkdir(path: str | os.PathLike[str], mode: int = 0o777, *, dir_fd: int | None = None) -> None:
        calls.append(path)
        real_mkdir(path, mode, dir_fd=dir_fd)

    with patch("os.mkdir", side_effect=forbidden_mkdir):
        authority = initialize_at(root, AuthorityId("3" * 64))
    assert calls == [] and not tuple(root.glob(".weekly-operations-init-*"))
    assert not hasattr(publisher, "_cleanup_stage") and not hasattr(publisher, "_verify_created_stage")
    authority.close()


def test_marker_symlink_preoccupation_is_preserved_without_target_mutation(tmp_path: Path) -> None:
    root = tmp_path / "authority"
    root.mkdir(mode=0o700)
    target = tmp_path / "outside-marker"
    _ = target.write_bytes(b"outside")
    target.chmod(0o600)
    (root / MARKER_NAME).symlink_to(target)
    with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
        _ = initialize_at(root, AuthorityId("4" * 64))
    assert target.read_bytes() == b"outside" and (root / MARKER_NAME).is_symlink()


def test_postlink_marker_substitution_never_returns_binding_or_deletes_unowned(tmp_path: Path) -> None:
    root = tmp_path / "authority"
    root.mkdir(mode=0o700)
    stolen = tmp_path / "retained-linked-marker"
    real_reconcile = publisher.reconcile_owned_marker
    substituted = False

    def substitute_then_reconcile(directory: int, expected: tuple[int, int], payload: bytes):
        nonlocal substituted
        if not substituted:
            substituted = True
            _ = (root / MARKER_NAME).replace(stolen)
            _ = (root / MARKER_NAME).write_bytes(b"unowned-marker")
            (root / MARKER_NAME).chmod(0o600)
        return real_reconcile(directory, expected, payload)

    with patch.object(publisher, "reconcile_owned_marker", side_effect=substitute_then_reconcile):
        with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
            _ = initialize_at(root, AuthorityId("5" * 64))
    assert stolen.is_file() and (root / MARKER_NAME).read_bytes() == b"unowned-marker"


def test_ambiguous_link_exception_reconciles_only_exact_unnamed_inode(tmp_path: Path) -> None:
    root = tmp_path / "authority"
    root.mkdir(mode=0o700)

    def link_then_fail(source: int, directory: int, name: str) -> None:
        link_unnamed_file(source, directory, name)
        raise OSError("injected ambiguous post-link exception")

    with patch.object(publisher, "link_unnamed_file", side_effect=link_then_fail):
        authority = initialize_at(root, AuthorityId("6" * 64))
    assert authority.binding.marker_inode == (root / MARKER_NAME).stat().st_ino
    authority.close()


def test_valid_existing_marker_is_typed_already_initialized(tmp_path: Path) -> None:
    root = tmp_path / "authority"
    root.mkdir(mode=0o700)
    authority = initialize_at(root, AuthorityId("7" * 64))
    marker = (root / MARKER_NAME).read_bytes()
    with ASSERTIONS.assertRaises(WeeklyOperationsAlreadyInitialized):
        _ = initialize_at(root, AuthorityId("8" * 64))
    assert (root / MARKER_NAME).read_bytes() == marker
    authority.close()