tracer-cloud--opensre/surfaces/cli/tests/discover.py

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import cast

import yaml
from typing_extensions import TypedDict

from config.constants.paths import REPO_ROOT, SYNTHETIC_SCENARIOS_DIR
from surfaces.cli.tests.catalog import TestCatalog, TestCatalogItem, TestRequirement

MAKEFILE_PATH = REPO_ROOT / "Makefile"
RCA_DIR = REPO_ROOT / "tests" / "e2e" / "rca"
OPENCLAW_SYNTHETIC_SCENARIOS_DIR = REPO_ROOT / "tests" / "synthetic" / "openclaw" / "scenarios"
CLOUDOPSBENCH_DIR = REPO_ROOT / "tests" / "benchmarks" / "cloudopsbench"

__all__ = (
    "MAKEFILE_PATH",
    "discover_cli_commands",
    "discover_make_targets",
    "discover_rca_files",
    "load_test_catalog",
)

_TARGETS_TO_INDEX = (
    "test",
    "test-full",
    "test-cov",
    "test-grafana",
    "demo",
    "cloudwatch-demo",
    "datadog-demo",
    "crashloop-demo",
    "prefect-demo",
    "simulate-k8s-alert",
    "test-k8s-local",
    "test-k8s",
    "test-k8s-datadog",
    "test-k8s-eks",
    "download-cloudopsbench-hf",
    "test-cloudopsbench",
    "trigger-alert",
    "trigger-alert-verify",
    "prefect-local-test",
    "upstream-downstream",
    "flink-demo",
    "deploy",
    "destroy",
    "deploy-lambda",
    "deploy-prefect",
    "deploy-flink",
    "destroy-lambda",
    "destroy-prefect",
    "destroy-flink",
    "deploy-dd-monitors",
    "cleanup-dd-monitors",
    "deploy-eks",
    "destroy-eks",
    "test-openclaw",
    "test-openclaw-synthetic",
)


class _TargetMetadata(TypedDict, total=False):
    display_name: str
    tags: tuple[str, ...]
    requirements: TestRequirement


_TARGET_METADATA: dict[str, _TargetMetadata] = {
    "test": {
        "display_name": "Fast Unit + Prefect E2E",
        "tags": ("ci-safe", "test", "pytest"),
        "requirements": TestRequirement(),
    },
    "test-full": {
        "display_name": "Full Pytest Suite",
        "tags": ("ci-safe", "test", "pytest"),
        "requirements": TestRequirement(),
    },
    "test-cov": {
        "display_name": "Coverage Suite",
        "tags": ("ci-safe", "test", "coverage"),
        "requirements": TestRequirement(),
    },
    "test-grafana": {
        "display_name": "Grafana Integration Tests",
        "tags": ("test", "grafana"),
        "requirements": TestRequirement(env_vars=("ANTHROPIC_API_KEY", "OPENAI_API_KEY")),
    },
    "demo": {
        "display_name": "Prefect ECS Demo",
        "tags": ("demo", "aws"),
        "requirements": TestRequirement(notes=("AWS infra",)),
    },
    "cloudwatch-demo": {
        "display_name": "CloudWatch Demo",
        "tags": ("demo", "aws", "cloudwatch"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "datadog-demo": {
        "display_name": "Datadog Demo",
        "tags": ("demo", "datadog", "k8s", "infra-heavy"),
        "requirements": TestRequirement(
            env_vars=("DD_API_KEY", "DD_APP_KEY"), notes=("Docker/Kubernetes",)
        ),
    },
    "crashloop-demo": {
        "display_name": "CrashLoopBackOff Demo",
        "tags": ("demo", "datadog", "k8s"),
        "requirements": TestRequirement(env_vars=("DD_API_KEY", "DD_APP_KEY")),
    },
    "prefect-demo": {
        "display_name": "Prefect Demo Alias",
        "tags": ("demo", "aws"),
        "requirements": TestRequirement(notes=("AWS infra",)),
    },
    "simulate-k8s-alert": {
        "display_name": "Simulate Kubernetes Alert",
        "tags": ("k8s", "datadog", "infra-heavy"),
        "requirements": TestRequirement(notes=("Kubernetes context",)),
    },
    "test-k8s-local": {
        "display_name": "Kubernetes Local Test",
        "tags": ("k8s", "ci-safe"),
        "requirements": TestRequirement(notes=("Local cluster",)),
    },
    "test-k8s": {
        "display_name": "Kubernetes Test",
        "tags": ("k8s", "infra-heavy"),
        "requirements": TestRequirement(notes=("Kubernetes test env",)),
    },
    "test-k8s-datadog": {
        "display_name": "Kubernetes + Datadog Test",
        "tags": ("k8s", "datadog", "infra-heavy"),
        "requirements": TestRequirement(env_vars=("DD_API_KEY", "DD_APP_KEY")),
    },
    "test-k8s-eks": {
        "display_name": "Kubernetes + Datadog On EKS",
        "tags": ("k8s", "aws", "datadog", "infra-heavy"),
        "requirements": TestRequirement(
            env_vars=("DD_API_KEY", "DD_APP_KEY"), notes=("EKS cluster",)
        ),
    },
    "test-cloudopsbench": {
        "display_name": "Cloud-OpsBench RCA Benchmark",
        "tags": ("synthetic", "cloudopsbench", "k8s", "benchmark"),
        "requirements": TestRequirement(env_vars=("ANTHROPIC_API_KEY",)),
    },
    "download-cloudopsbench-hf": {
        "display_name": "Download Cloud-OpsBench Dataset",
        "tags": ("cloudopsbench", "benchmark", "huggingface"),
        "requirements": TestRequirement(notes=("Hugging Face CLI",)),
    },
    "trigger-alert": {
        "display_name": "Trigger K8s Alert",
        "tags": ("k8s", "datadog"),
        "requirements": TestRequirement(notes=("Kubernetes alert env",)),
    },
    "trigger-alert-verify": {
        "display_name": "Trigger K8s Alert + Verify",
        "tags": ("k8s", "datadog", "infra-heavy"),
        "requirements": TestRequirement(notes=("Slack + Datadog configured",)),
    },
    "prefect-local-test": {
        "display_name": "Prefect Local Test",
        "tags": ("aws", "local"),
        "requirements": TestRequirement(notes=("Optional CLOUD=1",)),
    },
    "upstream-downstream": {
        "display_name": "Upstream/Downstream Lambda E2E",
        "tags": ("aws", "demo"),
        "requirements": TestRequirement(notes=("AWS infra",)),
    },
    "flink-demo": {
        "display_name": "Apache Flink ECS Demo",
        "tags": ("aws", "demo"),
        "requirements": TestRequirement(notes=("AWS infra",)),
    },
    "deploy": {
        "display_name": "Deploy All Test Stacks",
        "tags": ("aws", "infra-heavy", "deploy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "destroy": {
        "display_name": "Destroy All Test Stacks",
        "tags": ("aws", "infra-heavy", "destroy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "deploy-lambda": {
        "display_name": "Deploy Lambda Stack",
        "tags": ("aws", "deploy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "deploy-prefect": {
        "display_name": "Deploy Prefect Stack",
        "tags": ("aws", "deploy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "deploy-flink": {
        "display_name": "Deploy Flink Stack",
        "tags": ("aws", "deploy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "destroy-lambda": {
        "display_name": "Destroy Lambda Stack",
        "tags": ("aws", "destroy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "destroy-prefect": {
        "display_name": "Destroy Prefect Stack",
        "tags": ("aws", "destroy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "destroy-flink": {
        "display_name": "Destroy Flink Stack",
        "tags": ("aws", "destroy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "deploy-dd-monitors": {
        "display_name": "Deploy Datadog Monitors",
        "tags": ("datadog", "deploy"),
        "requirements": TestRequirement(env_vars=("DD_API_KEY", "DD_APP_KEY")),
    },
    "cleanup-dd-monitors": {
        "display_name": "Cleanup Datadog Monitors",
        "tags": ("datadog", "destroy"),
        "requirements": TestRequirement(env_vars=("DD_API_KEY", "DD_APP_KEY")),
    },
    "deploy-eks": {
        "display_name": "Deploy EKS Test Cluster",
        "tags": ("aws", "k8s", "deploy", "infra-heavy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "destroy-eks": {
        "display_name": "Destroy EKS Test Cluster",
        "tags": ("aws", "k8s", "destroy", "infra-heavy"),
        "requirements": TestRequirement(notes=("AWS credentials",)),
    },
    "test-openclaw": {
        "display_name": "OpenClaw Integration Tests",
        "tags": ("ci-safe", "test", "openclaw"),
        "requirements": TestRequirement(),
    },
    "test-openclaw-synthetic": {
        "display_name": "OpenClaw Synthetic Scenario Suite",
        "tags": ("ci-safe", "test", "openclaw", "synthetic"),
        "requirements": TestRequirement(env_vars=("ANTHROPIC_API_KEY",)),
    },
}


def _comment_map_for_makefile(path: Path) -> dict[str, str]:
    lines = path.read_text(encoding="utf-8").splitlines()
    target_comments: dict[str, str] = {}
    comment_buffer: list[str] = []

    for line in lines:
        stripped = line.strip()
        if stripped.startswith("#"):
            comment_buffer.append(stripped.lstrip("# ").strip())
            continue
        if not stripped:
            comment_buffer = []
            continue

        match = re.match(r"^([A-Za-z0-9_-]+):", stripped)
        if match:
            target_comments[match.group(1)] = " ".join(part for part in comment_buffer if part)
            comment_buffer = []
            continue

        comment_buffer = []

    return target_comments


def discover_make_targets() -> list[TestCatalogItem]:
    if not MAKEFILE_PATH.is_file():
        # PyInstaller-bundled builds ship only the ``config/`` tree (see
        # the release workflow), so the Makefile is absent at runtime.
        # Return an empty catalog slice so ``opensre tests`` still launches
        # against whatever sources *are* bundled.
        return []
    comment_map = _comment_map_for_makefile(MAKEFILE_PATH)
    makefile_text = MAKEFILE_PATH.read_text(encoding="utf-8")
    items: list[TestCatalogItem] = []

    for target in _TARGETS_TO_INDEX:
        if not re.search(rf"^{re.escape(target)}:", makefile_text, re.MULTILINE):
            continue
        metadata = _TARGET_METADATA.get(target, {})
        tags = cast(tuple[str, ...], metadata.get("tags") or ("make",))
        requirements = cast(TestRequirement, metadata.get("requirements") or TestRequirement())
        items.append(
            TestCatalogItem(
                id=f"make:{target}",
                kind="make_target",
                display_name=str(metadata.get("display_name") or target),
                description=comment_map.get(target) or f"Run `{target}` from the Makefile.",
                command=("make", target),
                tags=tags,
                source_path=str(MAKEFILE_PATH),
                requirements=requirements,
            )
        )

    return items


def discover_rca_files() -> list[TestCatalogItem]:
    items: list[TestCatalogItem] = []
    if not RCA_DIR.is_dir():
        # ``Path.glob`` on a missing parent returns an empty iterator on
        # CPython, so this isn't a crash today — but the explicit guard
        # documents the bundled-binary contract and matches the shape of
        # the other ``discover_*`` helpers below.
        return items
    for path in sorted(RCA_DIR.glob("*.md")):
        title = path.stem.replace("_", " ").title()
        first_line = path.read_text(encoding="utf-8").splitlines()[0].strip()
        if first_line.startswith("# "):
            title = first_line[2:].strip()
        extra_tags: tuple[str, ...] = ("openclaw",) if path.stem.startswith("openclaw_") else ()
        items.append(
            TestCatalogItem(
                id=f"rca:{path.stem}",
                kind="rca_file",
                display_name=title,
                description="Run a bundled markdown RCA alert fixture.",
                command=("make", "test-rca", f"FILE={path.stem}"),
                tags=("rca", "fixture") + extra_tags,
                source_path=str(path),
                requirements=TestRequirement(env_vars=("ANTHROPIC_API_KEY", "OPENAI_API_KEY")),
            )
        )
    return items


def _discover_rds_synthetic_scenarios() -> list[TestCatalogItem]:
    """One catalog item per RDS synthetic scenario directory.

    Bundled (PyInstaller) builds collect only ``config/`` data files (see
    the release workflow), so ``tests/synthetic/rds_postgres`` is
    absent at runtime and ``iterdir()`` would raise ``FileNotFoundError``.
    Skip cleanly in that case — the synthetic-suite catalog entries are
    only meaningful when the scenarios are on disk anyway.
    """
    items: list[TestCatalogItem] = []
    if not SYNTHETIC_SCENARIOS_DIR.is_dir():
        return items
    req = TestRequirement(env_vars=("ANTHROPIC_API_KEY",))
    for scenario_dir in sorted(SYNTHETIC_SCENARIOS_DIR.iterdir()):
        if not scenario_dir.is_dir() or scenario_dir.name.startswith("_"):
            continue
        scenario_id = scenario_dir.name
        # Read display name from scenario.yml if present, else use directory name.
        display_name = scenario_id
        scenario_yml = scenario_dir / "scenario.yml"
        if scenario_yml.exists():
            try:
                meta = yaml.safe_load(scenario_yml.read_text(encoding="utf-8")) or {}
                failure_mode = meta.get("failure_mode", "")
                if failure_mode:
                    display_name = f"{scenario_id}  [{failure_mode}]"
            except (OSError, UnicodeDecodeError, yaml.YAMLError, TypeError, ValueError):
                display_name = scenario_id
        items.append(
            TestCatalogItem(
                id=f"synthetic:{scenario_id}",
                kind="cli_command",
                display_name=display_name,
                description=f"Run the '{scenario_id}' synthetic RCA scenario against the mock backend.",
                command=("opensre", "tests", "synthetic", "--scenario", scenario_id),
                tags=("synthetic", "rds", "test"),
                source_path=str(scenario_dir),
                requirements=req,
            )
        )
    return items


def _discover_openclaw_synthetic_scenarios() -> list[TestCatalogItem]:
    items: list[TestCatalogItem] = []
    if not OPENCLAW_SYNTHETIC_SCENARIOS_DIR.is_dir():
        return items

    requirements = TestRequirement(notes=("Configured LLM provider",))
    for scenario_dir in sorted(OPENCLAW_SYNTHETIC_SCENARIOS_DIR.iterdir()):
        if not scenario_dir.is_dir() or scenario_dir.name.startswith("_"):
            continue
        scenario_id = scenario_dir.name
        display_name = scenario_id.replace("_", " ")
        description = "Run a synthetic OpenClaw-backed RCA scenario against the fixture bridge."

        scenario_json = scenario_dir / "scenario.json"
        if scenario_json.exists():
            try:
                meta = json.loads(scenario_json.read_text(encoding="utf-8"))
                scenario_description = str(meta.get("description", "")).strip()
                if scenario_description:
                    display_name = scenario_description[:80]
                    description = scenario_description
            except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError):
                # Ignore unreadable/invalid metadata and keep default name/description.
                pass

        items.append(
            TestCatalogItem(
                id=f"openclaw-synthetic:{scenario_id}",
                kind="cli_command",
                display_name=display_name,
                description=description,
                command=("opensre", "tests", "openclaw-synthetic", "--scenario", scenario_id),
                tags=("synthetic", "openclaw", "rca", "ci-safe"),
                source_path=str(scenario_dir),
                requirements=requirements,
            )
        )

    return items


def _discover_cloudopsbench_suite() -> list[TestCatalogItem]:
    benchmark_dir = CLOUDOPSBENCH_DIR / "benchmark"
    if not benchmark_dir.is_dir():
        return []
    return [
        TestCatalogItem(
            id="synthetic:cloudopsbench",
            kind="cli_command",
            display_name="Cloud-OpsBench RCA Benchmark",
            description="Run the downloaded Cloud-OpsBench corpus through the OpenSRE runner.",
            command=("opensre", "tests", "cloudopsbench"),
            tags=("synthetic", "cloudopsbench", "k8s", "benchmark"),
            source_path=str(CLOUDOPSBENCH_DIR),
            requirements=TestRequirement(env_vars=("ANTHROPIC_API_KEY",)),
        )
    ]


def discover_cli_commands() -> list[TestCatalogItem]:
    """Catalog entries for opensre sub-commands that have no Makefile equivalent."""
    return [
        *_discover_rds_synthetic_scenarios(),
        *_discover_openclaw_synthetic_scenarios(),
        *_discover_cloudopsbench_suite(),
    ]


def load_test_catalog() -> TestCatalog:
    items: list[TestCatalogItem] = []
    items.extend(discover_cli_commands())
    items.extend(discover_make_targets())
    items.extend(discover_rca_files())
    items.sort(key=lambda item: item.display_name.lower())
    return TestCatalog(items=tuple(items))