314 lines
13 KiB
Python
314 lines
13 KiB
Python
"""Tests for core/followup.py — follow-up task generation with permission handling."""
|
||
|
||
import json
|
||
import pytest
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
from core.db import init_db
|
||
from core import models
|
||
from core.followup import (
|
||
generate_followups, resolve_pending_action, auto_resolve_pending_actions,
|
||
_collect_pipeline_output, _next_task_id, _is_permission_blocked,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def conn():
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "vdol", "ВДОЛЬ", "~/projects/vdolipoperek",
|
||
tech_stack=["vue3"], language="ru")
|
||
models.create_task(c, "VDOL-001", "vdol", "Security audit",
|
||
status="done", brief={"route_type": "security_audit"})
|
||
models.log_agent_run(c, "vdol", "security", "execute",
|
||
task_id="VDOL-001",
|
||
output_summary=json.dumps({
|
||
"summary": "8 уязвимостей найдено",
|
||
"findings": [
|
||
{"severity": "HIGH", "title": "Admin endpoint без auth",
|
||
"file": "index.js", "line": 42},
|
||
{"severity": "MEDIUM", "title": "Нет rate limiting на login",
|
||
"file": "auth.js", "line": 15},
|
||
],
|
||
}, ensure_ascii=False),
|
||
success=True)
|
||
yield c
|
||
c.close()
|
||
|
||
|
||
class TestCollectPipelineOutput:
|
||
def test_collects_all_steps(self, conn):
|
||
output = _collect_pipeline_output(conn, "VDOL-001")
|
||
assert "security" in output
|
||
assert "Admin endpoint" in output
|
||
|
||
def test_empty_for_no_logs(self, conn):
|
||
assert _collect_pipeline_output(conn, "NONEXISTENT") == ""
|
||
|
||
|
||
class TestNextTaskId:
|
||
def test_increments(self, conn):
|
||
assert _next_task_id(conn, "vdol") == "VDOL-002"
|
||
|
||
def test_handles_obs_ids(self, conn):
|
||
models.create_task(conn, "VDOL-OBS-001", "vdol", "Obsidian task")
|
||
assert _next_task_id(conn, "vdol") == "VDOL-002"
|
||
|
||
|
||
class TestIsPermissionBlocked:
|
||
def test_detects_permission_denied(self):
|
||
assert _is_permission_blocked({"title": "Fix X", "brief": "permission denied on write"})
|
||
|
||
def test_detects_manual_application_ru(self):
|
||
assert _is_permission_blocked({"title": "Ручное применение фикса для auth.js"})
|
||
|
||
def test_detects_no_write_permission_ru(self):
|
||
assert _is_permission_blocked({"title": "X", "brief": "не получили разрешение на запись"})
|
||
|
||
def test_detects_read_only(self):
|
||
assert _is_permission_blocked({"title": "Apply manually", "brief": "file is read-only"})
|
||
|
||
def test_normal_item_not_blocked(self):
|
||
assert not _is_permission_blocked({"title": "Fix admin auth", "brief": "Add requireAuth"})
|
||
|
||
def test_empty_item(self):
|
||
assert not _is_permission_blocked({})
|
||
|
||
|
||
class TestGenerateFollowups:
|
||
@patch("agents.runner._run_claude")
|
||
def test_creates_followup_tasks(self, mock_claude, conn):
|
||
mock_claude.return_value = {
|
||
"output": json.dumps([
|
||
{"title": "Fix admin auth", "type": "hotfix", "priority": 2,
|
||
"brief": "Add requireAuth to admin endpoints"},
|
||
{"title": "Add rate limiting", "type": "feature", "priority": 4,
|
||
"brief": "Rate limit login to 5/15min"},
|
||
]),
|
||
"returncode": 0,
|
||
}
|
||
|
||
result = generate_followups(conn, "VDOL-001")
|
||
|
||
assert len(result["created"]) == 2
|
||
assert len(result["pending_actions"]) == 0
|
||
assert result["created"][0]["id"] == "VDOL-002"
|
||
assert result["created"][0]["parent_task_id"] == "VDOL-001"
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_separates_permission_items(self, mock_claude, conn):
|
||
mock_claude.return_value = {
|
||
"output": json.dumps([
|
||
{"title": "Fix admin auth", "type": "hotfix", "priority": 2,
|
||
"brief": "Add requireAuth"},
|
||
{"title": "Ручное применение .dockerignore",
|
||
"type": "hotfix", "priority": 3,
|
||
"brief": "Не получили разрешение на запись в файл"},
|
||
{"title": "Apply CSP headers manually",
|
||
"type": "feature", "priority": 4,
|
||
"brief": "Permission denied writing nginx.conf"},
|
||
]),
|
||
"returncode": 0,
|
||
}
|
||
|
||
result = generate_followups(conn, "VDOL-001")
|
||
|
||
assert len(result["created"]) == 1 # Only "Fix admin auth"
|
||
assert result["created"][0]["title"] == "Fix admin auth"
|
||
assert len(result["pending_actions"]) == 2
|
||
assert result["pending_actions"][0]["type"] == "permission_fix"
|
||
assert "options" in result["pending_actions"][0]
|
||
assert "rerun" in result["pending_actions"][0]["options"]
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_handles_empty_response(self, mock_claude, conn):
|
||
mock_claude.return_value = {"output": "[]", "returncode": 0}
|
||
result = generate_followups(conn, "VDOL-001")
|
||
assert result["created"] == []
|
||
assert result["pending_actions"] == []
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_handles_wrapped_response(self, mock_claude, conn):
|
||
mock_claude.return_value = {
|
||
"output": json.dumps({"tasks": [
|
||
{"title": "Fix X", "priority": 3},
|
||
]}),
|
||
"returncode": 0,
|
||
}
|
||
result = generate_followups(conn, "VDOL-001")
|
||
assert len(result["created"]) == 1
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_handles_invalid_json(self, mock_claude, conn):
|
||
mock_claude.return_value = {"output": "not json", "returncode": 0}
|
||
result = generate_followups(conn, "VDOL-001")
|
||
assert result["created"] == []
|
||
|
||
def test_no_logs_returns_empty(self, conn):
|
||
models.create_task(conn, "VDOL-999", "vdol", "Empty task")
|
||
result = generate_followups(conn, "VDOL-999")
|
||
assert result["created"] == []
|
||
|
||
def test_nonexistent_task(self, conn):
|
||
result = generate_followups(conn, "NOPE")
|
||
assert result["created"] == []
|
||
|
||
def test_dry_run(self, conn):
|
||
result = generate_followups(conn, "VDOL-001", dry_run=True)
|
||
assert len(result["created"]) == 1
|
||
assert result["created"][0]["_dry_run"] is True
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_logs_generation(self, mock_claude, conn):
|
||
mock_claude.return_value = {
|
||
"output": json.dumps([{"title": "Fix A", "priority": 2}]),
|
||
"returncode": 0,
|
||
}
|
||
generate_followups(conn, "VDOL-001")
|
||
|
||
logs = conn.execute(
|
||
"SELECT * FROM agent_logs WHERE agent_role='followup_pm'"
|
||
).fetchall()
|
||
assert len(logs) == 1
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_prompt_includes_language(self, mock_claude, conn):
|
||
mock_claude.return_value = {"output": "[]", "returncode": 0}
|
||
generate_followups(conn, "VDOL-001")
|
||
prompt = mock_claude.call_args[0][0]
|
||
assert "Russian" in prompt
|
||
|
||
|
||
class TestResolvePendingAction:
|
||
def test_skip_returns_none(self, conn):
|
||
action = {"type": "permission_fix", "original_item": {"title": "X"}}
|
||
assert resolve_pending_action(conn, "VDOL-001", action, "skip") is None
|
||
|
||
def test_manual_task_creates_task(self, conn):
|
||
action = {
|
||
"type": "permission_fix",
|
||
"original_item": {"title": "Fix .dockerignore", "type": "hotfix",
|
||
"priority": 3, "brief": "Create .dockerignore"},
|
||
}
|
||
result = resolve_pending_action(conn, "VDOL-001", action, "manual_task")
|
||
assert result is not None
|
||
assert result["title"] == "Fix .dockerignore"
|
||
assert result["parent_task_id"] == "VDOL-001"
|
||
assert result["priority"] == 3
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_rerun_launches_pipeline(self, mock_claude, conn):
|
||
mock_claude.return_value = {
|
||
"output": json.dumps({"result": "applied fix"}),
|
||
"returncode": 0,
|
||
}
|
||
action = {
|
||
"type": "permission_fix",
|
||
"original_item": {"title": "Fix X", "type": "frontend_dev",
|
||
"brief": "Apply the fix"},
|
||
}
|
||
result = resolve_pending_action(conn, "VDOL-001", action, "rerun")
|
||
assert "rerun_result" in result
|
||
|
||
# Verify --dangerously-skip-permissions was passed
|
||
call_args = mock_claude.call_args
|
||
cmd = call_args[0][0] if call_args[0] else None
|
||
# _run_claude is called with allow_write=True which adds the flag
|
||
# Check via the cmd list in subprocess.run mock... but _run_claude
|
||
# is mocked at a higher level. Let's check the allow_write param.
|
||
# The pipeline calls run_agent with allow_write=True which calls
|
||
# _run_claude with allow_write=True
|
||
assert result["rerun_result"]["success"] is True
|
||
|
||
def test_manual_task_brief_has_task_type_manual_escalation(self, conn):
|
||
"""brief["task_type"] должен быть 'manual_escalation' — KIN-020."""
|
||
action = {
|
||
"type": "permission_fix",
|
||
"original_item": {"title": "Fix .dockerignore", "type": "hotfix",
|
||
"priority": 3, "brief": "Create .dockerignore"},
|
||
}
|
||
result = resolve_pending_action(conn, "VDOL-001", action, "manual_task")
|
||
assert result is not None
|
||
assert result["brief"]["task_type"] == "manual_escalation"
|
||
|
||
def test_manual_task_brief_includes_source(self, conn):
|
||
"""brief["source"] должен содержать ссылку на родительскую задачу — KIN-020."""
|
||
action = {
|
||
"type": "permission_fix",
|
||
"original_item": {"title": "Fix X"},
|
||
}
|
||
result = resolve_pending_action(conn, "VDOL-001", action, "manual_task")
|
||
assert result["brief"]["source"] == "followup:VDOL-001"
|
||
|
||
def test_manual_task_brief_includes_description(self, conn):
|
||
"""brief["description"] копируется из original_item.brief — KIN-020."""
|
||
action = {
|
||
"type": "permission_fix",
|
||
"original_item": {"title": "Fix Y", "brief": "Detailed context here"},
|
||
}
|
||
result = resolve_pending_action(conn, "VDOL-001", action, "manual_task")
|
||
assert result["brief"]["description"] == "Detailed context here"
|
||
|
||
def test_nonexistent_task(self, conn):
|
||
action = {"type": "permission_fix", "original_item": {}}
|
||
assert resolve_pending_action(conn, "NOPE", action, "skip") is None
|
||
|
||
|
||
class TestAutoResolvePendingActions:
|
||
@patch("agents.runner._run_claude")
|
||
def test_rerun_success_resolves_as_rerun(self, mock_claude, conn):
|
||
"""Успешный rerun должен резолвиться как 'rerun'."""
|
||
mock_claude.return_value = {
|
||
"output": json.dumps({"result": "fixed"}),
|
||
"returncode": 0,
|
||
}
|
||
action = {
|
||
"type": "permission_fix",
|
||
"description": "Fix X",
|
||
"original_item": {"title": "Fix X", "type": "frontend_dev", "brief": "Apply fix"},
|
||
"options": ["rerun", "manual_task", "skip"],
|
||
}
|
||
results = auto_resolve_pending_actions(conn, "VDOL-001", [action])
|
||
|
||
assert len(results) == 1
|
||
assert results[0]["resolved"] == "rerun"
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_rerun_failure_escalates_to_manual_task(self, mock_claude, conn):
|
||
"""Провал rerun должен создавать manual_task для эскалации."""
|
||
mock_claude.return_value = {"output": "", "returncode": 1}
|
||
action = {
|
||
"type": "permission_fix",
|
||
"description": "Fix X",
|
||
"original_item": {"title": "Fix X", "type": "frontend_dev", "brief": "Apply fix"},
|
||
"options": ["rerun", "manual_task", "skip"],
|
||
}
|
||
results = auto_resolve_pending_actions(conn, "VDOL-001", [action])
|
||
|
||
assert len(results) == 1
|
||
assert results[0]["resolved"] == "manual_task"
|
||
# Manual task должна быть создана в DB
|
||
tasks = models.list_tasks(conn, project_id="vdol")
|
||
assert len(tasks) == 2 # VDOL-001 + новая manual task
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_escalated_manual_task_has_task_type_manual_escalation(self, mock_claude, conn):
|
||
"""При эскалации после провала rerun созданная задача имеет task_type='manual_escalation' — KIN-020."""
|
||
mock_claude.return_value = {"output": "", "returncode": 1}
|
||
action = {
|
||
"type": "permission_fix",
|
||
"description": "Fix X",
|
||
"original_item": {"title": "Fix X", "type": "frontend_dev", "brief": "Apply fix"},
|
||
"options": ["rerun", "manual_task", "skip"],
|
||
}
|
||
results = auto_resolve_pending_actions(conn, "VDOL-001", [action])
|
||
|
||
assert results[0]["resolved"] == "manual_task"
|
||
created_task = results[0]["result"]
|
||
assert created_task["brief"]["task_type"] == "manual_escalation"
|
||
|
||
@patch("agents.runner._run_claude")
|
||
def test_empty_pending_actions(self, mock_claude, conn):
|
||
"""Пустой список — пустой результат."""
|
||
results = auto_resolve_pending_actions(conn, "VDOL-001", [])
|
||
assert results == []
|
||
mock_claude.assert_not_called()
|