173 lines
7.0 KiB
Python
173 lines
7.0 KiB
Python
"""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
|