penti/scripts/verify_deployment.py

#!/usr/bin/env python3
"""
Deployment verification script
"""
import os
import sys
import time
import requests
from typing import Dict, List, Tuple


class DeploymentVerifier:
    """Verify deployment health and functionality"""

    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url
        self.results: List[Tuple[str, bool, str]] = []

    def verify_health(self) -> bool:
        """Verify health endpoint"""
        try:
            response = requests.get(f"{self.base_url}/health", timeout=5)
            success = response.status_code == 200
            message = "Health endpoint OK" if success else f"Status: {response.status_code}"
            self.results.append(("Health Check", success, message))
            return success
        except Exception as e:
            self.results.append(("Health Check", False, str(e)))
            return False

    def verify_strategies(self) -> bool:
        """Verify strategy list endpoint"""
        try:
            response = requests.get(
                f"{self.base_url}/api/v1/backtest/strategies/list",
                timeout=5
            )

            if response.status_code != 200:
                self.results.append((
                    "Strategy List",
                    False,
                    f"Status: {response.status_code}"
                ))
                return False

            data = response.json()
            strategies = data.get("strategies", [])

            if len(strategies) < 5:
                self.results.append((
                    "Strategy List",
                    False,
                    f"Only {len(strategies)} strategies found"
                ))
                return False

            self.results.append((
                "Strategy List",
                True,
                f"Found {len(strategies)} strategies"
            ))
            return True

        except Exception as e:
            self.results.append(("Strategy List", False, str(e)))
            return False

    def verify_database_stats(self) -> bool:
        """Verify database stats endpoint"""
        try:
            response = requests.get(
                f"{self.base_url}/api/v1/data/stats",
                timeout=5
            )

            if response.status_code != 200:
                self.results.append((
                    "Database Stats",
                    False,
                    f"Status: {response.status_code}"
                ))
                return False

            data = response.json()

            # Check if we have some data
            ticker_count = data.get("ticker_count", 0)
            price_count = data.get("price_count", 0)

            message = f"Tickers: {ticker_count}, Prices: {price_count}"

            # Warning if no data, but not a failure
            if ticker_count == 0:
                message += " (No ticker data - run data collection)"

            self.results.append(("Database Stats", True, message))
            return True

        except Exception as e:
            self.results.append(("Database Stats", False, str(e)))
            return False

    def verify_portfolio_api(self) -> bool:
        """Verify portfolio API"""
        try:
            # Test listing portfolios
            response = requests.get(
                f"{self.base_url}/api/v1/portfolios/?skip=0&limit=10",
                timeout=5
            )

            success = response.status_code == 200
            message = "Portfolio API OK" if success else f"Status: {response.status_code}"

            self.results.append(("Portfolio API", success, message))
            return success

        except Exception as e:
            self.results.append(("Portfolio API", False, str(e)))
            return False

    def verify_celery_flower(self, flower_url: str = "http://localhost:5555") -> bool:
        """Verify Celery Flower monitoring"""
        try:
            response = requests.get(flower_url, timeout=5)
            success = response.status_code == 200
            message = "Flower monitoring OK" if success else f"Status: {response.status_code}"

            self.results.append(("Celery Flower", success, message))
            return success

        except Exception as e:
            self.results.append(("Celery Flower", False, str(e)))
            return False

    def verify_frontend(self, frontend_url: str = "http://localhost:3000") -> bool:
        """Verify frontend accessibility"""
        try:
            response = requests.get(frontend_url, timeout=5)
            success = response.status_code == 200
            message = "Frontend OK" if success else f"Status: {response.status_code}"

            self.results.append(("Frontend", success, message))
            return success

        except Exception as e:
            self.results.append(("Frontend", False, str(e)))
            return False

    def print_results(self):
        """Print verification results"""
        print("\n" + "=" * 60)
        print("DEPLOYMENT VERIFICATION RESULTS")
        print("=" * 60)

        success_count = 0
        total_count = len(self.results)

        for name, success, message in self.results:
            status = "✓" if success else "✗"
            color = "\033[92m" if success else "\033[91m"
            reset = "\033[0m"

            print(f"{color}{status}{reset} {name:20s} - {message}")

            if success:
                success_count += 1

        print("=" * 60)
        print(f"Passed: {success_count}/{total_count}")

        if success_count == total_count:
            print("\033[92m✓ All checks passed!\033[0m")
            return True
        else:
            print(f"\033[91m✗ {total_count - success_count} checks failed\033[0m")
            return False

    def run_all_checks(self) -> bool:
        """Run all verification checks"""
        print("Starting deployment verification...")
        print("Waiting 5 seconds for services to be ready...")
        time.sleep(5)

        # Run all checks
        self.verify_health()
        self.verify_strategies()
        self.verify_database_stats()
        self.verify_portfolio_api()
        self.verify_celery_flower()
        self.verify_frontend()

        # Print results
        all_passed = self.print_results()

        return all_passed


def main():
    """Main entry point"""
    base_url = os.getenv("API_BASE_URL", "http://localhost:8000")

    verifier = DeploymentVerifier(base_url=base_url)
    success = verifier.run_all_checks()

    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()