"""Tests for the skill ``detail`` field (long-form annotation editable by
the skill owner or an admin).

Covers:
- ``SkillStore`` persists ``detail`` through create / ensure_legacy / update /
  update_admin, and owner-only ``update`` still refuses non-owners.
- ``_skill_to_response`` surfaces ``detail`` from the ownership record.
- ``SkillResponse`` / ``SkillDetailRequest`` model shapes.
"""

from __future__ import annotations

import json
import subprocess
import sys
import time
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.gateway.routers import mcp as mcp_router
from app.gateway.routers.skills import (
    SkillDetailRequest,
    SkillResponse,
    _can_view_skill_source,
    _skill_to_response,
)
from deerflow.client import DeerFlowClient
from deerflow.config import paths as paths_module
from deerflow.config.extensions_config import reset_extensions_config
from deerflow.persistence.skills import MemorySkillStore
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
from deerflow.skills.types import SkillCategory


@pytest.fixture
def isolated_paths(tmp_path, monkeypatch):
    """Redirect get_paths() at a fresh temp directory so usage lookups in
    ``_skill_to_response`` never touch real state."""
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    monkeypatch.setattr(paths_module, "_paths", None)
    yield tmp_path
    monkeypatch.setattr(paths_module, "_paths", None)


# ---------------------------------------------------------------------------
# SkillStore: detail persistence
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_skill_store_create_carries_detail() -> None:
    store = MemorySkillStore()
    created = await store.create(
        {"name": "alpha-tool", "owner_user_id": "user-a", "detail": "初始详情"}
    )
    assert created["detail"] == "初始详情"
    fetched = await store.get_any("alpha-tool")
    assert fetched is not None and fetched["detail"] == "初始详情"


@pytest.mark.asyncio
async def test_skill_store_create_defaults_detail_to_none() -> None:
    store = MemorySkillStore()
    created = await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
    assert created["detail"] is None


@pytest.mark.asyncio
async def test_skill_store_owner_can_set_and_clear_detail() -> None:
    store = MemorySkillStore()
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})

    updated = await store.update("alpha-tool", "user-a", {"detail": "技能用途说明"})
    assert updated is not None and updated["detail"] == "技能用途说明"

    # Clearing the annotation persists None.
    cleared = await store.update("alpha-tool", "user-a", {"detail": None})
    assert cleared is not None and cleared["detail"] is None


@pytest.mark.asyncio
async def test_skill_store_update_detail_rejects_non_owner() -> None:
    store = MemorySkillStore()
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})

    # A different user cannot write detail via the owner-scoped update.
    assert await store.update("alpha-tool", "user-b", {"detail": "x"}) is None
    record = await store.get_any("alpha-tool")
    assert record is not None and record["detail"] is None


@pytest.mark.asyncio
async def test_skill_store_admin_can_set_detail_on_any_skill() -> None:
    store = MemorySkillStore()
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})

    # Admin override writes detail regardless of ownership.
    updated = await store.update_admin("alpha-tool", {"detail": "管理员补充说明"})
    assert updated is not None and updated["detail"] == "管理员补充说明"


@pytest.mark.asyncio
async def test_skill_store_partial_update_preserves_detail() -> None:
    """Updating only ``published`` must not wipe an existing ``detail``."""
    store = MemorySkillStore()
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
    await store.update("alpha-tool", "user-a", {"detail": "保留我"})

    await store.update("alpha-tool", "user-a", {"published": True})
    record = await store.get_any("alpha-tool")
    assert record is not None
    assert record["detail"] == "保留我"
    assert record["published"] is True


@pytest.mark.asyncio
async def test_skill_store_ensure_legacy_carries_detail() -> None:
    store = MemorySkillStore()
    record = await store.ensure_legacy(
        {"name": "builtin-skill", "owner_user_id": None, "published": True, "detail": "内置技能说明"}
    )
    assert record["detail"] == "内置技能说明"


# ---------------------------------------------------------------------------
# Response mapping
# ---------------------------------------------------------------------------


def test_skill_response_detail_defaults_to_none() -> None:
    resp = SkillResponse(name="x", description="d", category=SkillCategory.CUSTOM)
    assert resp.detail is None


def test_skill_detail_request_defaults_to_empty_string() -> None:
    assert SkillDetailRequest().detail == ""
    assert SkillDetailRequest(detail="hello").detail == "hello"


def test_skill_to_response_surfaces_detail_from_ownership(isolated_paths) -> None:
    skill = SimpleNamespace(
        name="alpha-tool",
        description="desc",
        license=None,
        category=SkillCategory.CUSTOM,
        enabled=True,
    )
    ownership = {
        "name": "alpha-tool",
        "owner_user_id": "user-a",
        "published": False,
        "builtin": False,
        "detail": "这是技能的详细介绍",
    }
    resp = _skill_to_response(skill, ownership)
    assert resp.detail == "这是技能的详细介绍"
    assert resp.owner_user_id == "user-a"


def test_skill_to_response_detail_none_without_ownership(isolated_paths) -> None:
    skill = SimpleNamespace(
        name="builtin-skill",
        description="desc",
        license=None,
        category=SkillCategory.PUBLIC,
        enabled=True,
    )
    resp = _skill_to_response(skill, None)
    assert resp.detail is None
    assert resp.builtin is True


# ---------------------------------------------------------------------------
# SKILL.md source visibility — admin / publisher only
# ---------------------------------------------------------------------------


def test_admin_can_view_any_skill_source() -> None:
    # Admin sees a built-in skill's source (no owner).
    assert _can_view_skill_source(SkillCategory.PUBLIC, None, "user-a", is_admin=True) is True
    # ...and another user's custom skill.
    assert _can_view_skill_source(SkillCategory.CUSTOM, "user-b", "user-a", is_admin=True) is True


def test_publisher_can_view_own_custom_skill_source() -> None:
    assert _can_view_skill_source(SkillCategory.CUSTOM, "user-a", "user-a", is_admin=False) is True


def test_non_owner_cannot_view_custom_skill_source() -> None:
    # A different user — even if the skill is published to them — cannot read source.
    assert _can_view_skill_source(SkillCategory.CUSTOM, "user-b", "user-a", is_admin=False) is False


def test_non_admin_cannot_view_builtin_skill_source() -> None:
    # Built-in / public skills have no publisher, so non-admins never see source.
    assert _can_view_skill_source(SkillCategory.PUBLIC, None, "user-a", is_admin=False) is False


# ---------------------------------------------------------------------------
# Unified extensions-config writer regressions
# ---------------------------------------------------------------------------


@pytest.fixture
def raw_extensions_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    config_path = tmp_path / "extensions_config.json"
    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
    reset_extensions_config()
    try:
        yield config_path
    finally:
        reset_extensions_config()


@pytest.mark.asyncio
async def test_mcp_router_patches_raw_document_without_overwriting_skills(
    raw_extensions_config: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    original = {
        "mcpServers": {"old": {"command": "$OLD_COMMAND", "futureMcp": "old"}},
        "skills": {
            "offline-skill": {
                "enabled": False,
                "display": {"enabled": True, "mode": "list"},
                "futureSkill": {"keep": True},
            }
        },
        "futureTopLevel": {"placeholder": "$DO_NOT_RESOLVE"},
    }
    raw_extensions_config.write_text(json.dumps(original), encoding="utf-8")

    async def _allow_admin(_request) -> None:
        return None

    monkeypatch.setattr(mcp_router, "_require_admin", _allow_admin)
    before_update = await mcp_router.get_mcp_configuration(SimpleNamespace())
    assert before_update.mcp_servers["old"].command == "$OLD_COMMAND"
    assert before_update.mcp_servers["old"].model_extra == {"futureMcp": "old"}

    response = await mcp_router.update_mcp_configuration(
        mcp_router.McpConfigUpdateRequest(
            mcp_servers={
                "replacement": mcp_router.McpServerConfigResponse(
                    type="stdio",
                    command="python",
                    env={"TOKEN": "$MCP_TOKEN"},
                )
            }
        ),
        SimpleNamespace(),
    )

    persisted = json.loads(raw_extensions_config.read_text(encoding="utf-8"))
    assert set(response.mcp_servers) == {"replacement"}
    assert persisted["mcpServers"]["replacement"]["env"]["TOKEN"] == "$MCP_TOKEN"
    assert persisted["skills"] == original["skills"]
    assert persisted["futureTopLevel"] == original["futureTopLevel"]


def test_embedded_client_mcp_and_skill_updates_patch_only_owned_fields(
    raw_extensions_config: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    skills_root = tmp_path / "skills"
    skill_file = skills_root / "public" / "demo-dir" / "SKILL.md"
    skill_file.parent.mkdir(parents=True)
    skill_file.write_text(
        "---\nname: demo-skill\ndescription: Demo\n---\n",
        encoding="utf-8",
    )
    storage = LocalSkillStorage(host_path=str(skills_root))
    import deerflow.skills.storage as storage_module

    monkeypatch.setattr(storage_module, "get_or_new_skill_storage", lambda **_kwargs: storage)
    original = {
        "mcpServers": {"old": {"command": "$OLD_COMMAND"}},
        "skills": {
            "demo-skill": {
                "enabled": True,
                "display": {"enabled": True, "mode": "list"},
                "futureSkill": "keep",
            }
        },
        "futureTopLevel": ["keep"],
    }
    raw_extensions_config.write_text(json.dumps(original), encoding="utf-8")

    client = object.__new__(DeerFlowClient)
    client._agent = object()
    client._agent_config_key = ("cached",)
    assert client.get_mcp_config()["mcp_servers"]["old"]["command"] == "$OLD_COMMAND"

    before_invalid_update = raw_extensions_config.read_bytes()
    with pytest.raises(ValueError):
        client.update_mcp_config({"invalid": {"command": "python", "env": ["not-a-mapping"]}})
    assert raw_extensions_config.read_bytes() == before_invalid_update

    client.update_mcp_config(
        {
            "client-server": {
                "type": "stdio",
                "command": "python",
                "env": {"TOKEN": "$CLIENT_SECRET"},
                "futureMcp": {"keep": True},
            }
        }
    )
    persisted = json.loads(raw_extensions_config.read_text(encoding="utf-8"))
    assert persisted["mcpServers"]["client-server"]["env"]["TOKEN"] == "$CLIENT_SECRET"
    assert persisted["mcpServers"]["client-server"]["futureMcp"] == {"keep": True}
    assert persisted["skills"] == original["skills"]
    assert persisted["futureTopLevel"] == ["keep"]

    updated = client.update_skill("demo-skill", enabled=False)
    persisted = json.loads(raw_extensions_config.read_text(encoding="utf-8"))
    assert updated["enabled"] is False
    assert persisted["skills"]["demo-skill"]["enabled"] is False
    assert persisted["skills"]["demo-skill"]["display"] == original["skills"]["demo-skill"]["display"]
    assert persisted["skills"]["demo-skill"]["futureSkill"] == "keep"
    assert persisted["mcpServers"]["client-server"]["futureMcp"] == {"keep": True}
    assert persisted["futureTopLevel"] == ["keep"]
    assert client._agent is None
    assert client._agent_config_key is None


def test_extensions_config_writer_serializes_cross_process_patches(
    raw_extensions_config: Path,
) -> None:
    raw_extensions_config.write_text(
        json.dumps({"mcpServers": {}, "skills": {}, "concurrent": {}}),
        encoding="utf-8",
    )
    script = """
import sys
import time
from pathlib import Path
from deerflow.config.extensions_config import patch_extensions_config

Path(sys.argv[3]).write_text("ready", encoding="utf-8")
while not Path(sys.argv[4]).exists():
    time.sleep(0.01)

def update(document):
    current = document.setdefault("concurrent", {})
    time.sleep(0.25)
    current[sys.argv[2]] = True

patch_extensions_config(sys.argv[1], update)
"""
    backend_root = Path(__file__).resolve().parents[1]
    start_path = raw_extensions_config.with_suffix(".start")
    ready_paths = {
        key: raw_extensions_config.with_suffix(f".{key}.ready")
        for key in ("mcp", "skill")
    }
    processes = [
        subprocess.Popen(
            [
                sys.executable,
                "-c",
                script,
                str(raw_extensions_config),
                key,
                str(ready_paths[key]),
                str(start_path),
            ],
            cwd=backend_root,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        for key in ("mcp", "skill")
    ]
    deadline = time.monotonic() + 30
    while not all(path.exists() for path in ready_paths.values()):
        if time.monotonic() >= deadline:
            for process in processes:
                process.terminate()
            pytest.fail("Timed out waiting for config-writer subprocesses")
        time.sleep(0.01)
    start_path.touch()
    failures: list[str] = []
    for process in processes:
        _stdout, stderr = process.communicate(timeout=30)
        if process.returncode != 0:
            failures.append(stderr)
    assert failures == []

    persisted = json.loads(raw_extensions_config.read_text(encoding="utf-8"))
    assert persisted["concurrent"] == {"mcp": True, "skill": True}