Phase 5.13: CI/CD deploy script with build/test/deploy/rollback

scripts/ai_deploy.py

@@ -0,0 +1,442 @@
+#!/usr/bin/env python3
+"""AI Deploy Script — Build, Test, Deploy with automatic rollback.
+
+Usage:
+    python scripts/ai_deploy.py
+    python scripts/ai_deploy.py --skip-tests
+    python scripts/ai_deploy.py --dry-run
+    python scripts/ai_deploy.py --skip-build --skip-tests
+
+Pipeline phases:
+    1. Build   — docker build (or npm run build for frontend-only)
+    2. Tests   — pytest (backend) + vitest (frontend)
+    3. Deploy  — Coolify API call to redeploy
+    4. Rollback — on failure, restore previous deployment
+
+Environment variables:
+    COOLIFY_API_TOKEN  — Coolify API token for deployment
+    COOLIFY_APP_UUID   — Coolify application UUID
+    COOLIFY_BASE_URL   — Coolify base URL (default: https://server.media-on.de)
+
+Exit codes:
+    0 — deployment successful
+    1 — deployment failed (rollback attempted)
+    2 — configuration error
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+# Add project root to path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import httpx
+
+
+# ─── Data Structures ─────────────────────────────────────────────────
+
+@dataclass
+class PhaseResult:
+    """Result of a single pipeline phase."""
+    name: str
+    success: bool = False
+    duration_s: float = 0.0
+    output: str = ""
+    error: str = ""
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "phase": self.name,
+            "success": self.success,
+            "duration_s": round(self.duration_s, 2),
+            "output": self.output[:500] if self.output else "",
+            "error": self.error[:500] if self.error else "",
+            "metadata": self.metadata,
+        }
+
+
+@dataclass
+class DeployReport:
+    """Full deployment report."""
+    started_at: str = ""
+    finished_at: str = ""
+    phases: list[PhaseResult] = field(default_factory=list)
+    overall_success: bool = False
+    rollback_performed: bool = False
+    rollback_success: bool = False
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "started_at": self.started_at,
+            "finished_at": self.finished_at,
+            "overall_success": self.overall_success,
+            "rollback_performed": self.rollback_performed,
+            "rollback_success": self.rollback_success,
+            "phases": [p.to_dict() for p in self.phases],
+        }
+
+
+# ─── Phase Implementations ───────────────────────────────────────────
+
+def run_command(cmd: list[str], cwd: str | None = None, timeout: int = 600) -> tuple[bool, str, str]:
+    """Run a shell command and return (success, stdout, stderr)."""
+    try:
+        result = subprocess.run(
+            cmd,
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+        return result.returncode == 0, result.stdout, result.stderr
+    except subprocess.TimeoutExpired:
+        return False, "", f"Command timed out after {timeout}s"
+    except FileNotFoundError as exc:
+        return False, "", str(exc)
+    except Exception as exc:
+        return False, "", str(exc)
+
+
+def phase_build(skip_build: bool, dry_run: bool) -> PhaseResult:
+    """Build phase: docker build or npm run build."""
+    start = time.perf_counter()
+    result = PhaseResult(name="build")
+
+    if skip_build:
+        result.success = True
+        result.output = "Build skipped (--skip-build)"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+    if dry_run:
+        result.success = True
+        result.output = "[DRY-RUN] Would run: docker build -t leocrm:latest ."
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    # Try docker build first
+    docker_available = shutil.which("docker") is not None
+    if docker_available:
+        success, stdout, stderr = run_command(
+            ["docker", "build", "-t", "leocrm:latest", "."],
+            cwd=project_root,
+            timeout=900,
+        )
+        result.success = success
+        result.output = stdout
+        result.error = stderr
+        result.metadata["build_tool"] = "docker"
+    else:
+        # Fallback: frontend-only build with npm
+        frontend_dir = os.path.join(project_root, "frontend")
+        npm_available = shutil.which("npm") is not None
+        if npm_available and os.path.isdir(frontend_dir):
+            success, stdout, stderr = run_command(
+                ["npm", "run", "build"],
+                cwd=frontend_dir,
+                timeout=300,
+            )
+            result.success = success
+            result.output = stdout
+            result.error = stderr
+            result.metadata["build_tool"] = "npm"
+        else:
+            result.success = False
+            result.error = "Neither docker nor npm available for build"
+
+    result.duration_s = time.perf_counter() - start
+    return result
+
+
+def phase_tests(skip_tests: bool, dry_run: bool) -> PhaseResult:
+    """Test phase: pytest (backend) + vitest (frontend)."""
+    start = time.perf_counter()
+    result = PhaseResult(name="tests")
+
+    if skip_tests:
+        result.success = True
+        result.output = "Tests skipped (--skip-tests)"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    if dry_run:
+        result.success = True
+        result.output = "[DRY-RUN] Would run: pytest + vitest"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    python_bin = os.environ.get("LEOCRM_PYTHON", "/opt/venv/bin/python")
+    test_outputs: list[str] = []
+    test_errors: list[str] = []
+    all_success = True
+
+    # Backend tests: pytest
+    if os.path.isfile(os.path.join(project_root, "pyproject.toml")):
+        success, stdout, stderr = run_command(
+            [python_bin, "-m", "pytest", "tests/", "-x", "--tb=short", "-q"],
+            cwd=project_root,
+            timeout=300,
+        )
+        test_outputs.append(f"=== pytest ===\n{stdout}")
+        if stderr:
+            test_errors.append(f"=== pytest stderr ===\n{stderr}")
+        if not success:
+            all_success = False
+            result.metadata["pytest_passed"] = False
+        else:
+            result.metadata["pytest_passed"] = True
+
+    # Frontend tests: vitest
+    frontend_dir = os.path.join(project_root, "frontend")
+    npm_available = shutil.which("npm") is not None
+    if npm_available and os.path.isdir(frontend_dir):
+        success, stdout, stderr = run_command(
+            ["npm", "run", "test"],
+            cwd=frontend_dir,
+            timeout=300,
+        )
+        test_outputs.append(f"=== vitest ===\n{stdout}")
+        if stderr:
+            test_errors.append(f"=== vitest stderr ===\n{stderr}")
+        if not success:
+            all_success = False
+            result.metadata["vitest_passed"] = False
+        else:
+            result.metadata["vitest_passed"] = True
+
+    result.success = all_success
+    result.output = "\n".join(test_outputs)
+    result.error = "\n".join(test_errors)
+    result.duration_s = time.perf_counter() - start
+    return result
+
+
+def phase_deploy(dry_run: bool) -> PhaseResult:
+    """Deploy phase: Coolify API call to redeploy."""
+    start = time.perf_counter()
+    result = PhaseResult(name="deploy")
+
+    api_token = os.environ.get("COOLIFY_API_TOKEN", "")
+    app_uuid = os.environ.get("COOLIFY_APP_UUID", "")
+    base_url = os.environ.get("COOLIFY_BASE_URL", "https://server.media-on.de")
+
+    if not api_token or not app_uuid:
+        result.success = False
+        result.error = "Missing COOLIFY_API_TOKEN or COOLIFY_APP_UUID environment variables"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    if dry_run:
+        result.success = True
+        result.output = f"[DRY-RUN] Would call: POST {base_url}/api/v1/applications/{app_uuid}/deploy"
+        result.metadata["api_url"] = f"{base_url}/api/v1/applications/{app_uuid}/deploy"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    deploy_url = f"{base_url}/api/v1/applications/{app_uuid}/deploy"
+    headers = {"Authorization": f"Bearer {api_token}"}
+
+    try:
+        response = httpx.post(deploy_url, headers=headers, timeout=120.0)
+        if response.status_code < 400:
+            result.success = True
+            result.output = f"Deploy triggered: HTTP {response.status_code}"
+            try:
+                result.metadata["response"] = response.json()
+            except Exception:
+                result.metadata["response_text"] = response.text[:200]
+        else:
+            result.success = False
+            result.error = f"Deploy failed: HTTP {response.status_code} - {response.text[:300]}"
+    except httpx.TimeoutException:
+        result.success = False
+        result.error = "Deploy request timed out"
+    except httpx.ConnectError as exc:
+        result.success = False
+        result.error = f"Connection error: {exc}"
+    except Exception as exc:
+        result.success = False
+        result.error = str(exc)
+
+    result.duration_s = time.perf_counter() - start
+    return result
+
+
+def phase_rollback(dry_run: bool) -> PhaseResult:
+    """Rollback phase: restore previous deployment via Coolify API."""
+    start = time.perf_counter()
+    result = PhaseResult(name="rollback")
+
+    api_token = os.environ.get("COOLIFY_API_TOKEN", "")
+    app_uuid = os.environ.get("COOLIFY_APP_UUID", "")
+    base_url = os.environ.get("COOLIFY_BASE_URL", "https://server.media-on.de")
+
+    if not api_token or not app_uuid:
+        result.success = False
+        result.error = "Missing COOLIFY_API_TOKEN or COOLIFY_APP_UUID for rollback"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    if dry_run:
+        result.success = True
+        result.output = f"[DRY-RUN] Would call: POST {base_url}/api/v1/applications/{app_uuid}/rollback"
+        result.duration_s = time.perf_counter() - start
+        return result
+
+    # Coolify rollback: restart with previous image
+    rollback_url = f"{base_url}/api/v1/applications/{app_uuid}/restart"
+    headers = {"Authorization": f"Bearer {api_token}"}
+
+    try:
+        response = httpx.post(rollback_url, headers=headers, timeout=120.0)
+        if response.status_code < 400:
+            result.success = True
+            result.output = f"Rollback triggered: HTTP {response.status_code}"
+        else:
+            result.success = False
+            result.error = f"Rollback failed: HTTP {response.status_code} - {response.text[:300]}"
+    except Exception as exc:
+        result.success = False
+        result.error = str(exc)
+
+    result.duration_s = time.perf_counter() - start
+    return result
+
+
+# ─── Pipeline Orchestration ──────────────────────────────────────────
+
+def run_pipeline(skip_build: bool, skip_tests: bool, dry_run: bool) -> DeployReport:
+    """Run the full deployment pipeline."""
+    from datetime import datetime, timezone
+
+    report = DeployReport()
+    report.started_at = datetime.now(timezone.utc).isoformat()
+
+    print(f"\n{'='*60}")
+    print(f"  LeoCRM AI Deploy Pipeline")
+    print(f"  {'[DRY-RUN]' if dry_run else '[LIVE]'}")
+    print(f"{'='*60}\n")
+
+    # Phase 1: Build
+    print("▶ Phase 1: Build")
+    build_result = phase_build(skip_build, dry_run)
+    report.phases.append(build_result)
+    _print_phase_result(build_result)
+
+    if not build_result.success:
+        print("\n✗ Build failed — aborting pipeline.")
+        report.overall_success = False
+        report.finished_at = datetime.now(timezone.utc).isoformat()
+        return report
+
+    # Phase 2: Tests
+    print("\n▶ Phase 2: Tests")
+    test_result = phase_tests(skip_tests, dry_run)
+    report.phases.append(test_result)
+    _print_phase_result(test_result)
+
+    if not test_result.success:
+        print("\n✗ Tests failed — aborting pipeline (no deploy).")
+        report.overall_success = False
+        report.finished_at = datetime.now(timezone.utc).isoformat()
+        return report
+
+    # Phase 3: Deploy
+    print("\n▶ Phase 3: Deploy")
+    deploy_result = phase_deploy(dry_run)
+    report.phases.append(deploy_result)
+    _print_phase_result(deploy_result)
+
+    if not deploy_result.success:
+        # Phase 4: Rollback
+        print("\n✗ Deploy failed — initiating rollback...")
+        rollback_result = phase_rollback(dry_run)
+        report.phases.append(rollback_result)
+        report.rollback_performed = True
+        report.rollback_success = rollback_result.success
+        _print_phase_result(rollback_result)
+        report.overall_success = False
+    else:
+        report.overall_success = True
+        print("\n✓ Deployment successful!")
+
+    report.finished_at = datetime.now(timezone.utc).isoformat()
+    return report
+
+
+def _print_phase_result(result: PhaseResult) -> None:
+    """Print a phase result summary."""
+    status = "✓" if result.success else "✗"
+    print(f"  {status} {result.name}: {'PASS' if result.success else 'FAIL'} ({result.duration_s:.1f}s)")
+    if result.output:
+        for line in result.output.strip().split("\n")[-5:]:
+            print(f"    │ {line}")
+    if result.error:
+        for line in result.error.strip().split("\n")[-3:]:
+            print(f"    │ ERROR: {line}", file=sys.stderr)
+
+
+# ─── CLI Entry Point ─────────────────────────────────────────────────
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="LeoCRM AI Deploy — Build, Test, Deploy with automatic rollback."
+    )
+    parser.add_argument(
+        "--skip-tests",
+        action="store_true",
+        help="Skip the test phase",
+    )
+    parser.add_argument(
+        "--skip-build",
+        action="store_true",
+        help="Skip the build phase",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Simulate the pipeline without making changes",
+    )
+    parser.add_argument(
+        "--output",
+        default="-",
+        help="Write structured JSON report to file (default: stdout)",
+    )
+    args = parser.parse_args()
+
+    report = run_pipeline(
+        skip_build=args.skip_build,
+        skip_tests=args.skip_tests,
+        dry_run=args.dry_run,
+    )
+
+    # Output structured report
+    report_json = json.dumps(report.to_dict(), indent=2, ensure_ascii=False)
+    if args.output == "-":
+        print(f"\n{'='*60}")
+        print("  Deployment Report")
+        print(f"{'='*60}")
+        print(report_json)
+    else:
+        with open(args.output, "w", encoding="utf-8") as f:
+            f.write(report_json)
+        print(f"\nReport written to {args.output}", file=sys.stderr)
+
+    return 0 if report.overall_success else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_ai_deploy.py

@@ -0,0 +1,132 @@
+"""Tests for the AI deploy script (scripts/ai_deploy.py)."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from unittest.mock import patch
+
+import pytest
+
+# Add scripts dir to path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
+
+
+def test_phase_result_to_dict():
+    """PhaseResult.to_dict should serialize correctly."""
+    from ai_deploy import PhaseResult
+
+    result = PhaseResult(name="build", success=True, duration_s=1.5, output="ok")
+    d = result.to_dict()
+    assert d["phase"] == "build"
+    assert d["success"] is True
+    assert d["duration_s"] == 1.5
+    assert d["output"] == "ok"
+
+
+def test_deploy_report_to_dict():
+    """DeployReport.to_dict should include phases and rollback info."""
+    from ai_deploy import DeployReport, PhaseResult
+
+    report = DeployReport(
+        started_at="2026-01-01T00:00:00Z",
+        finished_at="2026-01-01T00:05:00Z",
+        overall_success=True,
+    )
+    report.phases.append(PhaseResult(name="build", success=True))
+    d = report.to_dict()
+    assert d["overall_success"] is True
+    assert len(d["phases"]) == 1
+    assert d["rollback_performed"] is False
+
+
+def test_phase_build_skip():
+    """phase_build should skip when --skip-build is set."""
+    from ai_deploy import phase_build
+
+    result = phase_build(skip_build=True, dry_run=False)
+    assert result.success is True
+    assert "skipped" in result.output.lower()
+
+
+def test_phase_build_dry_run():
+    """phase_build should return dry-run message in dry-run mode."""
+    from ai_deploy import phase_build
+
+    result = phase_build(skip_build=False, dry_run=True)
+    assert result.success is True
+    assert "DRY-RUN" in result.output
+
+
+def test_phase_tests_skip():
+    """phase_tests should skip when --skip-tests is set."""
+    from ai_deploy import phase_tests
+
+    result = phase_tests(skip_tests=True, dry_run=False)
+    assert result.success is True
+    assert "skipped" in result.output.lower()
+
+
+def test_phase_tests_dry_run():
+    """phase_tests should return dry-run message in dry-run mode."""
+    from ai_deploy import phase_tests
+
+    result = phase_tests(skip_tests=False, dry_run=True)
+    assert result.success is True
+    assert "DRY-RUN" in result.output
+
+
+def test_phase_deploy_missing_env():
+    """phase_deploy should fail when env vars are missing."""
+    from ai_deploy import phase_deploy
+
+    # Ensure env vars are not set
+    with patch.dict(os.environ, {}, clear=True):
+        result = phase_deploy(dry_run=False)
+    assert result.success is False
+    assert "COOLIFY" in result.error
+
+
+def test_phase_deploy_dry_run_with_env():
+    """phase_deploy should succeed in dry-run mode with env vars set."""
+    from ai_deploy import phase_deploy
+
+    with patch.dict(os.environ, {"COOLIFY_API_TOKEN": "test", "COOLIFY_APP_UUID": "uuid-123"}):
+        result = phase_deploy(dry_run=True)
+    assert result.success is True
+    assert "DRY-RUN" in result.output
+    assert "uuid-123" in result.output
+
+
+def test_phase_rollback_missing_env():
+    """phase_rollback should fail when env vars are missing."""
+    from ai_deploy import phase_rollback
+
+    with patch.dict(os.environ, {}, clear=True):
+        result = phase_rollback(dry_run=False)
+    assert result.success is False
+    assert "COOLIFY" in result.error
+
+
+def test_run_pipeline_dry_run_all_skip():
+    """Full pipeline in dry-run with all skips should succeed with env vars."""
+    from ai_deploy import run_pipeline
+
+    with patch.dict(os.environ, {"COOLIFY_API_TOKEN": "test", "COOLIFY_APP_UUID": "uuid-123"}):
+        report = run_pipeline(skip_build=True, skip_tests=True, dry_run=True)
+    assert report.overall_success is True
+    assert len(report.phases) == 3  # build, tests, deploy
+    assert all(p.success for p in report.phases)
+    assert report.rollback_performed is False
+
+
+def test_run_pipeline_deploy_fail_triggers_rollback():
+    """Pipeline should trigger rollback when deploy fails."""
+    from ai_deploy import run_pipeline
+
+    with patch.dict(os.environ, {}, clear=True):
+        report = run_pipeline(skip_build=True, skip_tests=True, dry_run=False)
+    assert report.overall_success is False
+    assert report.rollback_performed is True
+    assert len(report.phases) == 4  # build, tests, deploy, rollback