"""Tests for core/hooks.py — post-pipeline hook execution.""" import subprocess import pytest from unittest.mock import patch, MagicMock from core.db import init_db from core import models from core.hooks import ( create_hook, get_hooks, update_hook, delete_hook, run_hooks, get_hook_logs, HookResult, _substitute_vars, ) @pytest.fixture def conn(): c = init_db(":memory:") models.create_project(c, "vdol", "ВДОЛЬ", "~/projects/vdolipoperek", tech_stack=["vue3"]) models.create_task(c, "VDOL-001", "vdol", "Fix bug") yield c c.close() @pytest.fixture def frontend_hook(conn): return create_hook( conn, project_id="vdol", name="rebuild-frontend", event="pipeline_completed", command="npm run build", trigger_module_path="web/frontend/*", working_dir="/tmp", timeout_seconds=60, ) # --------------------------------------------------------------------------- # CRUD # --------------------------------------------------------------------------- class TestCrud: def test_create_hook(self, conn): hook = create_hook(conn, "vdol", "my-hook", "pipeline_completed", "make build") assert hook["id"] is not None assert hook["project_id"] == "vdol" assert hook["name"] == "my-hook" assert hook["command"] == "make build" assert hook["enabled"] == 1 def test_get_hooks_by_project(self, conn, frontend_hook): hooks = get_hooks(conn, "vdol") assert len(hooks) == 1 assert hooks[0]["name"] == "rebuild-frontend" def test_get_hooks_filter_by_event(self, conn, frontend_hook): create_hook(conn, "vdol", "other", "step_completed", "echo done") hooks = get_hooks(conn, "vdol", event="pipeline_completed") assert len(hooks) == 1 assert hooks[0]["name"] == "rebuild-frontend" def test_get_hooks_disabled_excluded(self, conn, frontend_hook): update_hook(conn, frontend_hook["id"], enabled=0) hooks = get_hooks(conn, "vdol", enabled_only=True) assert len(hooks) == 0 def test_get_hooks_disabled_included_when_flag_off(self, conn, frontend_hook): update_hook(conn, frontend_hook["id"], enabled=0) hooks = get_hooks(conn, "vdol", enabled_only=False) assert len(hooks) == 1 def test_update_hook(self, conn, frontend_hook): update_hook(conn, frontend_hook["id"], command="npm run build:prod", timeout_seconds=180) hooks = get_hooks(conn, "vdol", enabled_only=False) assert hooks[0]["command"] == "npm run build:prod" assert hooks[0]["timeout_seconds"] == 180 def test_delete_hook(self, conn, frontend_hook): delete_hook(conn, frontend_hook["id"]) hooks = get_hooks(conn, "vdol", enabled_only=False) assert len(hooks) == 0 def test_get_hooks_wrong_project(self, conn, frontend_hook): hooks = get_hooks(conn, "nonexistent") assert hooks == [] # --------------------------------------------------------------------------- # Module matching (fnmatch) # --------------------------------------------------------------------------- class TestModuleMatching: def _make_proc(self, returncode=0, stdout="ok", stderr=""): m = MagicMock() m.returncode = returncode m.stdout = stdout m.stderr = stderr return m @patch("core.hooks.subprocess.run") def test_hook_runs_when_module_matches(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc() modules = [{"path": "web/frontend/App.vue", "name": "App"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) assert len(results) == 1 assert results[0].name == "rebuild-frontend" mock_run.assert_called_once() @patch("core.hooks.subprocess.run") def test_hook_skipped_when_no_module_matches(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc() modules = [{"path": "core/models.py", "name": "models"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) assert len(results) == 0 mock_run.assert_not_called() @patch("core.hooks.subprocess.run") def test_hook_runs_without_module_filter(self, mock_run, conn): mock_run.return_value = self._make_proc() create_hook(conn, "vdol", "always-run", "pipeline_completed", "echo done") results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=[]) assert len(results) == 1 @patch("core.hooks.subprocess.run") def test_hook_skipped_on_wrong_event(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc() modules = [{"path": "web/frontend/App.vue", "name": "App"}] results = run_hooks(conn, "vdol", "VDOL-001", event="step_completed", task_modules=modules) assert len(results) == 0 @patch("core.hooks.subprocess.run") def test_hook_skipped_when_disabled(self, mock_run, conn, frontend_hook): update_hook(conn, frontend_hook["id"], enabled=0) mock_run.return_value = self._make_proc() modules = [{"path": "web/frontend/App.vue", "name": "App"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) assert len(results) == 0 # --------------------------------------------------------------------------- # Execution and logging # --------------------------------------------------------------------------- class TestExecution: def _make_proc(self, returncode=0, stdout="built!", stderr=""): m = MagicMock() m.returncode = returncode m.stdout = stdout m.stderr = stderr return m @patch("core.hooks.subprocess.run") def test_successful_hook_result(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc(returncode=0, stdout="built!") modules = [{"path": "web/frontend/index.ts", "name": "index"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) r = results[0] assert r.success is True assert r.exit_code == 0 assert r.output == "built!" @patch("core.hooks.subprocess.run") def test_failed_hook_result(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc(returncode=1, stderr="Module not found") modules = [{"path": "web/frontend/index.ts", "name": "index"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) r = results[0] assert r.success is False assert r.exit_code == 1 assert "Module not found" in r.error @patch("core.hooks.subprocess.run") def test_hook_run_logged_to_db(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc(returncode=0, stdout="ok") modules = [{"path": "web/frontend/App.vue", "name": "App"}] run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) logs = get_hook_logs(conn, project_id="vdol") assert len(logs) == 1 assert logs[0]["hook_id"] == frontend_hook["id"] assert logs[0]["task_id"] == "VDOL-001" assert logs[0]["success"] == 1 assert logs[0]["exit_code"] == 0 assert logs[0]["output"] == "ok" @patch("core.hooks.subprocess.run") def test_failed_hook_logged_to_db(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc(returncode=2, stderr="error!") modules = [{"path": "web/frontend/App.vue", "name": "App"}] run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) logs = get_hook_logs(conn, project_id="vdol") assert logs[0]["success"] == 0 assert logs[0]["exit_code"] == 2 assert "error!" in logs[0]["error"] @patch("core.hooks.subprocess.run") def test_timeout_handled_gracefully(self, mock_run, conn, frontend_hook): mock_run.side_effect = subprocess.TimeoutExpired(cmd="npm run build", timeout=60) modules = [{"path": "web/frontend/App.vue", "name": "App"}] # Must not raise results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) r = results[0] assert r.success is False assert r.exit_code == 124 assert "timed out" in r.error logs = get_hook_logs(conn, project_id="vdol") assert logs[0]["success"] == 0 @patch("core.hooks.subprocess.run") def test_exception_handled_gracefully(self, mock_run, conn, frontend_hook): mock_run.side_effect = OSError("npm not found") modules = [{"path": "web/frontend/App.vue", "name": "App"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) r = results[0] assert r.success is False assert "npm not found" in r.error @patch("core.hooks.subprocess.run") def test_command_uses_working_dir(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc() modules = [{"path": "web/frontend/App.vue", "name": "App"}] run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) call_kwargs = mock_run.call_args[1] assert call_kwargs["cwd"] == "/tmp" @patch("core.hooks.subprocess.run") def test_shell_true_used(self, mock_run, conn, frontend_hook): mock_run.return_value = self._make_proc() modules = [{"path": "web/frontend/App.vue", "name": "App"}] run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) call_kwargs = mock_run.call_args[1] assert call_kwargs["shell"] is True # --------------------------------------------------------------------------- # get_hook_logs filters # --------------------------------------------------------------------------- class TestGetHookLogs: @patch("core.hooks.subprocess.run") def test_filter_by_hook_id(self, mock_run, conn, frontend_hook): mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") hook2 = create_hook(conn, "vdol", "second", "pipeline_completed", "echo 2") modules = [{"path": "web/frontend/App.vue", "name": "App"}] run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) logs = get_hook_logs(conn, hook_id=frontend_hook["id"]) assert all(l["hook_id"] == frontend_hook["id"] for l in logs) @patch("core.hooks.subprocess.run") def test_limit_respected(self, mock_run, conn, frontend_hook): mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") modules = [{"path": "web/frontend/App.vue", "name": "App"}] for _ in range(5): run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=modules) logs = get_hook_logs(conn, project_id="vdol", limit=3) assert len(logs) == 3 # --------------------------------------------------------------------------- # Variable substitution in hook commands # --------------------------------------------------------------------------- class TestSubstituteVars: def test_substitutes_task_id_and_title(self, conn): result = _substitute_vars( 'git commit -m "kin: {task_id} {title}"', "VDOL-001", conn, ) assert result == 'git commit -m "kin: VDOL-001 Fix bug"' def test_no_substitution_when_task_id_is_none(self, conn): cmd = 'git commit -m "kin: {task_id} {title}"' result = _substitute_vars(cmd, None, conn) assert result == cmd def test_sanitizes_double_quotes_in_title(self, conn): conn.execute('UPDATE tasks SET title = ? WHERE id = ?', ('Fix "bug" here', "VDOL-001")) conn.commit() result = _substitute_vars( 'git commit -m "kin: {task_id} {title}"', "VDOL-001", conn, ) assert '"' not in result.split('"kin:')[1].split('"')[0] assert "Fix 'bug' here" in result def test_sanitizes_newlines_in_title(self, conn): conn.execute('UPDATE tasks SET title = ? WHERE id = ?', ("Fix\nbug\r\nhere", "VDOL-001")) conn.commit() result = _substitute_vars("{title}", "VDOL-001", conn) assert "\n" not in result assert "\r" not in result def test_unknown_task_id_uses_empty_title(self, conn): result = _substitute_vars("{task_id} {title}", "NONEXISTENT", conn) assert result == "NONEXISTENT " def test_no_placeholders_returns_command_unchanged(self, conn): cmd = "npm run build" result = _substitute_vars(cmd, "VDOL-001", conn) assert result == cmd @patch("core.hooks.subprocess.run") def test_autocommit_hook_command_substituted(self, mock_run, conn): """auto-commit hook должен получать реальные task_id и title в команде.""" mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") create_hook(conn, "vdol", "auto-commit", "task_done", 'git add -A && git commit -m "kin: {task_id} {title}"', working_dir="/tmp") run_hooks(conn, "vdol", "VDOL-001", event="task_done", task_modules=[]) call_kwargs = mock_run.call_args[1] # shell=True: command is the first positional arg command = mock_run.call_args[0][0] assert "VDOL-001" in command assert "Fix bug" in command # --------------------------------------------------------------------------- # KIN-050: rebuild-frontend hook — unconditional firing after pipeline # --------------------------------------------------------------------------- class TestRebuildFrontendHookSetup: """Regression tests for KIN-050. Баг: rebuild-frontend не срабатывал, если pipeline не трогал web/frontend/*. Фикс: убран trigger_module_path из hook_setup — хук должен срабатывать всегда. """ def test_rebuild_frontend_created_without_trigger_module_path(self, conn): """rebuild-frontend hook должен быть создан без trigger_module_path (KIN-050). Воспроизводит логику hook_setup: создаём хук без фильтра и убеждаемся, что он сохраняется в БД с trigger_module_path=NULL. """ hook = create_hook( conn, "vdol", name="rebuild-frontend", event="pipeline_completed", command="scripts/rebuild-frontend.sh", trigger_module_path=None, # фикс KIN-050: без фильтра working_dir="/tmp", timeout_seconds=300, ) assert hook["trigger_module_path"] is None, ( "trigger_module_path должен быть NULL — хук не должен фильтровать по модулям" ) # Перечитываем из БД — убеждаемся, что NULL сохранился hooks = get_hooks(conn, "vdol", enabled_only=False) rebuild = next((h for h in hooks if h["name"] == "rebuild-frontend"), None) assert rebuild is not None assert rebuild["trigger_module_path"] is None @patch("core.hooks.subprocess.run") def test_rebuild_frontend_fires_when_only_backend_modules_changed(self, mock_run, conn): """Хук без trigger_module_path должен срабатывать при изменении backend-файлов. Регрессия KIN-050: раньше хук молчал, если не было web/frontend/* файлов. """ mock_run.return_value = MagicMock(returncode=0, stdout="built!", stderr="") create_hook( conn, "vdol", "rebuild-frontend", "pipeline_completed", "npm run build", trigger_module_path=None, # фикс: нет фильтра working_dir="/tmp", ) backend_modules = [ {"path": "core/models.py", "name": "models"}, {"path": "web/api.py", "name": "api"}, ] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=backend_modules) assert len(results) == 1, "Хук должен сработать несмотря на отсутствие frontend-файлов" assert results[0].name == "rebuild-frontend" assert results[0].success is True mock_run.assert_called_once() @patch("core.hooks.subprocess.run") def test_rebuild_frontend_fires_exactly_once_per_pipeline(self, mock_run, conn): """Хук rebuild-frontend должен срабатывать ровно один раз за pipeline_completed.""" mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") create_hook( conn, "vdol", "rebuild-frontend", "pipeline_completed", "npm run build", trigger_module_path=None, working_dir="/tmp", ) any_modules = [ {"path": "core/hooks.py", "name": "hooks"}, {"path": "web/frontend/App.vue", "name": "App"}, {"path": "web/api.py", "name": "api"}, ] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=any_modules) assert len(results) == 1, "Хук должен выполниться ровно один раз" mock_run.assert_called_once() @patch("core.hooks.subprocess.run") def test_rebuild_frontend_fires_with_empty_module_list(self, mock_run, conn): """Хук без trigger_module_path должен срабатывать даже с пустым списком модулей.""" mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") create_hook( conn, "vdol", "rebuild-frontend", "pipeline_completed", "npm run build", trigger_module_path=None, working_dir="/tmp", ) results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=[]) assert len(results) == 1 assert results[0].name == "rebuild-frontend" mock_run.assert_called_once() @patch("core.hooks.subprocess.run") def test_rebuild_frontend_with_module_path_skips_non_frontend(self, mock_run, conn): """Контрольный тест: хук С trigger_module_path НЕ срабатывает на backend-файлы. Подтверждает, что фикс (удаление trigger_module_path) был необходим. """ mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") create_hook( conn, "vdol", "rebuild-frontend-filtered", "pipeline_completed", "npm run build", trigger_module_path="web/frontend/*", # старое (сломанное) поведение working_dir="/tmp", ) backend_modules = [{"path": "core/models.py", "name": "models"}] results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=backend_modules) assert len(results) == 0, ( "Хук с trigger_module_path НЕ должен срабатывать на backend-файлы — " "именно это было первопричиной бага KIN-050" ) # --------------------------------------------------------------------------- # KIN-052: rebuild-frontend hook — команда cd+&& и персистентность в БД # --------------------------------------------------------------------------- class TestKIN052RebuildFrontendCommand: """Регрессионные тесты для KIN-052. Хук rebuild-frontend использует команду вида: cd /path/to/frontend && npm run build — то есть цепочку shell-команд без working_dir. Тесты проверяют, что такая форма работает корректно и хук переживает пересоздание соединения с БД (симуляция рестарта). """ @patch("core.hooks.subprocess.run") def test_cd_chained_command_passes_as_string_to_shell(self, mock_run, conn): """Команда с && должна передаваться в subprocess как строка (не список) с shell=True. Если передать список ['cd', '/path', '&&', 'npm', 'run', 'build'] с shell=True, shell проигнорирует аргументы после первого. Строковая форма обязательна. """ mock_run.return_value = MagicMock(returncode=0, stdout="built!", stderr="") cmd = "cd /Users/grosfrumos/projects/kin/web/frontend && npm run build" create_hook(conn, "vdol", "rebuild-frontend", "pipeline_completed", cmd, trigger_module_path=None, working_dir=None) run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=[]) call_args = mock_run.call_args passed_cmd = call_args[0][0] assert isinstance(passed_cmd, str), ( "Команда с && должна передаваться как строка, иначе shell не раскроет &&" ) assert "&&" in passed_cmd assert call_args[1].get("shell") is True @patch("core.hooks.subprocess.run") def test_cd_command_without_working_dir_uses_cwd_none(self, mock_run, conn): """Хук с cd-командой и working_dir=None должен вызывать subprocess с cwd=None. Директория смены задаётся через cd в самой команде, а не через cwd. """ mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") cmd = "cd /Users/grosfrumos/projects/kin/web/frontend && npm run build" create_hook(conn, "vdol", "rebuild-frontend", "pipeline_completed", cmd, trigger_module_path=None, working_dir=None) run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=[]) cwd = mock_run.call_args[1].get("cwd") assert cwd is None, ( f"cwd должен быть None когда working_dir не задан, получили: {cwd!r}" ) @patch("core.hooks.subprocess.run") def test_cd_command_exits_zero_returns_success(self, mock_run, conn): """Хук с cd+npm run build при returncode=0 должен вернуть success=True.""" mock_run.return_value = MagicMock(returncode=0, stdout="✓ build complete", stderr="") cmd = "cd /Users/grosfrumos/projects/kin/web/frontend && npm run build" create_hook(conn, "vdol", "rebuild-frontend", "pipeline_completed", cmd, trigger_module_path=None) results = run_hooks(conn, "vdol", "VDOL-001", event="pipeline_completed", task_modules=[]) assert len(results) == 1 assert results[0].success is True assert results[0].name == "rebuild-frontend" @patch("core.hooks.subprocess.run") def test_hook_persists_after_db_reconnect(self, mock_run): """Хук должен сохраняться в файловой БД и быть доступен после пересоздания соединения. Симулирует рестарт: создаём хук, закрываем соединение, открываем новое — хук на месте. """ import tempfile import os from core.db import init_db with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: db_path = f.name try: # Первое соединение — создаём проект и хук conn1 = init_db(db_path) from core import models as _models _models.create_project(conn1, "kin", "Kin", "/projects/kin", tech_stack=["vue3"]) cmd = "cd /Users/grosfrumos/projects/kin/web/frontend && npm run build" hook = create_hook(conn1, "kin", "rebuild-frontend", "pipeline_completed", cmd, trigger_module_path=None) hook_id = hook["id"] conn1.close() # Второе соединение — «рестарт», хук должен быть на месте conn2 = init_db(db_path) hooks = get_hooks(conn2, "kin", event="pipeline_completed", enabled_only=True) conn2.close() assert len(hooks) == 1, "После пересоздания соединения хук должен оставаться в БД" assert hooks[0]["id"] == hook_id assert hooks[0]["name"] == "rebuild-frontend" assert hooks[0]["command"] == cmd assert hooks[0]["trigger_module_path"] is None finally: os.unlink(db_path)