How to Build a GitOps Argo CD Sync Mock in Python

import random
import time
from dataclasses import dataclass, field
from typing import List, Dict


@dataclass
class Application:
    name: str
    source_repo: str
    target_revision: str
    synced: bool = False
    health_status: str = "Healthy"
    history: List[Dict] = field(default_factory=list)

    def sync(self) -> None:
        """Simulate Argo CD sync operation."""
        print(f"Syncing {self.name}...")
        time.sleep(0.5)  # Simulate network/delay

        success = random.random() > 0.2  # 80% success rate
        self.synced = success

        if success:
            self.health_status = "Healthy"
            print(f"  ✓ {self.name} synced to {self.target_revision}")
        else:
            self.health_status = "Degraded"
            print(f"  ✗ {self.name} sync failed")

        self.history.append({
            "revision": self.target_revision,
            "success": success,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
        })


def force_sync(apps: List[Application]) -> None:
    """Simulate Argo CD force sync (bypasses hooks)."""
    force_sync_apps = [app for app in apps if not app.synced]
    if not force_sync_apps:
        print("All applications are already in sync.")
        return

    print(f"Force syncing {len(force_sync_apps)} out-of-sync app(s):")
    for app in force_sync_apps:
        app.sync()
        if app.synced:
            print(f"  → {app.name} forced to healthy state")


def main() -> None:
    # Mock GitOps applications
    applications = [
        Application("payment-api", "https://github.com/company/payment-api.git", "v1.4.0"),
        Application("inventory-service", "https://github.com/company/inventory-service.git", "v2.1.0"),
        Application("auth-gateway", "https://github.com/company/auth-gateway.git", "v0.9.1"),
        Application("notification-worker", "https://github.com/company/notification-worker.git", "v1.0.2"),
    ]

    print("=== Argo CD Sync Mock ===")
    print(f"Initial status: {len([a for a in applications if a.synced])}/{len(applications)} synced\n")

    # Simulate sync process
    for app in applications:
        app.sync()
        print()

    print("=== Post-sync status ===")
    for app in applications:
        state = "✓" if app.synced else "✗"
        print(f"{state} {app.name:20} health={app.health_status:8} rev={app.target_revision}")

    # Check for failed syncs
    failed = [app for app in applications if not app.synced]
    if failed:
        print(f"\n{len(failed)} application(s) out of sync. Attempting force sync...")
        force_sync(failed)


if __name__ == "__main__":
    main()