# Copyright (c) 2023-2026, HaiyangLi <quantocean.li at gmail dot com>
# SPDX-License-Identifier: Apache-2.0

"""Regression tests for symlink containment, HookRegistry aliases, and related CLI behaviors."""

from __future__ import annotations

from pathlib import Path

from lionagi.cli._logging import _LazyStderrHandler
from lionagi.cli._providers import _clamp_claude_effort
from lionagi.cli.main import _handle_play_shortcut
from lionagi.cli.orchestrate import _resolve_playbook_path, _validate_spec_fields
from lionagi.cli.skill import resolve_skill_path
from lionagi.service.hooks._types import HookEventTypes
from lionagi.service.hooks.hook_registry import HookRegistry

# ── Symlink containment ────────────────────────────────


class TestSkillSymlinkContainment:
    def test_rejects_skill_md_symlink_pointing_outside_root(self, monkeypatch, tmp_path: Path):
        """A SKILL.md inside the skills root that is itself a symlink to
        an arbitrary file must be rejected before read_text() follows it.
        """
        secret = tmp_path / "secret.txt"
        secret.write_text("DO NOT LEAK")

        home = tmp_path / "home"
        skills = home / ".lionagi" / "skills" / "leak"
        skills.mkdir(parents=True)
        # Symlink SKILL.md -> secret.txt, simulating the exploit vector.
        (skills / "SKILL.md").symlink_to(secret)

        monkeypatch.setenv("HOME", str(home))

        path, err = resolve_skill_path("leak")
        assert path is None
        assert err is not None
        assert "symlink escape" in err or "outside" in err

    def test_accepts_legitimate_skill(self, monkeypatch, tmp_path: Path):
        home = tmp_path / "home"
        skills = home / ".lionagi" / "skills" / "ok"
        skills.mkdir(parents=True)
        (skills / "SKILL.md").write_text("---\nname: ok\ndescription: legit\n---\nbody")
        monkeypatch.setenv("HOME", str(home))

        path, err = resolve_skill_path("ok")
        assert err is None
        assert path is not None
        assert path.read_text().startswith("---")

    def test_accepts_root_symlink(self, monkeypatch, tmp_path: Path):
        """The skills *root* itself may be a symlink (users point it at
        any directory they manage); the resolve check must accept that.
        """
        real_skills = tmp_path / "real" / "skills"
        (real_skills / "ok").mkdir(parents=True)
        (real_skills / "ok" / "SKILL.md").write_text("---\nname: ok\n---\nbody")
        home = tmp_path / "home"
        (home / ".lionagi").mkdir(parents=True)
        (home / ".lionagi" / "skills").symlink_to(real_skills)
        monkeypatch.setenv("HOME", str(home))

        path, err = resolve_skill_path("ok")
        assert err is None
        assert path is not None


class TestPlaybookSymlinkContainment:
    def test_rejects_playbook_symlink_pointing_outside_root(self, monkeypatch, tmp_path: Path):
        secret = tmp_path / "secret.yaml"
        secret.write_text("evil: true\n")

        home = tmp_path / "home"
        playbooks = home / ".lionagi" / "playbooks"
        playbooks.mkdir(parents=True)
        (playbooks / "leak.playbook.yaml").symlink_to(secret)

        monkeypatch.setenv("HOME", str(home))

        path, err = _resolve_playbook_path("leak")
        assert path is None
        assert err is not None
        assert "symlink escape" in err or "outside" in err


# ── HookRegistry alias both spellings ──────────────────


class TestHookRegistryAliases:
    def test_pre_event_create_accepted(self):
        def hook(*a, **kw):
            pass

        reg = HookRegistry(hooks={"pre_event_create": hook})
        assert HookEventTypes.PreEventCreate in reg._hooks

    def test_pre_event_create_hook_accepted(self):
        """Legacy alias — constructor decorator method is called
        `pre_event_create_hook`, so callers often use this spelling.
        """

        def hook(*a, **kw):
            pass

        reg = HookRegistry(hooks={"pre_event_create_hook": hook})
        assert HookEventTypes.PreEventCreate in reg._hooks


# ── max_ops/max_agents 0 = unlimited ────────────────────


class TestMaxOpsZeroUnlimited:
    def test_max_ops_zero_accepted(self):
        assert _validate_spec_fields({"max_ops": 0}) is None

    def test_max_agents_zero_accepted(self):
        assert _validate_spec_fields({"max_agents": 0}) is None

    def test_negative_still_rejected(self):
        err = _validate_spec_fields({"max_ops": -1})
        assert err is not None

    def test_too_large_still_rejected(self):
        err = _validate_spec_fields({"max_ops": 51})
        assert err is not None


# ── _clamp_claude_effort coverage ────────────────────


class TestClampClaudeEffort:
    def test_xhigh_preserved_on_opus_4_7(self):
        assert _clamp_claude_effort("xhigh", "claude/claude-opus-4-7") == "xhigh"

    def test_xhigh_preserved_on_bare_opus_4_7(self):
        assert _clamp_claude_effort("xhigh", "opus-4-7") == "xhigh"

    def test_xhigh_preserved_on_bare_opus(self):
        assert _clamp_claude_effort("xhigh", "opus") == "xhigh"

    def test_xhigh_clamped_on_sonnet(self):
        assert _clamp_claude_effort("xhigh", "claude/claude-sonnet-4-6") == "high"

    def test_xhigh_clamped_on_haiku(self):
        assert _clamp_claude_effort("xhigh", "claude/claude-haiku-4-5") == "high"

    def test_non_xhigh_untouched_on_any_model(self):
        for effort in ("none", "minimal", "low", "medium", "high", "max"):
            assert _clamp_claude_effort(effort, "claude/claude-sonnet-4-6") == effort
            assert _clamp_claude_effort(effort, "claude/claude-opus-4-7") == effort


# ── _handle_play_shortcut coverage ───────────────────


class TestHandlePlayShortcut:
    def test_empty_argv_returns_unchanged(self):
        assert _handle_play_shortcut([]) == []

    def test_non_play_passthrough(self):
        argv = ["agent", "claude/sonnet", "hi"]
        assert _handle_play_shortcut(argv) == argv

    def test_play_no_args_prints_usage(self, capsys):
        code = _handle_play_shortcut(["play"])
        assert code == 1
        out = capsys.readouterr().out
        assert "Usage" in out

    def test_play_rewrite(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HOME", str(tmp_path))
        rewritten = _handle_play_shortcut(["play", "rewrite", "--tabs", "5", "query text"])
        assert rewritten == [
            "o",
            "flow",
            "-p",
            "rewrite",
            "--tabs",
            "5",
            "query text",
        ]

    def test_play_list_empty_dir(self, monkeypatch, tmp_path, capsys):
        monkeypatch.setenv("HOME", str(tmp_path))
        code = _handle_play_shortcut(["play", "list"])
        assert code == 0
        out = capsys.readouterr().out
        assert "no playbooks" in out.lower()

    def test_play_list_with_playbooks(self, monkeypatch, tmp_path, capsys):
        pb = tmp_path / ".lionagi" / "playbooks"
        pb.mkdir(parents=True)
        (pb / "alpha.playbook.yaml").write_text("prompt: a\n")
        (pb / "beta.playbook.yaml").write_text("prompt: b\n")
        monkeypatch.setenv("HOME", str(tmp_path))
        code = _handle_play_shortcut(["play", "list"])
        assert code == 0
        out = capsys.readouterr().out
        assert "alpha" in out
        assert "beta" in out

    def test_play_flag_before_name_errors(self):
        code = _handle_play_shortcut(["play", "--bogus", "foo"])
        assert code == 1


# ── _LazyStderrHandler re-binds stream ───────────────


class TestLazyStderrHandler:
    def test_emit_uses_current_stderr(self, capsys, monkeypatch):
        import logging

        handler = _LazyStderrHandler()
        handler.setFormatter(logging.Formatter("%(message)s"))

        logger = logging.getLogger("lionagi.cli._logging_test")
        logger.handlers.clear()
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
        logger.propagate = False

        logger.info("first-message")
        err1 = capsys.readouterr().err
        assert "first-message" in err1

        logger.info("second-message")
        err2 = capsys.readouterr().err
        assert "second-message" in err2
