87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""Query-slowdown diagnosis agent stays evidence-first and preserves local edits."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
|
|
def _load_checklist():
|
|
path = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "app"
|
|
/ "gateway"
|
|
/ "routers"
|
|
/ "_query_slowdown_diagnosis_seed_assets"
|
|
/ "skills"
|
|
/ "query-slowdown-diagnosis"
|
|
/ "scripts"
|
|
/ "print_checklist.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location("query_slowdown_checklist", path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_soul_checks_layers_before_blaming_mysql() -> None:
|
|
from app.gateway.routers import _query_slowdown_diagnosis_seed as seed
|
|
|
|
soul = (
|
|
seed._ASSETS_DIR / seed.QUERY_SLOWDOWN_DIAGNOSIS_AGENT_ID / "SOUL.md"
|
|
).read_text(encoding="utf-8")
|
|
browser_rule = soul.index("先分清慢在哪一层")
|
|
mysql_rule = soul.index("MySQL 当前有没有卡住的会话")
|
|
assert browser_rule < mysql_rule
|
|
assert "没有对应证据时不要下结论" in soul
|
|
assert "只有一个用户也可能" in soul or "一人" in soul
|
|
assert "空间大不能说明读写快" in soul
|
|
assert "重启就快" in soul
|
|
|
|
|
|
def test_seed_is_idempotent_and_preserves_local_edits(tmp_path, monkeypatch) -> None:
|
|
from app.gateway.routers import _query_slowdown_diagnosis_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, "_ensure_skill", lambda: None)
|
|
|
|
assert seed.ensure_query_slowdown_diagnosis_agent() is True
|
|
soul = tmp_path / "agents" / seed.QUERY_SLOWDOWN_DIAGNOSIS_AGENT_ID / "SOUL.md"
|
|
soul.write_text("现场调整", encoding="utf-8")
|
|
|
|
assert seed.ensure_query_slowdown_diagnosis_agent() is False
|
|
assert soul.read_text(encoding="utf-8") == "现场调整"
|
|
|
|
|
|
def test_checklist_prints_mysql_and_browser_steps(tmp_path, capsys) -> None:
|
|
checklist = _load_checklist()
|
|
config = tmp_path / "config.yaml"
|
|
config.write_text(
|
|
"\n".join(
|
|
[
|
|
"database:",
|
|
" backend: mysql",
|
|
" pool_size: 20",
|
|
" max_overflow: 20",
|
|
" pool_timeout: 30",
|
|
" pool_recycle: 1800",
|
|
" url: mysql+aiomysql://user:secret@host/db",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
checklist._config_candidates = lambda: [config]
|
|
checklist.main()
|
|
printed = capsys.readouterr().out
|
|
assert "database.backend=mysql" in printed
|
|
assert "secret" not in printed
|
|
assert "SHOW FULL PROCESSLIST" in printed
|
|
assert "Waiting/TTFB" in printed
|
|
assert "iostat" in printed
|