Added observability tests.

Author: pantierra

.github/workflows/ci.yml

@@ -114,6 +114,16 @@ jobs:
           echo "=== Post-deployment validation ==="
           ./scripts/test.sh check-deployment
 
+      - name: Wait for monitoring stack
+        run: |
+          echo "=== Waiting for monitoring components (required for autoscaling) ==="
+          kubectl wait --for=condition=Ready pod -l app.kubernetes.io/component=server,app.kubernetes.io/name=prometheus -n eoapi --timeout=120s &
+          kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=grafana -n eoapi --timeout=120s &
+          kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=prometheus-adapter -n eoapi --timeout=120s &
+          wait # Wait for all background jobs
+          echo "✅ Monitoring stack ready"
+          kubectl get hpa -n eoapi
+
       - name: Run integration tests
         run: |
           export RELEASE_NAME="$RELEASE_NAME"

.github/workflows/tests/conftest.py

@@ -1,9 +1,13 @@
+import json
 import os
-from typing import Any, Generator
+import subprocess
+import time
+from typing import Any, Dict, Generator, List, Optional, cast
 
 import psycopg2
 import psycopg2.extensions
 import pytest
+import requests
 
 
 @pytest.fixture(scope="session")
@@ -22,8 +26,7 @@ def stac_endpoint() -> str:
 
 
 @pytest.fixture(scope="session")
-def db_connection() -> Generator[Any, None, None]:
-    """Create database connection for testing."""
+def db_connection() -> Generator[psycopg2.extensions.connection, None, None]:
     # Require all database connection parameters to be explicitly set
     required_vars = ["PGHOST", "PGPORT", "PGDATABASE", "PGUSER", "PGPASSWORD"]
     missing_vars = [var for var in required_vars if not os.getenv(var)]
@@ -33,6 +36,18 @@ def db_connection() -> Generator[Any, None, None]:
             f"Required environment variables not set: {', '.join(missing_vars)}"
         )
 
+    port_str = os.getenv("PGPORT")
+    if not port_str:
+        pytest.fail("PGPORT environment variable not set")
+
+    connection_params = {
+        "host": os.getenv("PGHOST"),
+        "port": int(port_str),
+        "database": os.getenv("PGDATABASE"),
+        "user": os.getenv("PGUSER"),
+        "password": os.getenv("PGPASSWORD"),
+    }
+
     # All required vars are guaranteed to exist due to check above
     try:
         conn = psycopg2.connect(
@@ -47,3 +62,186 @@ def db_connection() -> Generator[Any, None, None]:
         conn.close()
     except psycopg2.Error as e:
         pytest.fail(f"Cannot connect to database: {e}")
+
+
+def get_namespace() -> str:
+    return os.environ.get("NAMESPACE", "eoapi")
+
+
+def get_release_name() -> str:
+    return os.environ.get("RELEASE_NAME", "eoapi")
+
+
+def kubectl_get(
+    resource: str,
+    namespace: Optional[str] = None,
+    label_selector: Optional[str] = None,
+    output: str = "json",
+) -> subprocess.CompletedProcess[str]:
+    cmd: List[str] = ["kubectl", "get", resource]
+
+    if namespace:
+        cmd.extend(["-n", namespace])
+
+    if label_selector:
+        cmd.extend(["-l", label_selector])
+
+    if output:
+        cmd.extend(["-o", output])
+
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    return result
+
+
+def kubectl_port_forward(
+    service: str, local_port: int, remote_port: int, namespace: str
+) -> subprocess.Popen[str]:
+    cmd = [
+        "kubectl",
+        "port-forward",
+        f"svc/{service}",
+        f"{local_port}:{remote_port}",
+        "-n",
+        namespace,
+    ]
+
+    process = subprocess.Popen(
+        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
+    )
+
+    time.sleep(3)
+    return process
+
+
+def wait_for_url(url: str, timeout: int = 30, interval: int = 2) -> bool:
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        try:
+            response = requests.get(url, timeout=5)
+            if response.status_code == 200:
+                return True
+        except (requests.RequestException, requests.ConnectionError):
+            pass
+        time.sleep(interval)
+    return False
+
+
+def make_request(url: str, timeout: int = 10) -> bool:
+    try:
+        response = requests.get(url, timeout=timeout)
+        return response.status_code == 200
+    except requests.RequestException:
+        return False
+
+
+def get_base_url() -> str:
+    namespace = get_namespace()
+
+    # Check if we have an ingress
+    result = subprocess.run(
+        ["kubectl", "get", "ingress", "-n", namespace, "-o", "json"],
+        capture_output=True,
+        text=True,
+    )
+
+    if result.returncode == 0:
+        ingress_data = json.loads(result.stdout)
+        if ingress_data["items"]:
+            ingress = ingress_data["items"][0]
+            rules = ingress.get("spec", {}).get("rules", [])
+            if rules:
+                host = rules[0].get("host", "localhost")
+                # Check if host is accessible
+                try:
+                    response = requests.get(
+                        f"http://{host}/stac/collections", timeout=5
+                    )
+                    if response.status_code == 200:
+                        return f"http://{host}"
+                except requests.RequestException:
+                    pass
+
+    return "http://localhost:8080"
+
+
+def get_pod_metrics(namespace: str, service_name: str) -> List[Dict[str, str]]:
+    """Get CPU and memory metrics for pods of a specific service."""
+    release_name = get_release_name()
+    result = subprocess.run(
+        [
+            "kubectl",
+            "top",
+            "pods",
+            "-n",
+            namespace,
+            "-l",
+            f"app={release_name}-{service_name}",
+            "--no-headers",
+        ],
+        capture_output=True,
+        text=True,
+    )
+
+    if result.returncode != 0:
+        return []
+
+    metrics: List[Dict[str, str]] = []
+    for line in result.stdout.strip().split("\n"):
+        if line.strip():
+            parts = line.split()
+            if len(parts) >= 3:
+                pod_name = parts[0]
+                cpu = parts[1]  # e.g., "25m"
+                memory = parts[2]  # e.g., "128Mi"
+                metrics.append({"pod": pod_name, "cpu": cpu, "memory": memory})
+
+    return metrics
+
+
+def get_hpa_status(namespace: str, hpa_name: str) -> Optional[Dict[str, Any]]:
+    """Get HPA status for a specific HPA."""
+    result = kubectl_get("hpa", namespace=namespace, output="json")
+    if result.returncode != 0:
+        return None
+
+    hpas = json.loads(result.stdout)
+    for hpa in hpas["items"]:
+        if hpa["metadata"]["name"] == hpa_name:
+            return cast(Dict[str, Any], hpa)
+
+    return None
+
+
+def get_pod_count(namespace: str, service_name: str) -> int:
+    """Get the count of running pods for a specific service."""
+    release_name = get_release_name()
+    result = kubectl_get(
+        "pods",
+        namespace=namespace,
+        label_selector=f"app={release_name}-{service_name}",
+    )
+
+    if result.returncode != 0:
+        return 0
+
+    pods = json.loads(result.stdout)
+    running_pods = [
+        pod for pod in pods["items"] if pod["status"]["phase"] == "Running"
+    ]
+
+    return len(running_pods)
+
+
+@pytest.fixture(scope="session")
+def namespace() -> str:
+    return get_namespace()
+
+
+@pytest.fixture(scope="session")
+def release_name() -> str:
+    return get_release_name()
+
+
+@pytest.fixture(scope="session")
+def base_url() -> str:
+    return get_base_url()