diff --git a/tests/test_kin_115_backup.py b/tests/test_kin_115_backup.py new file mode 100644 index 0000000..9b3819a --- /dev/null +++ b/tests/test_kin_115_backup.py @@ -0,0 +1,220 @@ +"""Tests for KIN-115 — pre-deploy backup hook. + +Acceptance criteria: + AC-1: Deploy without backup is physically impossible (pre_deploy_backup is called + before any deploy step and abort on failure). + AC-2: Backup filename contains a timestamp (YYYYMMDD_HHMMSS format). + AC-3: Backup failure = deploy failure (execute_deploy returns success=False if backup fails). +""" + +import re +import sqlite3 +import pytest +from pathlib import Path +from unittest.mock import patch, MagicMock, call + +from core.db import init_db +from core import models +from core.deploy import execute_deploy, pre_deploy_backup # AC-1: must be importable + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def conn(): + c = init_db(db_path=":memory:") + yield c + c.close() + + +@pytest.fixture +def local_project(conn): + """Project without deploy_host — local SQLite backup path.""" + models.create_project(conn, "local-svc", "Local Service", "/srv/local-svc") + models.update_project(conn, "local-svc", deploy_runtime="python", deploy_path="/srv/local-svc") + return models.get_project(conn, "local-svc") + + +@pytest.fixture +def sqlite_project_with_db(tmp_path, conn): + """Project with a real kin.db file in project root for backup tests.""" + db_file = tmp_path / "kin.db" + db_file.write_bytes(b"SQLite data") + models.create_project(conn, "db-svc", "DB Service", str(tmp_path)) + models.update_project(conn, "db-svc", deploy_runtime="python", deploy_path=str(tmp_path)) + p = models.get_project(conn, "db-svc") + return p, db_file + + +# --------------------------------------------------------------------------- +# AC-1 + AC-3: pre_deploy_backup is called inside execute_deploy +# --------------------------------------------------------------------------- + +class TestPreDeployBackupIntegration: + """Verify that execute_deploy integrates backup as a mandatory pre-step.""" + + def test_execute_deploy_calls_pre_deploy_backup(self, conn, local_project): + """AC-1: pre_deploy_backup must be called before any deploy step.""" + call_order = [] + + def mock_backup(project): + call_order.append("backup") + return {"status": "ok", "path": "/tmp/kin.db.bak"} + + def mock_subprocess(*args, **kwargs): + call_order.append("deploy") + m = MagicMock() + m.returncode = 0 + m.stdout = "ok" + m.stderr = "" + return m + + with patch("core.deploy.pre_deploy_backup", side_effect=mock_backup), \ + patch("core.deploy.subprocess.run", side_effect=mock_subprocess): + execute_deploy(local_project, conn) + + assert "backup" in call_order, "pre_deploy_backup was never called" + if "deploy" in call_order: + assert call_order.index("backup") < call_order.index("deploy"), \ + "backup must happen BEFORE any deploy step" + + def test_execute_deploy_without_backup_is_impossible(self, conn, local_project): + """AC-1: execute_deploy result must contain 'backup' key proving hook ran.""" + with patch("core.deploy.pre_deploy_backup", return_value={"status": "ok", "path": "/tmp/x.bak"}), \ + patch("core.deploy.subprocess.run", return_value=MagicMock(returncode=0, stdout="", stderr="")): + result = execute_deploy(local_project, conn) + + assert "backup" in result, \ + "execute_deploy result must include 'backup' key — deploy without backup is impossible" + + def test_backup_failure_aborts_deploy(self, conn, local_project): + """AC-3: If backup raises an exception, deploy must not proceed.""" + deploy_called = [] + + def mock_subprocess(*args, **kwargs): + deploy_called.append(True) + m = MagicMock() + m.returncode = 0 + m.stdout = "ok" + m.stderr = "" + return m + + with patch("core.deploy.pre_deploy_backup", side_effect=PermissionError("no write permission")), \ + patch("core.deploy.subprocess.run", side_effect=mock_subprocess): + result = execute_deploy(local_project, conn) + + assert result["success"] is False, "Deploy must fail when backup fails" + assert not deploy_called, "Deploy steps must NOT execute when backup failed" + + def test_backup_failure_returns_error_in_result(self, conn, local_project): + """AC-3: Result must include error info when backup fails.""" + with patch("core.deploy.pre_deploy_backup", side_effect=PermissionError("no write")), \ + patch("core.deploy.subprocess.run"): + result = execute_deploy(local_project, conn) + + assert "error" in result or ("backup" in result and result.get("backup", {}).get("status") == "error"), \ + "Result must indicate why deploy was aborted (backup failure)" + + +# --------------------------------------------------------------------------- +# AC-2: Backup filename contains timestamp +# --------------------------------------------------------------------------- + +class TestBackupTimestamp: + """Verify that created backup file path contains a YYYYMMDD_HHMMSS timestamp.""" + + TIMESTAMP_RE = re.compile(r'\d{8}_\d{6}') + + def test_sqlite_backup_path_contains_timestamp(self, sqlite_project_with_db): + """AC-2: Backup file must have a timestamp in its name.""" + project, db_file = sqlite_project_with_db + result = pre_deploy_backup(project) + + backup_path = result.get("path") or result.get("backup_path") or "" + assert self.TIMESTAMP_RE.search(str(backup_path)), \ + f"Backup path '{backup_path}' must contain YYYYMMDD_HHMMSS timestamp" + + def test_sqlite_backup_file_is_created_on_disk(self, sqlite_project_with_db): + """AC-1+AC-2: Actual .bak file must be created in the project dir.""" + project, db_file = sqlite_project_with_db + project_path = Path(project.get("deploy_path") or project.get("path")) + + result = pre_deploy_backup(project) + + backup_path = result.get("path") or result.get("backup_path") or "" + bak = Path(backup_path) if backup_path else None + + # Either a file was created or the result says status=ok with a valid path + if bak and bak.exists(): + assert self.TIMESTAMP_RE.search(bak.name), \ + f"Backup filename '{bak.name}' must contain timestamp" + else: + # Accept any backup files in project dir with timestamp pattern + bak_files = list(project_path.glob("*.bak")) + timestamped = [f for f in bak_files if self.TIMESTAMP_RE.search(f.name)] + assert len(timestamped) > 0, \ + "No timestamped backup file found in project directory" + + def test_backup_timestamps_are_unique_on_consecutive_calls(self, sqlite_project_with_db): + """AC-2: Two rapid backups should produce different timestamp suffixes.""" + import time + project, db_file = sqlite_project_with_db + + result1 = pre_deploy_backup(project) + time.sleep(1.1) # ensure second tick + # Re-create the db file in case first call consumed it + db_file.write_bytes(b"SQLite data v2") + result2 = pre_deploy_backup(project) + + path1 = result1.get("path") or result1.get("backup_path") or "" + path2 = result2.get("path") or result2.get("backup_path") or "" + assert path1 != path2, "Consecutive backup paths must differ (unique timestamps)" + + +# --------------------------------------------------------------------------- +# AC-3: backup failure = deploy failure — additional edge cases +# --------------------------------------------------------------------------- + +class TestBackupFailureHandling: + """Edge cases where backup errors must propagate as deploy errors.""" + + def test_permission_error_propagates_as_deploy_failure(self, conn, local_project): + """PermissionError during backup → deploy fails immediately.""" + with patch("core.deploy.pre_deploy_backup", side_effect=PermissionError("read-only fs")): + result = execute_deploy(local_project, conn) + assert result["success"] is False + + def test_os_error_propagates_as_deploy_failure(self, conn, local_project): + """Generic OSError during backup → deploy fails.""" + with patch("core.deploy.pre_deploy_backup", side_effect=OSError("disk full")): + result = execute_deploy(local_project, conn) + assert result["success"] is False + + def test_backup_status_error_propagates_as_deploy_failure(self, conn, local_project): + """If backup returns status='error', deploy must not proceed.""" + with patch("core.deploy.pre_deploy_backup", return_value={"status": "error", "reason": "no space"}), \ + patch("core.deploy.subprocess.run") as mock_run: + result = execute_deploy(local_project, conn) + + assert result["success"] is False, \ + "backup status='error' must abort deploy" + mock_run.assert_not_called(), \ + "subprocess.run must not be called when backup returns error status" + + def test_project_without_sqlite_skips_backup_gracefully(self, conn): + """Project with no detectable DB file: backup skips without error.""" + # Project path has no *.db or *.sqlite files + import tempfile, os + with tempfile.TemporaryDirectory() as tmpdir: + models.create_project(conn, "no-db-svc", "No DB Service", tmpdir) + models.update_project(conn, "no-db-svc", deploy_runtime="static", deploy_path=tmpdir) + project = models.get_project(conn, "no-db-svc") + + result = pre_deploy_backup(project) + + # Acceptable outcomes: status 'skipped' or 'ok' with no file (no DB to back up) + status = result.get("status") + assert status in ("skipped", "ok", "no_db"), \ + f"Expected skipped/ok/no_db when no DB present, got '{status}'"