410 lines
20 KiB
Python
410 lines
20 KiB
Python
"""Regression tests for KIN-DOCS-003 — knowledge_synthesizer role.
|
||
|
||
Acceptance criteria:
|
||
1. Структурный тест промпта knowledge_synthesizer: 5 обязательных секций + специфические поля (#940)
|
||
2. Тест регистрации роли в specialists.yaml (model, tools, permissions, output_schema)
|
||
3. Тест корректности pipeline research_head с aggregation шагом:
|
||
- build_phase_order с ≥2 исследователями вставляет knowledge_synthesizer перед architect
|
||
- create_project_with_phases НЕ создаёт knowledge_synthesizer как фазу (авто-управляемая роль)
|
||
4. Тест context_builder: knowledge_synthesizer получает decisions, но не modules
|
||
5. Тест activate_phase: при ≥2 одобренных фазах инжектирует phases_context в brief
|
||
6. Проверка exclusion-листов (#921, #917): knowledge_synthesizer.md не исключён из проверок структуры
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
from core.db import init_db
|
||
from core import models
|
||
from core.phases import build_phase_order, create_project_with_phases, activate_phase
|
||
from core.context_builder import build_context
|
||
|
||
SPECIALISTS_YAML = Path(__file__).parent.parent / "agents" / "specialists.yaml"
|
||
PROMPTS_DIR = Path(__file__).parent.parent / "agents" / "prompts"
|
||
|
||
REQUIRED_SECTIONS = [
|
||
"## Working Mode",
|
||
"## Focus On",
|
||
"## Quality Checks",
|
||
"## Return Format",
|
||
"## Constraints",
|
||
]
|
||
|
||
OUTPUT_SCHEMA_FIELDS = [
|
||
"unified_findings",
|
||
"confidence_rated_conclusions",
|
||
"unresolved_conflicts",
|
||
"prioritized_actions",
|
||
"phases_context_used",
|
||
]
|
||
|
||
|
||
def _load_yaml():
|
||
return yaml.safe_load(SPECIALISTS_YAML.read_text(encoding="utf-8"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture
|
||
def conn():
|
||
c = init_db(":memory:")
|
||
yield c
|
||
c.close()
|
||
|
||
|
||
@pytest.fixture
|
||
def conn_with_project(conn):
|
||
models.create_project(conn, "p1", "TestProject", "/p1", tech_stack=["python"])
|
||
models.create_task(conn, "P1-001", "p1", "Research task")
|
||
yield conn
|
||
|
||
|
||
# ===========================================================================
|
||
# 1. Структурный тест промпта (#940)
|
||
# ===========================================================================
|
||
|
||
class TestKnowledgeSynthesizerPrompt:
|
||
"""Структурный тест agents/prompts/knowledge_synthesizer.md."""
|
||
|
||
def test_prompt_file_exists(self):
|
||
"""Файл agents/prompts/knowledge_synthesizer.md существует."""
|
||
path = PROMPTS_DIR / "knowledge_synthesizer.md"
|
||
assert path.exists(), "knowledge_synthesizer.md не найден в agents/prompts/"
|
||
|
||
@pytest.mark.parametrize("section", REQUIRED_SECTIONS)
|
||
def test_prompt_has_required_section(self, section):
|
||
"""Промпт содержит все 5 обязательных секций (REQUIRED_SECTIONS)."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert section in content, (
|
||
f"knowledge_synthesizer.md не содержит обязательную секцию {section!r}"
|
||
)
|
||
|
||
def test_prompt_has_input_section(self):
|
||
"""Промпт содержит секцию ## Input — агент-специфичная секция."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert "## Input" in content, (
|
||
"knowledge_synthesizer.md не содержит секцию '## Input'"
|
||
)
|
||
|
||
def test_prompt_contains_confidence_rated_conclusions(self):
|
||
"""Промпт содержит поле confidence_rated_conclusions в Return Format."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert "confidence_rated_conclusions" in content, (
|
||
"knowledge_synthesizer.md не содержит 'confidence_rated_conclusions'"
|
||
)
|
||
|
||
def test_prompt_contains_phases_context_used(self):
|
||
"""Промпт содержит поле phases_context_used в Return Format."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert "phases_context_used" in content, (
|
||
"knowledge_synthesizer.md не содержит 'phases_context_used'"
|
||
)
|
||
|
||
def test_prompt_contains_unresolved_conflicts(self):
|
||
"""Промпт содержит поле unresolved_conflicts в Return Format."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert "unresolved_conflicts" in content, (
|
||
"knowledge_synthesizer.md не содержит 'unresolved_conflicts'"
|
||
)
|
||
|
||
def test_prompt_contains_blocked_protocol(self):
|
||
"""Промпт содержит Blocked Protocol с инструкцией blocked_reason."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert "blocked_reason" in content, (
|
||
"knowledge_synthesizer.md не содержит 'blocked_reason' — Blocked Protocol обязателен"
|
||
)
|
||
|
||
def test_prompt_no_legacy_output_format_header(self):
|
||
"""Промпт НЕ содержит устаревшей секции '## Output format'."""
|
||
content = (PROMPTS_DIR / "knowledge_synthesizer.md").read_text(encoding="utf-8")
|
||
assert "## Output format" not in content, (
|
||
"knowledge_synthesizer.md содержит устаревшую секцию '## Output format'"
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# 2. Регистрация в specialists.yaml
|
||
# ===========================================================================
|
||
|
||
class TestKnowledgeSynthesizerSpecialists:
|
||
"""Тесты регистрации knowledge_synthesizer в agents/specialists.yaml."""
|
||
|
||
def test_role_exists_in_specialists(self):
|
||
"""specialists.yaml содержит роль knowledge_synthesizer."""
|
||
data = _load_yaml()
|
||
assert "knowledge_synthesizer" in data.get("specialists", {}), (
|
||
"knowledge_synthesizer отсутствует в specialists.yaml"
|
||
)
|
||
|
||
def test_role_model_is_sonnet(self):
|
||
"""knowledge_synthesizer использует модель sonnet."""
|
||
data = _load_yaml()
|
||
role = data["specialists"]["knowledge_synthesizer"]
|
||
assert role.get("model") == "sonnet", (
|
||
f"Ожидался model=sonnet, получили: {role.get('model')}"
|
||
)
|
||
|
||
def test_role_tools_include_read_grep_glob(self):
|
||
"""knowledge_synthesizer имеет инструменты Read, Grep, Glob."""
|
||
data = _load_yaml()
|
||
tools = data["specialists"]["knowledge_synthesizer"].get("tools", [])
|
||
for required_tool in ("Read", "Grep", "Glob"):
|
||
assert required_tool in tools, (
|
||
f"knowledge_synthesizer должен иметь инструмент {required_tool!r}"
|
||
)
|
||
|
||
def test_role_has_no_write_tools(self):
|
||
"""knowledge_synthesizer НЕ имеет write-инструментов (Read-only роль)."""
|
||
data = _load_yaml()
|
||
tools = set(data["specialists"]["knowledge_synthesizer"].get("tools", []))
|
||
write_tools = {"Write", "Edit", "Bash"}
|
||
unexpected = write_tools & tools
|
||
assert not unexpected, (
|
||
f"knowledge_synthesizer не должен иметь write-инструменты: {unexpected}"
|
||
)
|
||
|
||
def test_role_permissions_is_read_only(self):
|
||
"""knowledge_synthesizer имеет permissions=read_only."""
|
||
data = _load_yaml()
|
||
role = data["specialists"]["knowledge_synthesizer"]
|
||
assert role.get("permissions") == "read_only", (
|
||
f"Ожидался permissions=read_only, получили: {role.get('permissions')}"
|
||
)
|
||
|
||
def test_role_context_rules_decisions_all(self):
|
||
"""knowledge_synthesizer получает все decisions (context_rules.decisions=all)."""
|
||
data = _load_yaml()
|
||
role = data["specialists"]["knowledge_synthesizer"]
|
||
decisions = role.get("context_rules", {}).get("decisions")
|
||
assert decisions == "all", (
|
||
f"Ожидался context_rules.decisions=all, получили: {decisions}"
|
||
)
|
||
|
||
def test_role_has_output_schema(self):
|
||
"""knowledge_synthesizer имеет поле output_schema."""
|
||
data = _load_yaml()
|
||
role = data["specialists"]["knowledge_synthesizer"]
|
||
assert "output_schema" in role, (
|
||
"knowledge_synthesizer должен иметь output_schema"
|
||
)
|
||
|
||
@pytest.mark.parametrize("field", OUTPUT_SCHEMA_FIELDS)
|
||
def test_output_schema_has_required_field(self, field):
|
||
"""output_schema содержит каждое из 5 обязательных полей."""
|
||
data = _load_yaml()
|
||
schema = data["specialists"]["knowledge_synthesizer"]["output_schema"]
|
||
assert field in schema, (
|
||
f"output_schema knowledge_synthesizer не содержит обязательного поля {field!r}"
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# 3. Корректность pipeline с aggregation шагом
|
||
# ===========================================================================
|
||
|
||
class TestResearchPipelineWithSynthesizer:
|
||
"""Тесты корректности research pipeline с knowledge_synthesizer aggregation шагом."""
|
||
|
||
def test_build_phase_order_two_researchers_includes_synthesizer(self):
|
||
"""build_phase_order с 2 исследователями вставляет knowledge_synthesizer перед architect."""
|
||
result = build_phase_order(["business_analyst", "market_researcher"])
|
||
assert "knowledge_synthesizer" in result, (
|
||
"knowledge_synthesizer должен быть в pipeline при ≥2 исследователях"
|
||
)
|
||
|
||
def test_build_phase_order_synthesizer_before_architect(self):
|
||
"""knowledge_synthesizer всегда стоит непосредственно перед architect."""
|
||
result = build_phase_order(["tech_researcher", "ux_designer"])
|
||
idx_syn = result.index("knowledge_synthesizer")
|
||
idx_arch = result.index("architect")
|
||
assert idx_syn == idx_arch - 1, (
|
||
f"knowledge_synthesizer (pos={idx_syn}) должен идти прямо перед architect (pos={idx_arch})"
|
||
)
|
||
|
||
def test_build_phase_order_one_researcher_no_synthesizer(self):
|
||
"""build_phase_order с 1 исследователем НЕ вставляет knowledge_synthesizer."""
|
||
result = build_phase_order(["business_analyst"])
|
||
assert "knowledge_synthesizer" not in result, (
|
||
"knowledge_synthesizer не должен вставляться при 1 исследователе"
|
||
)
|
||
|
||
def test_create_project_two_researchers_excludes_synthesizer_from_phases(self, conn):
|
||
"""create_project_with_phases НЕ создаёт фазу knowledge_synthesizer — авто-управляемая роль."""
|
||
result = create_project_with_phases(
|
||
conn, "p1", "P1", "/p1",
|
||
description="Test", selected_roles=["market_researcher", "tech_researcher"],
|
||
)
|
||
phase_roles = [ph["role"] for ph in result["phases"]]
|
||
assert "knowledge_synthesizer" not in phase_roles, (
|
||
f"knowledge_synthesizer не должен быть в pipeline-фазах. Фазы: {phase_roles}"
|
||
)
|
||
|
||
def test_create_project_two_researchers_phase_count(self, conn):
|
||
"""create_project_with_phases с 2 исследователями создаёт 3 фазы (без synthesizer)."""
|
||
result = create_project_with_phases(
|
||
conn, "p1", "P1", "/p1",
|
||
description="Test", selected_roles=["market_researcher", "tech_researcher"],
|
||
)
|
||
assert len(result["phases"]) == 3, (
|
||
f"Ожидалось 3 фазы (market_researcher, tech_researcher, architect), "
|
||
f"получили {len(result['phases'])}: {[ph['role'] for ph in result['phases']]}"
|
||
)
|
||
|
||
def test_create_project_two_researchers_architect_last(self, conn):
|
||
"""create_project_with_phases с 2 исследователями ставит architect последним."""
|
||
result = create_project_with_phases(
|
||
conn, "p1", "P1", "/p1",
|
||
description="Test", selected_roles=["market_researcher", "tech_researcher"],
|
||
)
|
||
assert result["phases"][-1]["role"] == "architect"
|
||
|
||
|
||
# ===========================================================================
|
||
# 4. context_builder — knowledge_synthesizer
|
||
# ===========================================================================
|
||
|
||
class TestKnowledgeSynthesizerContextBuilder:
|
||
"""Тесты context_builder для роли knowledge_synthesizer."""
|
||
|
||
def test_knowledge_synthesizer_gets_decisions(self, conn_with_project):
|
||
"""build_context для knowledge_synthesizer включает decisions."""
|
||
models.add_decision(
|
||
conn_with_project, "p1", "gotcha", "Some decision", "Details"
|
||
)
|
||
ctx = build_context(conn_with_project, "P1-001", "knowledge_synthesizer", "p1")
|
||
assert "decisions" in ctx, (
|
||
"knowledge_synthesizer должен получать decisions в контексте"
|
||
)
|
||
|
||
def test_knowledge_synthesizer_does_not_get_modules(self, conn_with_project):
|
||
"""build_context для knowledge_synthesizer НЕ включает modules (в отличие от architect)."""
|
||
models.add_module(conn_with_project, "p1", "api", "backend", "src/api/")
|
||
ctx = build_context(conn_with_project, "P1-001", "knowledge_synthesizer", "p1")
|
||
assert "modules" not in ctx, (
|
||
"knowledge_synthesizer не должен получать modules — только decisions"
|
||
)
|
||
|
||
def test_architect_gets_both_decisions_and_modules(self, conn_with_project):
|
||
"""Регрессия: architect по-прежнему получает decisions И modules."""
|
||
models.add_module(conn_with_project, "p1", "api", "backend", "src/api/")
|
||
models.add_decision(
|
||
conn_with_project, "p1", "convention", "Use WAL", "SQLite WAL"
|
||
)
|
||
ctx = build_context(conn_with_project, "P1-001", "architect", "p1")
|
||
assert "decisions" in ctx
|
||
assert "modules" in ctx
|
||
|
||
|
||
# ===========================================================================
|
||
# 5. activate_phase — phases_context injection
|
||
# ===========================================================================
|
||
|
||
class TestActivatePhaseKnowledgeSynthesizer:
|
||
"""Тесты activate_phase для роли knowledge_synthesizer."""
|
||
|
||
def test_activate_phase_synthesizer_no_approved_phases_no_context(self, conn):
|
||
"""activate_phase для knowledge_synthesizer без одобренных фаз не добавляет phases_context."""
|
||
models.create_project(conn, "p1", "P1", "/p1", description="Test")
|
||
phase = models.create_phase(conn, "p1", "knowledge_synthesizer", 0)
|
||
result = activate_phase(conn, phase["id"])
|
||
task = models.get_task(conn, result["task_id"])
|
||
brief = task["brief"]
|
||
assert "phases_context" not in brief, (
|
||
"phases_context не должен быть в brief, если нет одобренных фаз"
|
||
)
|
||
|
||
def test_activate_phase_synthesizer_with_approved_phases_injects_context(self, conn):
|
||
"""activate_phase для knowledge_synthesizer собирает phases_context из одобренных фаз."""
|
||
models.create_project(conn, "p1", "P1", "/p1", description="Test")
|
||
|
||
# Создать и одобрить исследовательскую фазу
|
||
researcher_phase = models.create_phase(conn, "p1", "business_analyst", 0)
|
||
researcher_phase = activate_phase(conn, researcher_phase["id"])
|
||
researcher_task_id = researcher_phase["task_id"]
|
||
|
||
# Записать лог агента с output_summary для задачи исследователя
|
||
conn.execute(
|
||
"""INSERT INTO agent_logs (project_id, task_id, agent_role, action, output_summary, success)
|
||
VALUES (?, ?, ?, ?, ?, ?)""",
|
||
("p1", researcher_task_id, "business_analyst", "run",
|
||
"Market shows strong demand for product.", 1),
|
||
)
|
||
conn.commit()
|
||
|
||
# Одобрить исследовательскую фазу
|
||
models.update_phase(conn, researcher_phase["id"], status="approved")
|
||
|
||
# Создать и активировать фазу knowledge_synthesizer
|
||
synth_phase = models.create_phase(conn, "p1", "knowledge_synthesizer", 1)
|
||
result = activate_phase(conn, synth_phase["id"])
|
||
task = models.get_task(conn, result["task_id"])
|
||
brief = task["brief"]
|
||
|
||
assert "phases_context" in brief, (
|
||
"phases_context должен быть в brief при наличии одобренных фаз с agent_logs"
|
||
)
|
||
assert "business_analyst" in brief["phases_context"], (
|
||
"phases_context должен содержать ключ 'business_analyst'"
|
||
)
|
||
assert "Market shows strong demand" in brief["phases_context"]["business_analyst"], (
|
||
"phases_context['business_analyst'] должен содержать output_summary из agent_logs"
|
||
)
|
||
|
||
def test_activate_phase_synthesizer_ignores_pending_phases(self, conn):
|
||
"""activate_phase для knowledge_synthesizer не включает в phases_context неодобренные фазы."""
|
||
models.create_project(conn, "p1", "P1", "/p1", description="Test")
|
||
|
||
# Создать исследовательскую фазу в статусе active (не approved)
|
||
researcher_phase = models.create_phase(conn, "p1", "tech_researcher", 0)
|
||
researcher_phase = activate_phase(conn, researcher_phase["id"])
|
||
researcher_task_id = researcher_phase["task_id"]
|
||
|
||
# Записать лог агента
|
||
conn.execute(
|
||
"""INSERT INTO agent_logs (project_id, task_id, agent_role, action, output_summary, success)
|
||
VALUES (?, ?, ?, ?, ?, ?)""",
|
||
("p1", researcher_task_id, "tech_researcher", "run", "Tech analysis done.", 1),
|
||
)
|
||
conn.commit()
|
||
# Фаза остаётся в статусе active (не approved)
|
||
|
||
# Активировать knowledge_synthesizer
|
||
synth_phase = models.create_phase(conn, "p1", "knowledge_synthesizer", 1)
|
||
result = activate_phase(conn, synth_phase["id"])
|
||
task = models.get_task(conn, result["task_id"])
|
||
brief = task["brief"]
|
||
|
||
# phases_context должен отсутствовать или быть пустым
|
||
phases_context = brief.get("phases_context", {})
|
||
assert "tech_researcher" not in phases_context, (
|
||
"phases_context не должен включать фазы без статуса approved"
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# 6. Exclusion list guard (#921, #917)
|
||
# ===========================================================================
|
||
|
||
class TestExclusionListRegression:
|
||
"""Верифицирует, что knowledge_synthesizer.md не добавлен в exclusion list (#921, #917)."""
|
||
|
||
def test_knowledge_synthesizer_not_in_exclusion_list(self):
|
||
"""knowledge_synthesizer.md не включён в EXCLUDED_FROM_STRUCTURE_CHECK."""
|
||
from tests.test_kin_docs_002_regression import EXCLUDED_FROM_STRUCTURE_CHECK
|
||
assert "knowledge_synthesizer.md" not in EXCLUDED_FROM_STRUCTURE_CHECK, (
|
||
"knowledge_synthesizer.md не должен быть в EXCLUDED_FROM_STRUCTURE_CHECK — "
|
||
"роль должна проходить все стандартные структурные проверки"
|
||
)
|
||
|
||
def test_exclusion_list_is_empty(self):
|
||
"""EXCLUDED_FROM_STRUCTURE_CHECK остаётся пустым — guard от #929."""
|
||
from tests.test_kin_docs_002_regression import EXCLUDED_FROM_STRUCTURE_CHECK
|
||
assert EXCLUDED_FROM_STRUCTURE_CHECK == [], (
|
||
f"EXCLUDED_FROM_STRUCTURE_CHECK должен быть пустым, "
|
||
f"но содержит: {EXCLUDED_FROM_STRUCTURE_CHECK}"
|
||
)
|