"""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, _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_nonexistent_task(self, conn): action = {"type": "permission_fix", "original_item": {}} assert resolve_pending_action(conn, "NOPE", action, "skip") is None