diff --git a/scripts/ai_deploy.py b/scripts/ai_deploy.py new file mode 100644 index 0000000..111bb11 --- /dev/null +++ b/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()) diff --git a/tests/test_ai_deploy.py b/tests/test_ai_deploy.py new file mode 100644 index 0000000..bfaa5ab --- /dev/null +++ b/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