221 lines
9.8 KiB
Python
221 lines
9.8 KiB
Python
|
"""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}'"
|