"""Built-in Elasticsearch offline operations assistant seed contracts."""

from __future__ import annotations

import importlib.util
from pathlib import Path

import pytest
import yaml


def _load_analyzer():
    from app.gateway.routers import _elasticsearch_offline_ops_seed as seed

    path = seed._ASSETS_DIR / "skills" / seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME / "scripts" / "analyze_diagnostics.py"
    spec = importlib.util.spec_from_file_location("elasticsearch_diagnostics", path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_seed_assets_define_an_offline_evidence_first_agent() -> None:
    from app.gateway.routers import _elasticsearch_offline_ops_seed as seed

    agent_root = seed._ASSETS_DIR / seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID
    config = yaml.safe_load((agent_root / "config.yaml").read_text(encoding="utf-8"))
    soul = (agent_root / "SOUL.md").read_text(encoding="utf-8")
    skill_root = seed._ASSETS_DIR / "skills" / seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME
    skill = (skill_root / "SKILL.md").read_text(encoding="utf-8")

    assert config["id"] == seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID
    assert config["skills"] == [seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME]
    assert set(config["tool_groups"]) == {"file:read", "file:write", "bash"}
    assert "web" not in config["tool_groups"]
    assert len(config["common_questions"]) >= 4

    assert "目标 ES 主机" in soul
    assert "沙箱" in soul
    assert "版本" in soul
    assert "只读" in soul
    assert "密码" in soul
    assert "快照" in soul
    assert "禁止" in soul and "猜" in soul

    assert skill.startswith("---")
    assert "name: elasticsearch-offline-ops" in skill
    assert "references/offline-deployment.md" in skill
    assert "references/troubleshooting.md" in skill
    assert "collect_diagnostics.sh" in skill
    assert "analyze_diagnostics.py" in skill
    assert (skill_root / "references" / "offline-deployment.md").is_file()
    assert (skill_root / "references" / "troubleshooting.md").is_file()
    assert (skill_root / "scripts" / "collect_diagnostics.sh").is_file()


def test_seed_is_idempotent_and_preserves_local_edits(tmp_path, monkeypatch) -> None:
    from app.gateway.routers import _elasticsearch_offline_ops_seed as seed

    class FakePaths:
        def agent_dir(self, agent_id: str) -> Path:
            return tmp_path / "agents" / agent_id

    monkeypatch.setattr(seed, "get_paths", lambda: FakePaths())
    monkeypatch.setattr(seed, "_skills_public_dir", lambda: tmp_path / "skills" / "public")

    assert seed.ensure_elasticsearch_offline_ops_agent() is True
    agent_soul = tmp_path / "agents" / seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID / "SOUL.md"
    skill_file = tmp_path / "skills" / "public" / seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME / "SKILL.md"
    assert agent_soul.is_file()
    assert skill_file.is_file()

    agent_soul.write_text("现场智能体修改", encoding="utf-8")
    skill_file.write_text("现场技能修改", encoding="utf-8")
    assert seed.ensure_elasticsearch_offline_ops_agent() is False
    assert agent_soul.read_text(encoding="utf-8") == "现场智能体修改"
    assert skill_file.read_text(encoding="utf-8") == "现场技能修改"


@pytest.mark.asyncio
async def test_seeded_agent_syncs_into_the_builtin_catalog(tmp_path, monkeypatch) -> None:
    from app.gateway.routers import _elasticsearch_offline_ops_seed as seed
    from app.gateway.routers.agents import _sync_legacy_agents
    from deerflow.config import agents_config
    from deerflow.persistence.agents import MemoryAgentStore

    class FakePaths:
        @property
        def agents_dir(self) -> Path:
            return tmp_path / "agents"

        def agent_dir(self, agent_id: str) -> Path:
            return self.agents_dir / agent_id

    paths = FakePaths()
    monkeypatch.setattr(seed, "get_paths", lambda: paths)
    monkeypatch.setattr(agents_config, "get_paths", lambda: paths)
    monkeypatch.setattr(seed, "_skills_public_dir", lambda: tmp_path / "skills" / "public")

    assert seed.ensure_elasticsearch_offline_ops_agent() is True
    store = MemoryAgentStore()
    await _sync_legacy_agents(store)

    row = await store.get_any(seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID)
    assert row is not None
    assert row["user_id"] is None
    assert row["published"] is True
    assert row["name"] == "Elasticsearch 离线运维助手"
    extras = await store.get_extras_for([seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID])
    assert extras[seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID]["skills"] == [seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME]
    assert set(extras[seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID]["tool_groups"]) == {
        "file:read",
        "file:write",
        "bash",
    }


def test_diagnostics_analyzer_returns_evidence_and_next_checks() -> None:
    analyzer = _load_analyzer()
    report = analyzer.analyze_text(
        """
        fatal exception while booting Elasticsearch
        bootstrap checks failed
        max virtual memory areas vm.max_map_count [65530] is too low
        received plaintext http traffic on an https channel
        cluster health status: red
        unassigned_shards: 12
        """
    )

    ids = {finding["id"] for finding in report["findings"]}
    assert {"vm_max_map_count", "http_tls_mismatch", "cluster_red"} <= ids
    assert report["evidence_count"] >= 3
    assert all(finding["evidence"] for finding in report["findings"])
    assert all(finding["next_check"] for finding in report["findings"])


def test_diagnostics_analyzer_does_not_invent_a_root_cause() -> None:
    analyzer = _load_analyzer()
    report = analyzer.analyze_text("节点偶尔有一点慢，但没有附日志或指标")

    assert report["findings"] == []
    assert report["status"] == "insufficient_evidence"
    assert report["recommended_inputs"]


def test_diagnostics_analyzer_redacts_secrets_from_evidence() -> None:
    analyzer = _load_analyzer()
    report = analyzer.analyze_text("SSLHandshakeException Authorization: ApiKey abc123 password=do-not-print")

    evidence = "\n".join(line for finding in report["findings"] for line in finding["evidence"])
    assert "abc123" not in evidence
    assert "do-not-print" not in evidence
    assert evidence.count("<redacted>") == 2


def test_collector_is_read_only_and_does_not_disable_tls() -> None:
    from app.gateway.routers import _elasticsearch_offline_ops_seed as seed

    collector = (seed._ASSETS_DIR / "skills" / seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME / "scripts" / "collect_diagnostics.sh").read_text(encoding="utf-8")

    forbidden = ("systemctl restart", "systemctl stop", "rm -rf", "swapoff", "sysctl -w")
    assert not any(command in collector for command in forbidden)
    assert "curl -k" not in collector
    assert "--insecure" not in collector
    assert "ES_NETRC_FILE" in collector
    assert "ES_HEADER_FILE" in collector
    assert "chmod 600" in collector
    assert "--config" not in collector
    assert "/_cluster/health" in collector
    assert "/_cat/nodes" in collector
    assert "journalctl" in collector