Augustin's interview

.gitignore

@@ -0,0 +1,42 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Distribution / packaging
+dist/
+build/
+*.egg-info/
+
+# Virtual environments
+venv/
+env/
+.env/
+.venv/
+
+# IDE specific files
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Testing
+.pytest_cache/
+.coverage
+htmlcov/
+
+# Logs
+*.log
+
+# Local development settings
+.env
+.env.local
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# mypy
+.mypy_cache/
+
+# pytest
+.pytest_cache/

README.md

@@ -21,3 +21,14 @@ You must:
 
 * https://docs.datadoghq.com/api/latest/downtimes/ 
 * https://cd.apps.argoproj.io/swagger-ui 
+
+## Potential improvements
+
+* Add a logging system instead of click.echo, improve error logging
+* Add unit tests
+* Use real API calls (ArgoCD REST API, DD Python SDK)
+* Check availability of the dependencies (Datadog, ArgoCD) before running the CLI and handle errors gracefully
+* Add a timeout for each step to avoid long running requests
+* Create a github action workflow to automate the CLI run
+* Instrument the CLI with telemetry to audit the runs
+* Add a configuration system to allow the CLI to be configured with environment variables

pyproject.toml

@@ -0,0 +1,19 @@
+[tool.poetry]
+name = "paas-coding-interview"
+version = "0.1.0"
+description = ""
+authors = ["alafanechere <augustin.lafanechere@gmail.com>"]
+readme = "README.md"
+packages = [{ include = "paas_coding_interview", from = "src" }]
+
+[tool.poetry.dependencies]
+python = "^3.13"
+click = "^8.1.8"
+
+[tool.poetry.scripts]
+paas = "paas_coding_interview.cli:paas"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

src/paas_coding_interview/argo_mocks.py

@@ -0,0 +1,11 @@
+
+
+def sync(application_name: str) -> tuple[bool, str]:
+    """
+    Mock Argo CD sync operation
+    Returns ArgoSyncStatus with success/failure and message
+    """
+    if application_name == "failing-app":
+        return False, "Failed to sync application, but that's not a real thing, it's just a mock"
+
+    return True, "Successfully synced application, but that's not a real thing, it's just a mock"

src/paas_coding_interview/cli.py

@@ -0,0 +1,38 @@
+import click
+from paas_coding_interview import datadog_mocks, argo_mocks, errors
+from paas_coding_interview.registries import registry
+
+
+@click.group()
+def paas():
+    pass
+
+@paas.command()
+@click.argument('application_name')
+def deploy(application_name: str):
+    # Validate application exists in registry
+    application = registry.get_application(application_name)
+
+    # We don't want to deploy if the application has an existing downtime
+    if application.has_datadog_downtime():
+        raise errors.ApplicationIsDownError(application)
+
+    # We can schedule a downtime for the application has it does not have an existing downtime
+    application.schedule_datadog_downtime()
+    click.echo("Datadog downtime scheduled")
+
+    # Sync the application with Argo CD - aka deploy the application
+    try:
+        sync_status = application.argo_sync()
+        if not sync_status.success:
+            raise errors.ApplicationSyncError(application)
+    finally:
+        # Unset the downtime in any case
+        # We assume Argo CD is self-healing
+        application.unset_datadog_downtime()
+        click.echo("Datadog downtime unset")
+
+    click.echo(sync_status.message)
+
+if __name__ == '__main__':
+    paas()

src/paas_coding_interview/datadog_mocks.py

@@ -0,0 +1,42 @@
+# Mock data to simulate Datadog API responses
+# We assume that Datadog API can give us downtime IDs from a specific application name
+
+from paas_coding_interview.models import Downtime
+
+# Simulate a set of active downtimes that would be DD source of truth
+_active_downtimes = set()
+
+
+def get_downtime(application_name: str) -> Downtime:
+    """
+    Mock checking if an application has an active downtime
+    Returns a Downtime object with active status and optional ID
+    """
+    # Simulate an application that's already in downtime
+    if application_name == "app-in-downtime":
+        return Downtime(active=True, downtime_id="mock-downtime-123")
+
+    # Default case - no active downtime
+    return Downtime(active=False)
+
+def has_downtime(application_name: str) -> bool:
+    """
+    Check if application has an active downtime
+    Returns True if downtime exists, False otherwise
+    """
+    return get_downtime(application_name).active
+
+def schedule_downtime(application_name: str) -> Downtime:
+    """
+    Schedule a downtime for an application
+    Returns a Downtime object with the new downtime details
+    """
+    _active_downtimes.add(application_name)
+    return Downtime(active=True, downtime_id="mock-downtime-456")
+
+def unset_downtime(downtime_id: str) -> Downtime:
+    """
+    Remove a downtime using its ID
+    Returns a Downtime object with the updated status
+    """
+    return Downtime(active=False)

src/paas_coding_interview/errors.py

@@ -0,0 +1,50 @@
+import os
+from functools import wraps
+from typing import Type, Callable
+from paas_coding_interview.notification_mocks import send_slack_message
+
+def notify_error_in_ci(error_channel: str = "#deployments-errors") -> Callable:
+    """
+    Decorator factory that sends notifications to Slack when in CI environment
+    before re-raising the original error
+    """
+    def decorator(error_class: Type[Exception]) -> Type[Exception]:
+        if not os.getenv("CI"):
+            return error_class
+
+        class NotifiedError(error_class):
+            def __init__(self, *args, **kwargs):
+                super().__init__(*args, **kwargs)
+                send_slack_message(
+                    channel=error_channel,
+                    message=f"Error in CI: {str(self)}"
+                )
+
+        NotifiedError.__name__ = error_class.__name__
+        return NotifiedError
+
+    return decorator
+
+# Example usage:
+@notify_error_in_ci()
+class ApplicationNotFoundError(Exception):
+    """Raised when application is not found in registry"""
+    def __init__(self, application_name: str = None):
+        message = f"Application '{application_name}' not found in registry" if application_name else "Application not found in registry"
+        super().__init__(message)
+
+@notify_error_in_ci()
+class ApplicationIsDownError(Exception):
+    """Raised when trying to deploy an application that is down"""
+    def __init__(self, application):
+        message = f"Application '{application.name}' is currently down. Could not deploy."
+        if application.downtime and application.downtime.downtime_id:
+            message += f" (downtime_id: {application.downtime.downtime_id})"
+        super().__init__(message)
+
+@notify_error_in_ci()
+class ApplicationSyncError(Exception):
+    """Raised when trying to sync an application that failed"""
+    def __init__(self, application):
+        message = f"Application '{application.name}' failed to sync to Argo CD. Could not deploy."
+        super().__init__(message)

src/paas_coding_interview/models.py

@@ -0,0 +1,39 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from paas_coding_interview import datadog_mocks
+from paas_coding_interview import argo_mocks
+
+
+@dataclass
+class ArgoSyncStatus:
+    success: bool
+    message: str
+
+@dataclass
+class Downtime:
+    active: bool
+    downtime_id: Optional[str] = None
+
+@dataclass
+class Application:
+    name: str
+    namespace: str
+    project: str
+    sync_status: Optional[str] = None
+    health_status: Optional[str] = None
+    downtime: Optional[Downtime] = None
+
+    def has_datadog_downtime(self) -> bool:
+        return datadog_mocks.has_downtime(self.name)
+
+    def schedule_datadog_downtime(self) -> Downtime:
+        return datadog_mocks.schedule_downtime(self.name)
+
+    def unset_datadog_downtime(self) -> None:
+        if not self.downtime:
+            raise ValueError("Downtime ID is not set")
+        return datadog_mocks.unset_downtime(self.downtime)
+
+    def argo_sync(self) -> ArgoSyncStatus:
+        return ArgoSyncStatus(*argo_mocks.sync(self.name))

src/paas_coding_interview/notification_mocks.py

@@ -0,0 +1,11 @@
+def send_slack_message(channel: str, message: str) -> bool:
+    """
+    Mock sending a Slack message to a channel
+    Returns True if message was sent successfully, False otherwise
+
+    For testing: Messages to 'invalid-channel' will fail
+    """
+    if channel == "invalid-channel":
+        return False
+    return True
+

src/paas_coding_interview/registries.py

@@ -0,0 +1,70 @@
+from dataclasses import dataclass
+import json
+from typing import List
+from paas_coding_interview.models import Application
+from paas_coding_interview.datadog_mocks import get_downtime
+
+class ApplicationNotFoundError(Exception):
+    """Raised when application is not found in registry"""
+    pass
+
+class ApplicationRegistry:
+    def __init__(self, applications: List[Application]):
+        self.applications = applications
+
+    @classmethod
+    def from_argo(cls, repo_path: str):
+        # Mock response from /api/v1/applications
+        mock_argo_response = {
+            "items": [
+                {
+                    "metadata": {
+                        "name": "backend",
+                        "namespace": "production",
+                    },
+                    "spec": {
+                        "project": "main"
+                    },
+                    "status": {
+                        "sync": {"status": "Synced"},
+                        "health": {"status": "Healthy"}
+                    }
+                },
+                {
+                    "metadata": {
+                        "name": "failing-app",
+                        "namespace": "production",
+                    },
+                    "spec": {
+                        "project": "main"
+                    },
+                    "status": {
+                        "sync": {"status": "OutOfSync"},
+                        "health": {"status": "Degraded"}
+                    }
+                }
+            ]
+        }
+
+        applications = []
+        for app_data in mock_argo_response["items"]:
+            app = Application(
+                name=app_data["metadata"]["name"],
+                namespace=app_data["metadata"]["namespace"],
+                project=app_data["spec"]["project"],
+                sync_status=app_data["status"]["sync"]["status"],
+                health_status=app_data["status"]["health"]["status"],
+                downtime=get_downtime(app_data["metadata"]["name"])
+            )
+            applications.append(app)
+
+        return cls(applications)
+
+    def get_application(self, application_name: str) -> Application:
+        for app in self.applications:
+            if app.name == application_name:
+                return app
+        raise ApplicationNotFoundError(f"Application '{application_name}' not found in registry")
+
+# Create a singleton instance of the registry
+registry = ApplicationRegistry.from_argo("mock/path")