Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ packages = [{include = "omnibase_infra", from = "src"}]
omni-infra = "omnibase_infra.cli.commands:cli"
onex-runtime = "omnibase_infra.runtime.kernel:main"
emit-daemon = "omnibase_infra.runtime.emit_daemon.cli:main"
onex-infra-test = "omnibase_infra.cli.infra_test.cli:cli"

[tool.poetry.dependencies]
python = "^3.12"
Expand Down
19 changes: 19 additions & 0 deletions src/omnibase_infra/cli/infra_test/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2025 OmniNode Team
"""ONEX Infrastructure Integration Test CLI.

Provides the ``onex-infra-test`` command for end-to-end validation of the
ONEX registration pipeline including smoke, idempotency, and failure suites.
"""

from omnibase_infra.cli.infra_test._helpers import (
get_broker,
get_consul_addr,
get_postgres_dsn,
)

__all__ = [
"get_broker",
"get_consul_addr",
"get_postgres_dsn",
]
61 changes: 61 additions & 0 deletions src/omnibase_infra/cli/infra_test/_helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2025 OmniNode Team
"""Shared helpers for the infra-test CLI.

Centralises environment-resolution functions used by multiple subcommands
(``verify``, ``run``, ``introspect``).
"""

from __future__ import annotations

import os
from urllib.parse import quote_plus

Comment thread
coderabbitai[bot] marked this conversation as resolved.

def get_broker() -> str:
"""Resolve Kafka broker address from environment.

Returns:
Kafka bootstrap server address.
"""
return os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:29092")


def get_consul_addr() -> str:
"""Resolve Consul HTTP address from environment.

Returns:
Full Consul HTTP URL (e.g. ``http://localhost:8500``).
"""
host = os.getenv("CONSUL_HOST", "localhost")
port = os.getenv("CONSUL_PORT", "8500")
scheme = os.getenv("CONSUL_SCHEME", "http")
if scheme not in ("http", "https"):
raise ValueError(f"CONSUL_SCHEME must be 'http' or 'https', got {scheme!r}.")
if not port.isdigit():
raise ValueError(f"CONSUL_PORT must be numeric, got {port!r}.")
return f"{scheme}://{host}:{port}"


def get_postgres_dsn() -> str:
"""Build PostgreSQL DSN from environment variables.

Defaults are for local E2E test environments only -- never use in
production. All values (including the fallback password) are overridden
via environment variables in real deployments.

Returns:
PostgreSQL connection string.
"""
host = os.getenv("POSTGRES_HOST", "localhost")
port = os.getenv("POSTGRES_PORT", "5433")
db = os.getenv("POSTGRES_DATABASE", "omninode_bridge")
user = os.getenv("POSTGRES_USER", "postgres")
password = os.getenv("POSTGRES_PASSWORD", "test-password")
if "@" in host:
raise ValueError(
f"POSTGRES_HOST contains '@' ({host!r}), which would produce a malformed DSN."
)
if not port.isdigit():
raise ValueError(f"POSTGRES_PORT must be numeric, got {port!r}.")
return f"postgresql://{quote_plus(user)}:{quote_plus(password)}@{host}:{port}/{quote_plus(db)}"
61 changes: 61 additions & 0 deletions src/omnibase_infra/cli/infra_test/cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2025 OmniNode Team
"""Main CLI entry point for ``onex-infra-test``.

Provides end-to-end integration testing commands for the ONEX infrastructure
stack. All subcommands are registered here.

Usage::

onex-infra-test env up # Start local infra
onex-infra-test env down # Stop local infra
onex-infra-test introspect # Trigger introspection
onex-infra-test verify registry # Check registry state
onex-infra-test verify topics # Check topic compliance
onex-infra-test verify snapshots # Check snapshot topic state
onex-infra-test verify idempotency # Run twice, assert no duplicates
onex-infra-test run --suite smoke # Happy path
onex-infra-test run --suite failure # Runtime-down-then-recovery
"""

from __future__ import annotations

import click

from omnibase_infra.cli.infra_test.env import env
from omnibase_infra.cli.infra_test.introspect import introspect
from omnibase_infra.cli.infra_test.run_suite import run_suite
from omnibase_infra.cli.infra_test.verify import verify


@click.group()
@click.option(
"--compose-file",
default="docker/docker-compose.e2e.yml",
envvar="ONEX_COMPOSE_FILE",
help="Path to Docker Compose file.",
show_default=True,
)
@click.option(
"--project-name",
default="omnibase-infra",
envvar="ONEX_PROJECT_NAME",
help="Docker Compose project name.",
show_default=True,
)
@click.pass_context
def cli(ctx: click.Context, compose_file: str, project_name: str) -> None:
"""ONEX Infrastructure Integration Test CLI."""
ctx.ensure_object(dict)
ctx.obj["compose_file"] = compose_file
ctx.obj["project_name"] = project_name


cli.add_command(env)
cli.add_command(introspect)
cli.add_command(verify)
cli.add_command(run_suite)


if __name__ == "__main__":
cli()
218 changes: 218 additions & 0 deletions src/omnibase_infra/cli/infra_test/env.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,218 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2025 OmniNode Team
"""Environment management commands (``env up`` / ``env down``).

Wraps Docker Compose to manage the E2E testing infrastructure stack.
"""

from __future__ import annotations

import subprocess
import time

import click
from rich.console import Console

console = Console()


def _run_compose(
compose_file: str,
project_name: str,
args: list[str],
*,
capture: bool = False,
) -> subprocess.CompletedProcess[str]:
"""Run a ``docker compose`` command.

Args:
compose_file: Path to the Docker Compose file.
project_name: Docker Compose project name.
args: Additional arguments to ``docker compose``.
capture: If True, capture stdout/stderr instead of streaming.

Returns:
CompletedProcess result.

Raises:
SystemExit: If the command fails.
"""
cmd = [
"docker",
"compose",
"-f",
compose_file,
"-p",
project_name,
*args,
]
result = subprocess.run(
cmd,
capture_output=capture,
text=True,
check=False,
timeout=300,
)
if result.returncode != 0:
if capture and result.stderr:
console.print(f"[red]{result.stderr.strip()}[/red]")
raise SystemExit(result.returncode)
return result


def _wait_for_healthy(
compose_file: str,
project_name: str,
timeout_seconds: int = 120,
) -> bool:
"""Poll ``docker compose ps`` until all services are healthy.

Args:
compose_file: Path to the Docker Compose file.
project_name: Docker Compose project name.
timeout_seconds: Maximum time to wait.

Returns:
True if all services healthy, False on timeout.
"""
start = time.monotonic()
while time.monotonic() - start < timeout_seconds:
result = subprocess.run(
[
"docker",
"compose",
"-f",
compose_file,
"-p",
project_name,
"ps",
"--format",
"{{.Service}}\t{{.Health}}",
],
capture_output=True,
text=True,
check=False,
timeout=30,
)
if result.returncode != 0:
time.sleep(2)
continue

lines = [
line.strip() for line in result.stdout.strip().splitlines() if line.strip()
]
if not lines:
time.sleep(2)
continue

# Check health status for each service
all_healthy = True
services_checked = 0
for line in lines:
parts = line.split("\t", 1)
if len(parts) < 2:
continue
services_checked += 1
service, health = parts[0].strip(), parts[1].strip().lower()
# Services without healthchecks show empty or "-"
if health and health not in ("healthy", "-"):
all_healthy = False
break

if services_checked > 0 and all_healthy:
return True

time.sleep(2)

return False


@click.group()
def env() -> None:
"""Manage the E2E testing environment."""


@env.command("up")
@click.option(
"--profile",
multiple=True,
help="Docker Compose profiles to activate (e.g. --profile runtime).",
)
@click.option(
"--wait/--no-wait",
default=True,
help="Wait for all services to become healthy.",
show_default=True,
)
@click.option(
"--timeout",
default=120,
help="Timeout in seconds for health checks.",
show_default=True,
)
@click.pass_context
def env_up(
ctx: click.Context,
profile: tuple[str, ...],
wait: bool,
timeout: int,
) -> None:
"""Start local infrastructure for E2E testing.

Starts PostgreSQL, Redpanda, Consul, Valkey, and Infisical via
Docker Compose. Optionally activates the ``runtime`` profile to
include the ONEX runtime container.
"""
compose_file = ctx.obj["compose_file"]
project_name = ctx.obj["project_name"]

console.print("[bold blue]Starting E2E infrastructure...[/bold blue]")

args = ["up", "-d"]
for p in profile:
args = ["--profile", p, *args]

_run_compose(compose_file, project_name, args)

if wait:
console.print(
f"[yellow]Waiting for services to become healthy (timeout={timeout}s)...[/yellow]"
)
if _wait_for_healthy(compose_file, project_name, timeout):
console.print("[bold green]All services healthy.[/bold green]")
else:
console.print("[bold red]Timeout: not all services are healthy.[/bold red]")
# Show current status for debugging
_run_compose(compose_file, project_name, ["ps"])
raise SystemExit(1)

# Show running services
_run_compose(compose_file, project_name, ["ps"])


@env.command("down")
@click.option(
"--volumes/--no-volumes",
"-v",
default=True,
help="Remove volumes (clean state).",
show_default=True,
)
@click.pass_context
def env_down(ctx: click.Context, volumes: bool) -> None:
"""Stop and remove E2E infrastructure.

Tears down all Docker Compose services. By default also removes
volumes so the next ``env up`` starts with a clean state.
"""
compose_file = ctx.obj["compose_file"]
project_name = ctx.obj["project_name"]

console.print("[bold blue]Stopping E2E infrastructure...[/bold blue]")

args = ["down"]
if volumes:
args.append("-v")

_run_compose(compose_file, project_name, args)
console.print("[bold green]Infrastructure stopped.[/bold green]")
Loading