666 lines
32 KiB
Python
666 lines
32 KiB
Python
"""Tests for core/context_builder.py — context assembly per role."""
|
||
|
||
import pytest
|
||
from core.db import init_db
|
||
from core import models
|
||
from core.context_builder import build_context, format_prompt
|
||
|
||
|
||
@pytest.fixture
|
||
def conn():
|
||
c = init_db(":memory:")
|
||
# Seed project, modules, decisions, tasks
|
||
models.create_project(c, "vdol", "ВДОЛЬ и ПОПЕРЕК", "~/projects/vdolipoperek",
|
||
tech_stack=["vue3", "typescript", "nodejs"])
|
||
models.add_module(c, "vdol", "search", "frontend", "src/search/")
|
||
models.add_module(c, "vdol", "api", "backend", "src/api/")
|
||
models.add_decision(c, "vdol", "gotcha", "Safari bug",
|
||
"position:fixed breaks", category="ui", tags=["ios"])
|
||
models.add_decision(c, "vdol", "workaround", "API rate limit",
|
||
"10 req/s max", category="api")
|
||
models.add_decision(c, "vdol", "convention", "Use WAL mode",
|
||
"Always use WAL for SQLite", category="architecture")
|
||
models.add_decision(c, "vdol", "decision", "Auth required",
|
||
"All endpoints need auth", category="security")
|
||
models.create_task(c, "VDOL-001", "vdol", "Fix search filters",
|
||
brief={"module": "search", "route_type": "debug"})
|
||
models.create_task(c, "VDOL-002", "vdol", "Add payments",
|
||
status="in_progress")
|
||
yield c
|
||
c.close()
|
||
|
||
|
||
class TestBuildContext:
|
||
def test_pm_gets_everything(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "pm", "vdol")
|
||
assert ctx["task"]["id"] == "VDOL-001"
|
||
assert ctx["project"]["id"] == "vdol"
|
||
assert len(ctx["modules"]) == 2
|
||
assert len(ctx["decisions"]) == 4 # all decisions
|
||
assert len(ctx["active_tasks"]) == 1 # VDOL-002 in_progress
|
||
assert "pm" in ctx["available_specialists"]
|
||
|
||
def test_architect_gets_all_decisions_and_modules(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "architect", "vdol")
|
||
assert len(ctx["modules"]) == 2
|
||
assert len(ctx["decisions"]) == 4
|
||
|
||
def test_debugger_gets_only_gotcha_workaround(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
types = {d["type"] for d in ctx["decisions"]}
|
||
assert types <= {"gotcha", "workaround"}
|
||
assert "convention" not in types
|
||
assert "decision" not in types
|
||
assert ctx["module_hint"] == "search"
|
||
|
||
def test_frontend_dev_gets_gotcha_workaround_convention(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "frontend_dev", "vdol")
|
||
types = {d["type"] for d in ctx["decisions"]}
|
||
assert "gotcha" in types
|
||
assert "workaround" in types
|
||
assert "convention" in types
|
||
assert "decision" not in types # plain decisions excluded
|
||
|
||
def test_backend_dev_same_as_frontend(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
types = {d["type"] for d in ctx["decisions"]}
|
||
assert types == {"gotcha", "workaround", "convention"}
|
||
|
||
def test_reviewer_gets_only_conventions(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "reviewer", "vdol")
|
||
types = {d["type"] for d in ctx["decisions"]}
|
||
assert types == {"convention"}
|
||
|
||
def test_tester_gets_minimal_context(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "tester", "vdol")
|
||
assert ctx["task"] is not None
|
||
assert ctx["project"] is not None
|
||
assert "decisions" not in ctx
|
||
assert "modules" not in ctx
|
||
|
||
def test_security_gets_security_decisions(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "security", "vdol")
|
||
categories = {d.get("category") for d in ctx["decisions"]}
|
||
assert categories == {"security"}
|
||
|
||
def test_unknown_role_gets_fallback(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "unknown_role", "vdol")
|
||
assert "decisions" in ctx
|
||
assert len(ctx["decisions"]) > 0
|
||
|
||
|
||
class TestFormatPrompt:
|
||
def test_format_with_template(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
prompt = format_prompt(ctx, "debugger", "You are a debugger. Find bugs.")
|
||
assert "You are a debugger" in prompt
|
||
assert "VDOL-001" in prompt
|
||
assert "Fix search filters" in prompt
|
||
assert "vdol" in prompt
|
||
assert "vue3" in prompt
|
||
|
||
def test_format_includes_decisions(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
prompt = format_prompt(ctx, "debugger", "Debug this.")
|
||
assert "Safari bug" in prompt
|
||
assert "API rate limit" in prompt
|
||
# Convention should NOT be here (debugger doesn't get it)
|
||
assert "WAL mode" not in prompt
|
||
|
||
def test_format_pm_includes_specialists(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "pm", "vdol")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "Available specialists" in prompt
|
||
assert "debugger" in prompt
|
||
assert "Active tasks" in prompt
|
||
assert "VDOL-002" in prompt
|
||
|
||
def test_format_with_previous_output(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "tester", "vdol")
|
||
ctx["previous_output"] = "Found race condition in useSearch.ts"
|
||
prompt = format_prompt(ctx, "tester", "Write tests.")
|
||
assert "Previous step output" in prompt
|
||
assert "race condition" in prompt
|
||
|
||
def test_format_loads_prompt_file(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "pm", "vdol")
|
||
prompt = format_prompt(ctx, "pm") # Should load from agents/prompts/pm.md
|
||
assert "decompose" in prompt.lower() or "pipeline" in prompt.lower()
|
||
|
||
def test_format_missing_prompt_file(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "analyst", "vdol")
|
||
prompt = format_prompt(ctx, "analyst") # No analyst.md exists
|
||
assert "analyst" in prompt.lower()
|
||
|
||
def test_format_includes_language_ru(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
prompt = format_prompt(ctx, "debugger", "Debug.")
|
||
assert "## Language" in prompt
|
||
assert "Russian" in prompt
|
||
assert "ALWAYS respond in Russian" in prompt
|
||
|
||
def test_format_includes_language_en(self, conn):
|
||
# Update project language to en
|
||
conn.execute("UPDATE projects SET language='en' WHERE id='vdol'")
|
||
conn.commit()
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
prompt = format_prompt(ctx, "debugger", "Debug.")
|
||
assert "ALWAYS respond in English" in prompt
|
||
|
||
|
||
class TestLanguageInProject:
|
||
def test_project_has_language_default(self, conn):
|
||
p = models.get_project(conn, "vdol")
|
||
assert p["language"] == "ru"
|
||
|
||
def test_create_project_with_language(self, conn):
|
||
p = models.create_project(conn, "en-proj", "English Project", "/en",
|
||
language="en")
|
||
assert p["language"] == "en"
|
||
|
||
def test_context_carries_language(self, conn):
|
||
ctx = build_context(conn, "VDOL-001", "pm", "vdol")
|
||
assert ctx["project"]["language"] == "ru"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# KIN-045: Revise context — revise_comment + last agent output injection
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestReviseContext:
|
||
"""build_context и format_prompt корректно инжектируют контекст ревизии."""
|
||
|
||
def test_build_context_includes_revise_comment_in_task(self, conn):
|
||
"""Если у задачи есть revise_comment, он попадает в ctx['task']."""
|
||
conn.execute(
|
||
"UPDATE tasks SET revise_comment=? WHERE id='VDOL-001'",
|
||
("Доисследуй edge case с пустым массивом",),
|
||
)
|
||
conn.commit()
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
assert ctx["task"]["revise_comment"] == "Доисследуй edge case с пустым массивом"
|
||
|
||
def test_build_context_fetches_last_agent_output_when_revise_comment_set(self, conn):
|
||
"""При revise_comment build_context достаёт last_agent_output из agent_logs."""
|
||
from core import models
|
||
models.log_agent_run(
|
||
conn, "vdol", "developer", "execute",
|
||
task_id="VDOL-001",
|
||
output_summary="Реализован endpoint POST /api/items",
|
||
success=True,
|
||
)
|
||
conn.execute(
|
||
"UPDATE tasks SET revise_comment=? WHERE id='VDOL-001'",
|
||
("Добавь валидацию входных данных",),
|
||
)
|
||
conn.commit()
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
assert ctx.get("last_agent_output") == "Реализован endpoint POST /api/items"
|
||
|
||
def test_build_context_no_last_agent_output_when_no_successful_logs(self, conn):
|
||
"""revise_comment есть, но нет успешных логов — last_agent_output отсутствует."""
|
||
from core import models
|
||
models.log_agent_run(
|
||
conn, "vdol", "developer", "execute",
|
||
task_id="VDOL-001",
|
||
output_summary="Permission denied",
|
||
success=False,
|
||
)
|
||
conn.execute(
|
||
"UPDATE tasks SET revise_comment=? WHERE id='VDOL-001'",
|
||
("Повтори без ошибок",),
|
||
)
|
||
conn.commit()
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
assert "last_agent_output" not in ctx
|
||
|
||
def test_build_context_no_revise_fields_when_no_revise_comment(self, conn):
|
||
"""Обычная задача без revise_comment не получает last_agent_output в контексте."""
|
||
from core import models
|
||
models.log_agent_run(
|
||
conn, "vdol", "developer", "execute",
|
||
task_id="VDOL-001",
|
||
output_summary="Всё готово",
|
||
success=True,
|
||
)
|
||
# revise_comment не устанавливаем
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
assert "last_agent_output" not in ctx
|
||
assert ctx["task"].get("revise_comment") is None
|
||
|
||
def test_format_prompt_includes_director_revision_request(self, conn):
|
||
"""format_prompt содержит секцию '## Director's revision request:' при revise_comment."""
|
||
conn.execute(
|
||
"UPDATE tasks SET revise_comment=? WHERE id='VDOL-001'",
|
||
("Обработай случай пустого списка",),
|
||
)
|
||
conn.commit()
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
prompt = format_prompt(ctx, "backend_dev", "You are a developer.")
|
||
assert "## Director's revision request:" in prompt
|
||
assert "Обработай случай пустого списка" in prompt
|
||
|
||
def test_format_prompt_includes_previous_output_before_revision(self, conn):
|
||
"""format_prompt содержит '## Your previous output (before revision):' при last_agent_output."""
|
||
from core import models
|
||
models.log_agent_run(
|
||
conn, "vdol", "developer", "execute",
|
||
task_id="VDOL-001",
|
||
output_summary="Сделал миграцию БД",
|
||
success=True,
|
||
)
|
||
conn.execute(
|
||
"UPDATE tasks SET revise_comment=? WHERE id='VDOL-001'",
|
||
("Ещё добавь индекс",),
|
||
)
|
||
conn.commit()
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
prompt = format_prompt(ctx, "backend_dev", "You are a developer.")
|
||
assert "## Your previous output (before revision):" in prompt
|
||
assert "Сделал миграцию БД" in prompt
|
||
|
||
def test_format_prompt_no_revision_sections_when_no_revise_comment(self, conn):
|
||
"""Без revise_comment в prompt нет секций ревизии."""
|
||
ctx = build_context(conn, "VDOL-001", "backend_dev", "vdol")
|
||
prompt = format_prompt(ctx, "backend_dev", "You are a developer.")
|
||
assert "## Director's revision request:" not in prompt
|
||
assert "## Your previous output (before revision):" not in prompt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# KIN-071: project_type and SSH context
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestOperationsProject:
|
||
"""KIN-071: operations project_type propagates to context and prompt."""
|
||
|
||
@pytest.fixture
|
||
def ops_conn(self):
|
||
c = init_db(":memory:")
|
||
models.create_project(
|
||
c, "srv", "My Server", "",
|
||
project_type="operations",
|
||
ssh_host="10.0.0.1",
|
||
ssh_user="root",
|
||
ssh_key_path="~/.ssh/id_rsa",
|
||
ssh_proxy_jump="jumpt",
|
||
)
|
||
models.create_task(c, "SRV-001", "srv", "Scan server")
|
||
yield c
|
||
c.close()
|
||
|
||
def test_slim_project_includes_project_type(self, ops_conn):
|
||
"""KIN-071: _slim_project включает project_type."""
|
||
ctx = build_context(ops_conn, "SRV-001", "sysadmin", "srv")
|
||
assert ctx["project"]["project_type"] == "operations"
|
||
|
||
def test_slim_project_includes_ssh_fields_for_operations(self, ops_conn):
|
||
"""KIN-071: _slim_project включает ssh_* поля для operations-проектов."""
|
||
ctx = build_context(ops_conn, "SRV-001", "sysadmin", "srv")
|
||
proj = ctx["project"]
|
||
assert proj["ssh_host"] == "10.0.0.1"
|
||
assert proj["ssh_user"] == "root"
|
||
assert proj["ssh_key_path"] == "~/.ssh/id_rsa"
|
||
assert proj["ssh_proxy_jump"] == "jumpt"
|
||
|
||
def test_slim_project_no_ssh_fields_for_development(self):
|
||
"""KIN-071: development-проект не получает ssh_* в slim."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "dev", "Dev", "/path")
|
||
models.create_task(c, "DEV-001", "dev", "A task")
|
||
ctx = build_context(c, "DEV-001", "backend_dev", "dev")
|
||
assert "ssh_host" not in ctx["project"]
|
||
c.close()
|
||
|
||
def test_sysadmin_context_gets_decisions_and_modules(self, ops_conn):
|
||
"""KIN-071: sysadmin роль получает все decisions и modules."""
|
||
models.add_module(ops_conn, "srv", "nginx", "service", "/etc/nginx")
|
||
models.add_decision(ops_conn, "srv", "gotcha", "Port 80 in use", "conflict")
|
||
ctx = build_context(ops_conn, "SRV-001", "sysadmin", "srv")
|
||
assert "decisions" in ctx
|
||
assert "modules" in ctx
|
||
assert len(ctx["modules"]) == 1
|
||
|
||
def test_format_prompt_includes_ssh_connection_section(self, ops_conn):
|
||
"""KIN-071: format_prompt добавляет '## SSH Connection' для operations."""
|
||
ctx = build_context(ops_conn, "SRV-001", "sysadmin", "srv")
|
||
prompt = format_prompt(ctx, "sysadmin", "You are sysadmin.")
|
||
assert "## SSH Connection" in prompt
|
||
assert "10.0.0.1" in prompt
|
||
assert "root" in prompt
|
||
assert "jumpt" in prompt
|
||
|
||
def test_format_prompt_no_ssh_section_for_development(self):
|
||
"""KIN-071: development-проект не получает SSH-секцию в prompt."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "dev", "Dev", "/path")
|
||
models.create_task(c, "DEV-001", "dev", "A task")
|
||
ctx = build_context(c, "DEV-001", "backend_dev", "dev")
|
||
prompt = format_prompt(ctx, "backend_dev", "You are a dev.")
|
||
assert "## SSH Connection" not in prompt
|
||
c.close()
|
||
|
||
def test_format_prompt_includes_project_type(self, ops_conn):
|
||
"""KIN-071: format_prompt включает Project type в секцию проекта."""
|
||
ctx = build_context(ops_conn, "SRV-001", "sysadmin", "srv")
|
||
prompt = format_prompt(ctx, "sysadmin", "You are sysadmin.")
|
||
assert "Project type: operations" in prompt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# KIN-071: PM routing — operations project routes PM to infra_* pipelines
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestPMRoutingOperations:
|
||
"""PM-контекст для operations-проекта должен содержать infra-маршруты,
|
||
не включающие architect/frontend_dev."""
|
||
|
||
@pytest.fixture
|
||
def ops_conn(self):
|
||
c = init_db(":memory:")
|
||
models.create_project(
|
||
c, "srv", "My Server", "",
|
||
project_type="operations",
|
||
ssh_host="10.0.0.1",
|
||
ssh_user="root",
|
||
)
|
||
models.create_task(c, "SRV-001", "srv", "Scan server")
|
||
yield c
|
||
c.close()
|
||
|
||
def test_pm_context_has_operations_project_type(self, ops_conn):
|
||
"""PM получает project_type=operations в контексте проекта."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
assert ctx["project"]["project_type"] == "operations"
|
||
|
||
def test_pm_context_has_infra_scan_route(self, ops_conn):
|
||
"""PM-контекст содержит маршрут infra_scan из specialists.yaml."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
assert "infra_scan" in ctx["routes"]
|
||
|
||
def test_pm_context_has_infra_debug_route(self, ops_conn):
|
||
"""PM-контекст содержит маршрут infra_debug из specialists.yaml."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
assert "infra_debug" in ctx["routes"]
|
||
|
||
def test_infra_scan_route_uses_sysadmin(self, ops_conn):
|
||
"""infra_scan маршрут включает sysadmin в шагах."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
steps = ctx["routes"]["infra_scan"]["steps"]
|
||
assert "sysadmin" in steps
|
||
|
||
def test_infra_scan_route_excludes_architect(self, ops_conn):
|
||
"""infra_scan маршрут не назначает architect."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
steps = ctx["routes"]["infra_scan"]["steps"]
|
||
assert "architect" not in steps
|
||
|
||
def test_infra_scan_route_excludes_frontend_dev(self, ops_conn):
|
||
"""infra_scan маршрут не назначает frontend_dev."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
steps = ctx["routes"]["infra_scan"]["steps"]
|
||
assert "frontend_dev" not in steps
|
||
|
||
def test_format_prompt_pm_operations_project_type_label(self, ops_conn):
|
||
"""format_prompt для PM с operations-проектом содержит 'Project type: operations'."""
|
||
ctx = build_context(ops_conn, "SRV-001", "pm", "srv")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "Project type: operations" in prompt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# KIN-090: Attachments — context builder includes attachment paths
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestAttachmentsInContext:
|
||
"""KIN-090: AC2 — агенты получают пути к вложениям в контексте задачи."""
|
||
|
||
@pytest.fixture
|
||
def conn_with_attachments(self):
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Project", "/tmp/prj")
|
||
models.create_task(c, "PRJ-001", "prj", "Fix bug")
|
||
models.create_attachment(
|
||
c, "PRJ-001", "screenshot.png",
|
||
"/tmp/prj/.kin/attachments/PRJ-001/screenshot.png",
|
||
"image/png", 1024,
|
||
)
|
||
models.create_attachment(
|
||
c, "PRJ-001", "mockup.jpg",
|
||
"/tmp/prj/.kin/attachments/PRJ-001/mockup.jpg",
|
||
"image/jpeg", 2048,
|
||
)
|
||
yield c
|
||
c.close()
|
||
|
||
def test_build_context_includes_attachments(self, conn_with_attachments):
|
||
"""KIN-090: AC2 — build_context включает вложения в контекст для всех ролей."""
|
||
ctx = build_context(conn_with_attachments, "PRJ-001", "debugger", "prj")
|
||
assert "attachments" in ctx
|
||
assert len(ctx["attachments"]) == 2
|
||
|
||
def test_build_context_attachments_have_filename_and_path(self, conn_with_attachments):
|
||
"""KIN-090: вложения в контексте содержат filename и path."""
|
||
ctx = build_context(conn_with_attachments, "PRJ-001", "debugger", "prj")
|
||
filenames = {a["filename"] for a in ctx["attachments"]}
|
||
paths = {a["path"] for a in ctx["attachments"]}
|
||
assert "screenshot.png" in filenames
|
||
assert "mockup.jpg" in filenames
|
||
assert "/tmp/prj/.kin/attachments/PRJ-001/screenshot.png" in paths
|
||
|
||
def test_build_context_attachments_key_always_present(self, conn):
|
||
"""KIN-094 #213: ключ 'attachments' всегда присутствует в контексте (пустой список если нет вложений)."""
|
||
# conn fixture has no attachments
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
assert "attachments" in ctx
|
||
assert ctx["attachments"] == []
|
||
|
||
def test_all_roles_get_attachments(self, conn_with_attachments):
|
||
"""KIN-090: AC2 — все роли (debugger, pm, tester, reviewer) получают вложения."""
|
||
for role in ("debugger", "pm", "tester", "reviewer", "backend_dev", "frontend_dev"):
|
||
ctx = build_context(conn_with_attachments, "PRJ-001", role, "prj")
|
||
assert "attachments" in ctx, f"Role '{role}' did not receive attachments"
|
||
|
||
def test_format_prompt_includes_attachments_section(self, conn_with_attachments):
|
||
"""KIN-090: format_prompt включает секцию '## Attachments' с именами и путями."""
|
||
ctx = build_context(conn_with_attachments, "PRJ-001", "debugger", "prj")
|
||
prompt = format_prompt(ctx, "debugger", "You are a debugger.")
|
||
assert "## Attachments" in prompt
|
||
assert "screenshot.png" in prompt
|
||
assert "/tmp/prj/.kin/attachments/PRJ-001/screenshot.png" in prompt
|
||
|
||
def test_format_prompt_no_attachments_section_when_none(self, conn):
|
||
"""KIN-090: format_prompt не добавляет секцию вложений, если их нет."""
|
||
ctx = build_context(conn, "VDOL-001", "debugger", "vdol")
|
||
prompt = format_prompt(ctx, "debugger", "Debug this.")
|
||
assert "## Attachments" not in prompt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# KIN-094: Attachments — ctx["attachments"] always present + inline text content
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestAttachmentsKIN094:
|
||
"""KIN-094: AC3 — PM и другие агенты всегда получают ключ attachments в контексте;
|
||
текстовые файлы <= 32 KB вставляются inline в промпт."""
|
||
|
||
@pytest.fixture
|
||
def conn_no_attachments(self):
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", "/tmp/prj")
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
yield c
|
||
c.close()
|
||
|
||
@pytest.fixture
|
||
def conn_text_attachment(self, tmp_path):
|
||
"""Проект с текстовым вложением <= 32 KB на диске."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", str(tmp_path))
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
txt_file = tmp_path / "spec.txt"
|
||
txt_file.write_text("Привет, это спека задачи", encoding="utf-8")
|
||
models.create_attachment(
|
||
c, "PRJ-001", "spec.txt", str(txt_file), "text/plain", txt_file.stat().st_size,
|
||
)
|
||
yield c
|
||
c.close()
|
||
|
||
@pytest.fixture
|
||
def conn_md_attachment(self, tmp_path):
|
||
"""Проект с .md вложением (text/markdown или определяется по расширению)."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", str(tmp_path))
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
md_file = tmp_path / "README.md"
|
||
md_file.write_text("# Title\n\nContent of readme", encoding="utf-8")
|
||
models.create_attachment(
|
||
c, "PRJ-001", "README.md", str(md_file), "text/markdown", md_file.stat().st_size,
|
||
)
|
||
yield c
|
||
c.close()
|
||
|
||
@pytest.fixture
|
||
def conn_json_attachment(self, tmp_path):
|
||
"""Проект с JSON-вложением (application/json)."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", str(tmp_path))
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
json_file = tmp_path / "config.json"
|
||
json_file.write_text('{"key": "value"}', encoding="utf-8")
|
||
models.create_attachment(
|
||
c, "PRJ-001", "config.json", str(json_file), "application/json", json_file.stat().st_size,
|
||
)
|
||
yield c
|
||
c.close()
|
||
|
||
@pytest.fixture
|
||
def conn_large_text_attachment(self, tmp_path):
|
||
"""Проект с текстовым вложением > 32 KB (не должно инлайниться)."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", str(tmp_path))
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
big_file = tmp_path / "big.txt"
|
||
big_file.write_text("x" * (32 * 1024 + 1), encoding="utf-8")
|
||
models.create_attachment(
|
||
c, "PRJ-001", "big.txt", str(big_file), "text/plain", big_file.stat().st_size,
|
||
)
|
||
yield c
|
||
c.close()
|
||
|
||
@pytest.fixture
|
||
def conn_image_attachment(self, tmp_path):
|
||
"""Проект с бинарным PNG-вложением (не должно инлайниться)."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", str(tmp_path))
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
png_file = tmp_path / "screen.png"
|
||
png_file.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64)
|
||
models.create_attachment(
|
||
c, "PRJ-001", "screen.png", str(png_file), "image/png", png_file.stat().st_size,
|
||
)
|
||
yield c
|
||
c.close()
|
||
|
||
# ------------------------------------------------------------------
|
||
# ctx["attachments"] always present
|
||
# ------------------------------------------------------------------
|
||
|
||
def test_pm_context_attachments_empty_list_when_no_attachments(self, conn_no_attachments):
|
||
"""KIN-094: PM получает пустой список attachments, а не отсутствующий ключ."""
|
||
ctx = build_context(conn_no_attachments, "PRJ-001", "pm", "prj")
|
||
assert "attachments" in ctx
|
||
assert ctx["attachments"] == []
|
||
|
||
def test_all_roles_attachments_key_present_when_empty(self, conn_no_attachments):
|
||
"""KIN-094: все роли получают ключ attachments (пустой список) даже без вложений."""
|
||
for role in ("pm", "debugger", "tester", "reviewer", "backend_dev", "frontend_dev", "architect"):
|
||
ctx = build_context(conn_no_attachments, "PRJ-001", role, "prj")
|
||
assert "attachments" in ctx, f"Role '{role}' missing 'attachments' key"
|
||
assert isinstance(ctx["attachments"], list), f"Role '{role}': attachments is not a list"
|
||
|
||
# ------------------------------------------------------------------
|
||
# Inline content for small text files
|
||
# ------------------------------------------------------------------
|
||
|
||
def test_format_prompt_inlines_small_text_file_content(self, conn_text_attachment):
|
||
"""KIN-094: содержимое текстового файла <= 32 KB вставляется inline в промпт."""
|
||
ctx = build_context(conn_text_attachment, "PRJ-001", "pm", "prj")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "Привет, это спека задачи" in prompt
|
||
|
||
def test_format_prompt_inlines_text_file_in_code_block(self, conn_text_attachment):
|
||
"""KIN-094: inline-контент обёрнут в блок кода (``` ... ```)."""
|
||
ctx = build_context(conn_text_attachment, "PRJ-001", "pm", "prj")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "```" in prompt
|
||
|
||
def test_format_prompt_inlines_md_file_by_extension(self, conn_md_attachment):
|
||
"""KIN-094: .md файл определяется по расширению и вставляется inline."""
|
||
ctx = build_context(conn_md_attachment, "PRJ-001", "pm", "prj")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "# Title" in prompt
|
||
assert "Content of readme" in prompt
|
||
|
||
def test_format_prompt_inlines_json_file_by_mime(self, conn_json_attachment):
|
||
"""KIN-094: application/json файл вставляется inline по MIME-типу."""
|
||
ctx = build_context(conn_json_attachment, "PRJ-001", "pm", "prj")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert '"key": "value"' in prompt
|
||
|
||
# ------------------------------------------------------------------
|
||
# NOT inlined: binary and large files
|
||
# ------------------------------------------------------------------
|
||
|
||
def test_format_prompt_does_not_inline_image_file(self, conn_image_attachment):
|
||
"""KIN-094: бинарный PNG файл НЕ вставляется inline."""
|
||
ctx = build_context(conn_image_attachment, "PRJ-001", "pm", "prj")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
# File is listed in ## Attachments section but no ``` block with binary content
|
||
assert "screen.png" in prompt # listed
|
||
assert "image/png" in prompt
|
||
# Should not contain raw binary or ``` code block for the PNG
|
||
# We verify the file content (PNG header) is NOT inlined
|
||
assert "\x89PNG" not in prompt
|
||
|
||
def test_format_prompt_does_not_inline_large_text_file(self, conn_large_text_attachment):
|
||
"""KIN-094: текстовый файл > 32 KB НЕ вставляется inline."""
|
||
ctx = build_context(conn_large_text_attachment, "PRJ-001", "pm", "prj")
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "big.txt" in prompt # listed
|
||
# Content should NOT be inlined (32KB+1 of 'x' chars)
|
||
assert "x" * 100 not in prompt
|
||
|
||
# ------------------------------------------------------------------
|
||
# Resilience: missing file on disk
|
||
# ------------------------------------------------------------------
|
||
|
||
def test_format_prompt_handles_missing_file_gracefully(self, tmp_path):
|
||
"""KIN-094: если файл отсутствует на диске, format_prompt не падает."""
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "prj", "Prj", str(tmp_path))
|
||
models.create_task(c, "PRJ-001", "prj", "Task")
|
||
# Register attachment pointing to non-existent file
|
||
models.create_attachment(
|
||
c, "PRJ-001", "missing.txt",
|
||
str(tmp_path / "missing.txt"),
|
||
"text/plain", 100,
|
||
)
|
||
ctx = build_context(c, "PRJ-001", "pm", "prj")
|
||
# Should not raise — exception is caught silently
|
||
prompt = format_prompt(ctx, "pm", "You are PM.")
|
||
assert "missing.txt" in prompt # still listed
|
||
c.close()
|
||
|
||
# ------------------------------------------------------------------
|
||
# PM pipeline: attachments available in brief context
|
||
# ------------------------------------------------------------------
|
||
|
||
def test_pm_context_includes_attachment_paths_for_pipeline(self, conn_text_attachment):
|
||
"""KIN-094: PM-агент получает пути к вложениям в контексте для старта pipeline."""
|
||
ctx = build_context(conn_text_attachment, "PRJ-001", "pm", "prj")
|
||
assert len(ctx["attachments"]) == 1
|
||
att = ctx["attachments"][0]
|
||
assert att["filename"] == "spec.txt"
|
||
assert att["mime_type"] == "text/plain"
|
||
assert "path" in att
|