"""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, ) @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