247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""Release-time regression checks for startup agent and skill seed assets."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from app.gateway.routers import _agentfx_seed as agentfx_seed
|
|
from app.gateway.routers import _latency_diagnosis_seed as latency_diagnosis_seed
|
|
from app.gateway.routers import _ai_writing_seed as ai_writing_seed
|
|
from app.gateway.routers import _cognitive_situation_seed as cognitive_situation_seed
|
|
from app.gateway.routers import _elasticsearch_offline_ops_seed as elasticsearch_ops_seed
|
|
from app.gateway.routers import _gongwen_seed as gongwen_seed
|
|
from app.gateway.routers import _page_report_designer_seed as page_report_seed
|
|
from app.gateway.routers import _query_slowdown_diagnosis_seed as query_slowdown_diagnosis_seed
|
|
from deerflow.config.paths import Paths
|
|
from deerflow.skills import storage as skill_storage_module
|
|
|
|
|
|
class _SkillStorage:
|
|
def __init__(self, root: Path) -> None:
|
|
self._root = root
|
|
|
|
def get_skills_root_path(self) -> Path:
|
|
return self._root
|
|
|
|
|
|
def _assert_nonempty_file(path: Path) -> None:
|
|
assert path.is_file(), f"missing seed asset: {path}"
|
|
assert path.stat().st_size > 0, f"empty seed asset: {path}"
|
|
|
|
|
|
def _assert_agent_assets(asset_root: Path, agent_ids: tuple[str, ...]) -> None:
|
|
for agent_id in agent_ids:
|
|
config_path = asset_root / agent_id / "config.yaml"
|
|
soul_path = asset_root / agent_id / "SOUL.md"
|
|
_assert_nonempty_file(config_path)
|
|
_assert_nonempty_file(soul_path)
|
|
|
|
config = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
|
|
assert config.get("id") == agent_id, config_path
|
|
assert str(config.get("name") or "").strip(), config_path
|
|
assert soul_path.read_text(encoding="utf-8").strip(), soul_path
|
|
|
|
|
|
def _assert_skill_asset(path: Path, expected_name: str) -> None:
|
|
_assert_nonempty_file(path)
|
|
text = path.read_text(encoding="utf-8")
|
|
assert text.startswith("---"), path
|
|
parts = text.split("---", 2)
|
|
assert len(parts) == 3, path
|
|
frontmatter = yaml.safe_load(parts[1]) or {}
|
|
assert frontmatter.get("name") == expected_name, path
|
|
assert str(frontmatter.get("description") or "").strip(), path
|
|
assert parts[2].strip(), path
|
|
|
|
|
|
def test_required_startup_seed_assets_are_complete() -> None:
|
|
"""Every file required by the startup skill seeders ships in source."""
|
|
_assert_agent_assets(
|
|
ai_writing_seed._ASSETS_DIR,
|
|
tuple(ai_writing_seed.required_agent_ids()),
|
|
)
|
|
_assert_agent_assets(
|
|
page_report_seed._ASSETS_DIR,
|
|
(page_report_seed.PAGE_REPORT_DESIGNER_AGENT_ID,),
|
|
)
|
|
_assert_agent_assets(
|
|
gongwen_seed._ASSETS_DIR,
|
|
(gongwen_seed.GONGWEN_AGENT_ID,),
|
|
)
|
|
_assert_agent_assets(
|
|
agentfx_seed._ASSETS_DIR,
|
|
(agentfx_seed.AGENTFX_AGENT_ID,),
|
|
)
|
|
_assert_agent_assets(
|
|
latency_diagnosis_seed._ASSETS_DIR,
|
|
(latency_diagnosis_seed.LATENCY_DIAGNOSIS_AGENT_ID,),
|
|
)
|
|
_assert_agent_assets(
|
|
query_slowdown_diagnosis_seed._ASSETS_DIR,
|
|
(query_slowdown_diagnosis_seed.QUERY_SLOWDOWN_DIAGNOSIS_AGENT_ID,),
|
|
)
|
|
_assert_agent_assets(
|
|
elasticsearch_ops_seed._ASSETS_DIR,
|
|
(elasticsearch_ops_seed.ELASTICSEARCH_OFFLINE_OPS_AGENT_ID,),
|
|
)
|
|
|
|
skill_assets = (
|
|
(
|
|
ai_writing_seed._ASSETS_DIR / "skills" / ai_writing_seed.KNOWLEDGE_SEARCH_SKILL_NAME / "SKILL.md",
|
|
ai_writing_seed.KNOWLEDGE_SEARCH_SKILL_NAME,
|
|
),
|
|
*(
|
|
(
|
|
page_report_seed._ASSETS_DIR / "skills" / name / "SKILL.md",
|
|
name,
|
|
)
|
|
for name in page_report_seed.PAGE_REPORT_SKILL_NAMES
|
|
),
|
|
(
|
|
gongwen_seed._ASSETS_DIR / "skills" / gongwen_seed.GONGWEN_WORKFLOW_SKILL / "SKILL.md",
|
|
gongwen_seed.GONGWEN_WORKFLOW_SKILL,
|
|
),
|
|
(
|
|
latency_diagnosis_seed._ASSETS_DIR
|
|
/ "skills"
|
|
/ latency_diagnosis_seed.LATENCY_DIAGNOSIS_SKILL_NAME
|
|
/ "SKILL.md",
|
|
latency_diagnosis_seed.LATENCY_DIAGNOSIS_SKILL_NAME,
|
|
),
|
|
(
|
|
query_slowdown_diagnosis_seed._ASSETS_DIR / "skills" / query_slowdown_diagnosis_seed.QUERY_SLOWDOWN_DIAGNOSIS_SKILL_NAME / "SKILL.md",
|
|
query_slowdown_diagnosis_seed.QUERY_SLOWDOWN_DIAGNOSIS_SKILL_NAME,
|
|
),
|
|
(
|
|
elasticsearch_ops_seed._ASSETS_DIR / "skills" / elasticsearch_ops_seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME / "SKILL.md",
|
|
elasticsearch_ops_seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME,
|
|
),
|
|
*(
|
|
(
|
|
agentfx_seed._ASSETS_DIR / "skills" / name / "SKILL.md",
|
|
name,
|
|
)
|
|
for name in agentfx_seed.TASK_REPORT_SKILL_NAMES
|
|
),
|
|
)
|
|
for skill_path, name in skill_assets:
|
|
_assert_skill_asset(skill_path, name)
|
|
|
|
elasticsearch_skill_root = elasticsearch_ops_seed._ASSETS_DIR / "skills" / elasticsearch_ops_seed.ELASTICSEARCH_OFFLINE_OPS_SKILL_NAME
|
|
for relative_path in (
|
|
"references/offline-deployment.md",
|
|
"references/troubleshooting.md",
|
|
"scripts/collect_diagnostics.sh",
|
|
"scripts/analyze_diagnostics.py",
|
|
):
|
|
_assert_nonempty_file(elasticsearch_skill_root / relative_path)
|
|
|
|
cognitive_skill_root = cognitive_situation_seed._ASSETS_DIR / "skills" / cognitive_situation_seed.COGNITIVE_SITUATION_SKILL_NAMES[0]
|
|
runtime_template = cognitive_skill_root / "assets" / "runtime" / "index.html"
|
|
_assert_nonempty_file(runtime_template)
|
|
assert not (cognitive_skill_root / "assets" / "runtime" / "assets").exists()
|
|
assert not (cognitive_skill_root / "assets" / "runtime" / "app.js").exists()
|
|
assert not (cognitive_skill_root / "assets" / "runtime" / "styles.css").exists()
|
|
assert sum(path.stat().st_size for path in cognitive_skill_root.rglob("*") if path.is_file()) < 200_000
|
|
|
|
agentfx_skills = agentfx_seed._ASSETS_DIR / "skills"
|
|
_assert_nonempty_file(agentfx_skills / agentfx_seed.TASK_REPORT_LIB_NAME / "convert_lib.py")
|
|
manifest_path = agentfx_skills / agentfx_seed.TASK_REPORT_LIB_NAME / ".deerflow-skill-support.json"
|
|
assert not (agentfx_skills / agentfx_seed.TASK_REPORT_LIB_NAME / "SKILL.md").exists(), "task-report-lib is support code and must not enter skill discovery"
|
|
_assert_nonempty_file(manifest_path)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
converter_skills = set(agentfx_seed.TASK_REPORT_SKILL_NAMES) - {agentfx_seed.TASK_REPORT_IMPORT_SKILL_NAME}
|
|
assert set(manifest.get("required_by") or ()) == converter_skills
|
|
|
|
_assert_nonempty_file(agentfx_skills / agentfx_seed.TASK_REPORT_IMPORT_SKILL_NAME / "scripts" / "task_report_import.py")
|
|
for name in converter_skills:
|
|
_assert_nonempty_file(agentfx_skills / name / "scripts" / "convert.py")
|
|
|
|
|
|
@pytest.fixture()
|
|
def isolated_seed_targets(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> tuple[Paths, Path]:
|
|
paths = Paths(tmp_path / "state")
|
|
skills_root = tmp_path / "skills"
|
|
storage = _SkillStorage(skills_root)
|
|
monkeypatch.setattr(page_report_seed, "get_paths", lambda: paths)
|
|
monkeypatch.setattr(gongwen_seed, "get_paths", lambda: paths)
|
|
monkeypatch.setattr(
|
|
skill_storage_module,
|
|
"get_or_new_skill_storage",
|
|
lambda: storage,
|
|
)
|
|
return paths, skills_root
|
|
|
|
|
|
def test_page_report_seeder_is_idempotent_and_preserves_local_edits(
|
|
isolated_seed_targets: tuple[Paths, Path],
|
|
) -> None:
|
|
paths, skills_root = isolated_seed_targets
|
|
|
|
assert page_report_seed.ensure_page_report_designer_agent() is True
|
|
agent_soul = paths.agent_dir(page_report_seed.PAGE_REPORT_DESIGNER_AGENT_ID) / "SOUL.md"
|
|
customized_skill = skills_root / "public" / page_report_seed.PAGE_REPORT_SKILL_NAMES[0] / "SKILL.md"
|
|
agent_soul.write_text("现场调整的智能体规则", encoding="utf-8")
|
|
customized_skill.write_text("现场调整的技能规则", encoding="utf-8")
|
|
|
|
assert page_report_seed.ensure_page_report_designer_agent() is False
|
|
assert agent_soul.read_text(encoding="utf-8") == "现场调整的智能体规则"
|
|
assert customized_skill.read_text(encoding="utf-8") == "现场调整的技能规则"
|
|
for name in page_report_seed.PAGE_REPORT_SKILL_NAMES:
|
|
assert (skills_root / "public" / name / "SKILL.md").is_file()
|
|
|
|
|
|
def test_gongwen_seeder_is_idempotent_and_preserves_local_edits(
|
|
isolated_seed_targets: tuple[Paths, Path],
|
|
) -> None:
|
|
paths, skills_root = isolated_seed_targets
|
|
|
|
assert gongwen_seed.ensure_gongwen_agent() is True
|
|
agent_soul = paths.agent_dir(gongwen_seed.GONGWEN_AGENT_ID) / "SOUL.md"
|
|
workflow_skill = skills_root / "public" / gongwen_seed.GONGWEN_WORKFLOW_SKILL / "SKILL.md"
|
|
agent_soul.write_text("现场调整的智能体规则", encoding="utf-8")
|
|
workflow_skill.write_text("现场调整的技能规则", encoding="utf-8")
|
|
|
|
assert gongwen_seed.ensure_gongwen_agent() is False
|
|
assert agent_soul.read_text(encoding="utf-8") == "现场调整的智能体规则"
|
|
assert workflow_skill.read_text(encoding="utf-8") == "现场调整的技能规则"
|
|
|
|
|
|
def test_cognitive_situation_seeder_synchronizes_the_lightweight_skill_and_prunes_legacy_runtime(
|
|
isolated_seed_targets: tuple[Paths, Path],
|
|
) -> None:
|
|
_, skills_root = isolated_seed_targets
|
|
|
|
assert cognitive_situation_seed.ensure_cognitive_situation_skills() is True
|
|
for name in cognitive_situation_seed.COGNITIVE_SITUATION_SKILL_NAMES:
|
|
target_root = skills_root / "public" / name
|
|
seeded_skill = target_root / "SKILL.md"
|
|
assert seeded_skill.is_file()
|
|
assert (target_root / "assets" / "runtime" / "index.html").is_file()
|
|
local_note = target_root / "local-note.md"
|
|
local_note.write_text("keep this user file", encoding="utf-8")
|
|
legacy_runtime = target_root / "assets" / "runtime"
|
|
(legacy_runtime / "assets" / "maps").mkdir(parents=True)
|
|
(legacy_runtime / "assets" / "maps" / "old-map.png").write_bytes(b"legacy")
|
|
(legacy_runtime / "app.js").write_text("legacy", encoding="utf-8")
|
|
(legacy_runtime / "styles.css").write_text("legacy", encoding="utf-8")
|
|
(legacy_runtime / "data").mkdir()
|
|
|
|
assert cognitive_situation_seed.ensure_cognitive_situation_skills() is True
|
|
for name in cognitive_situation_seed.COGNITIVE_SITUATION_SKILL_NAMES:
|
|
target_root = skills_root / "public" / name
|
|
runtime = target_root / "assets" / "runtime"
|
|
assert (target_root / "local-note.md").read_text(encoding="utf-8") == "keep this user file"
|
|
assert (runtime / "index.html").is_file()
|
|
assert not (runtime / "assets").exists()
|
|
assert not (runtime / "app.js").exists()
|
|
assert not (runtime / "styles.css").exists()
|
|
assert not (runtime / "data").exists()
|