from __future__ import annotations

import json
import os
import stat
import sys
import threading
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import account_lock
from account_lock import (
    ACCOUNT_LOCK_EXIT_CODE,
    AccountLockedError,
    AccountLockStateError,
    AccountLockStore,
    HumanOverrideError,
    HumanSessionVerdict,
    authorize_account_use,
    classify_human_session,
    confirm_human_action,
    confirmation_phrase,
    revalidate_account_use,
)


def test_store_round_trips_tool_scoped_locks_with_private_mode(tmp_path: Path) -> None:
    store = AccountLockStore(tmp_path)

    assert store.locked_keys() == frozenset()
    assert store.set_locked("codex", "work", True) is True
    assert store.set_locked("claude", "work", True) is True
    assert store.set_locked("codex", "work", True) is False

    assert store.locked_keys() == frozenset({"codex:work", "claude:work"})
    assert store.locked_slugs("codex") == frozenset({"work"})
    assert store.locked_slugs("claude") == frozenset({"work"})
    assert stat.S_IMODE(store.path.stat().st_mode) == 0o600
    assert json.loads(store.path.read_text(encoding="utf-8")) == {
        "version": "account-lock/v1",
        "locked": ["claude:work", "codex:work"],
    }


def test_store_serializes_concurrent_updates_without_losing_keys(tmp_path: Path) -> None:
    barrier = threading.Barrier(3)
    failures: list[BaseException] = []

    def lock(tool: str, slug: str) -> None:
        try:
            barrier.wait(timeout=5)
            AccountLockStore(tmp_path).set_locked(tool, slug, True)
        except BaseException as exc:  # noqa: BLE001
            failures.append(exc)

    first = threading.Thread(target=lock, args=("codex", "owner"))
    second = threading.Thread(target=lock, args=("claude", "agent"))
    first.start()
    second.start()
    barrier.wait(timeout=5)
    first.join(timeout=5)
    second.join(timeout=5)

    assert not first.is_alive()
    assert not second.is_alive()
    assert failures == []
    assert AccountLockStore(tmp_path).locked_keys() == frozenset(
        {"codex:owner", "claude:agent"}
    )


def test_store_rekeys_and_drops_locks(tmp_path: Path) -> None:
    store = AccountLockStore(tmp_path)
    store.set_locked("codex", "old", True)

    assert store.rekey("codex", "old", "new") is True
    assert store.is_locked("codex", "old") is False
    assert store.is_locked("codex", "new") is True
    assert store.drop("codex", "new") is True
    assert store.locked_keys() == frozenset()


def test_restore_key_states_preserves_unrelated_concurrent_updates(
    tmp_path: Path,
) -> None:
    store = AccountLockStore(tmp_path)
    store.set_locked("codex", "old", True)
    snapshot = {"codex:old": True, "codex:new": False}

    store.rekey("codex", "old", "new")
    store.set_locked("claude", "unrelated", True)
    assert store.restore_key_states(snapshot) is True

    assert store.locked_keys() == frozenset({"codex:old", "claude:unrelated"})


def test_store_rejects_malformed_or_unsafe_state(tmp_path: Path) -> None:
    store = AccountLockStore(tmp_path)
    store.path.write_text('{"version":"wrong","locked":[]}', encoding="utf-8")
    with pytest.raises(AccountLockStateError, match="version must be"):
        store.locked_keys()

    store.path.write_text(
        json.dumps({"version": "account-lock/v1", "locked": ["codex:work"]}),
        encoding="utf-8",
    )
    store.path.chmod(0o666)
    with pytest.raises(AccountLockStateError, match="group/world writable"):
        store.locked_keys()


def test_store_rejects_symlink_state(tmp_path: Path) -> None:
    target = tmp_path / "target.json"
    target.write_text(
        json.dumps({"version": "account-lock/v1", "locked": []}),
        encoding="utf-8",
    )
    store = AccountLockStore(tmp_path)
    store.path.symlink_to(target)

    with pytest.raises(AccountLockStateError, match="must not be a symlink"):
        store.locked_keys()


def test_human_session_classifier_is_agent_decisive_and_fail_closed() -> None:
    assert classify_human_session(
        ("/user.slice/human.slice", "/user.slice/agent.slice"),
        controlling_tty=True,
        container_signal=None,
    ) == HumanSessionVerdict(False, "agent-cgroup:/user.slice/agent.slice")
    assert classify_human_session(
        ("/user.slice/app-Dispatch.scope", "/user.slice/human.slice"),
        controlling_tty=True,
        container_signal=None,
    ).human is False
    assert classify_human_session(
        ("/user.slice/human.slice",),
        controlling_tty=False,
        container_signal=None,
    ).human is True
    assert classify_human_session(
        (), controlling_tty=True, container_signal="file:/.dockerenv"
    ).human is False
    assert classify_human_session(
        (), controlling_tty=True, container_signal=None
    ) == HumanSessionVerdict(True, "controlling-tty")
    assert classify_human_session(
        (), controlling_tty=False, container_signal=None
    ) == HumanSessionVerdict(False, "no-tty")


def test_human_confirmation_requires_exact_phrase() -> None:
    prompts: list[str] = []

    confirm_human_action(
        "unlock",
        "codex",
        "work",
        session_probe=lambda: HumanSessionVerdict(True, "controlling-tty"),
        confirmation_reader=lambda prompt: prompts.append(prompt) or "UNLOCK codex:work\n",
    )

    assert confirmation_phrase("use-once", "claude", "personal") == (
        "USE claude:personal ONCE"
    )
    assert prompts == ["Type 'UNLOCK codex:work' to confirm: "]

    with pytest.raises(HumanOverrideError, match="did not match"):
        confirm_human_action(
            "unlock",
            "codex",
            "work",
            session_probe=lambda: HumanSessionVerdict(True, "controlling-tty"),
            confirmation_reader=lambda _prompt: "unlock codex:work",
        )

    with pytest.raises(HumanOverrideError, match="requires a human terminal"):
        confirm_human_action(
            "unlock",
            "codex",
            "work",
            session_probe=lambda: HumanSessionVerdict(False, "agent-cgroup:/agent.slice"),
            confirmation_reader=lambda _prompt: pytest.fail("must not prompt an agent"),
        )


def test_authorization_denies_lock_and_allows_confirmed_one_shot(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    store = AccountLockStore(tmp_path)
    store.set_locked("codex", "work", True)

    with pytest.raises(AccountLockedError) as denied:
        authorize_account_use(
            tmp_path,
            "codex",
            "work",
            allow_locked=False,
            explicit_account=True,
        )
    assert denied.value.exit_code == ACCOUNT_LOCK_EXIT_CODE
    assert denied.value.refusal_payload()["accounts"] == ["codex:work"]

    with pytest.raises(HumanOverrideError, match="explicit --account"):
        authorize_account_use(
            tmp_path,
            "codex",
            "work",
            allow_locked=True,
            explicit_account=False,
        )

    confirmations: list[tuple[str, str, str]] = []
    monkeypatch.setattr(
        account_lock,
        "confirm_human_action",
        lambda action, tool, slug: confirmations.append((action, tool, slug)),
    )
    authorization = authorize_account_use(
        tmp_path,
        "codex",
        "work",
        allow_locked=True,
        explicit_account=True,
    )

    assert authorization.locked_override is True
    assert confirmations == [("use-once", "codex", "work")]
    assert store.is_locked("codex", "work") is True
    revalidate_account_use(tmp_path, authorization)


def test_override_revalidation_still_fails_closed_on_invalid_state(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    store = AccountLockStore(tmp_path)
    store.set_locked("codex", "work", True)
    monkeypatch.setattr(account_lock, "confirm_human_action", lambda *_args: None)
    authorization = authorize_account_use(
        tmp_path,
        "codex",
        "work",
        allow_locked=True,
        explicit_account=True,
    )
    store.path.write_text("not-json", encoding="utf-8")

    with pytest.raises(AccountLockStateError):
        revalidate_account_use(tmp_path, authorization)


def test_revalidation_blocks_a_race_that_locks_after_selection(tmp_path: Path) -> None:
    authorization = authorize_account_use(
        tmp_path,
        "claude",
        "work",
        allow_locked=False,
        explicit_account=True,
    )
    AccountLockStore(tmp_path).set_locked("claude", "work", True)

    with pytest.raises(AccountLockedError):
        revalidate_account_use(tmp_path, authorization)


def test_state_lock_file_is_private(tmp_path: Path) -> None:
    store = AccountLockStore(tmp_path)
    store.locked_keys()

    assert stat.S_IMODE(os.stat(store.lock_path).st_mode) == 0o600