189 lines
6.3 KiB
Python
189 lines
6.3 KiB
Python
"""Serve command module."""
|
|
|
|
from typing import Optional
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from api_snapshot.server.server import MockServer
|
|
from api_snapshot.snapshot.manager import SnapshotManager
|
|
|
|
console = Console()
|
|
|
|
|
|
@click.command(name="serve")
|
|
@click.argument("name")
|
|
@click.option("--host", "-H", default=None, help="Host to bind to")
|
|
@click.option("--port", "-p", type=int, default=None, help="Port to listen on")
|
|
@click.option("--snapshot-dir", default=None, help="Override snapshot directory")
|
|
@click.option("--latency", "-l", default="original", type=click.Choice([
|
|
"original", "fixed", "random", "none"
|
|
]), help="Latency mode for replay")
|
|
@click.option("--latency-ms", type=int, default=None, help="Fixed latency in milliseconds")
|
|
@click.option("--latency-min", type=int, default=None, help="Minimum latency for random mode (ms)")
|
|
@click.option("--latency-max", type=int, default=None, help="Maximum latency for random mode (ms)")
|
|
@click.option("--no-header", is_flag=True, help="Suppress startup banner")
|
|
@click.pass_context
|
|
def serve_command(
|
|
ctx: click.Context,
|
|
name: str,
|
|
host: Optional[str],
|
|
port: Optional[int],
|
|
snapshot_dir: Optional[str],
|
|
latency: str,
|
|
latency_ms: Optional[int],
|
|
latency_min: Optional[int],
|
|
latency_max: Optional[int],
|
|
no_header: bool
|
|
) -> None:
|
|
"""Start a mock server from a snapshot.
|
|
|
|
NAME is the name of the snapshot to serve.
|
|
|
|
Examples:
|
|
api-snapshot serve my-api
|
|
api-snapshot serve my-api --host 0.0.0.0 --port 9000
|
|
api-snapshot serve my-api --latency fixed --latency-ms 100
|
|
"""
|
|
import os
|
|
import signal
|
|
import sys
|
|
|
|
snapshot_dir = snapshot_dir or ctx.obj.get("snapshot_dir", "./snapshots")
|
|
|
|
manager = SnapshotManager(snapshot_dir)
|
|
|
|
try:
|
|
snapshot = manager.load_snapshot(name)
|
|
except FileNotFoundError:
|
|
console.print(f"[red]Error: Snapshot '{name}' not found[/red]")
|
|
|
|
snapshots = manager.list_snapshots()
|
|
if snapshots:
|
|
console.print("\nAvailable snapshots:")
|
|
table = Table()
|
|
table.add_column("Name", style="cyan")
|
|
for s in snapshots:
|
|
table.add_row(s["name"])
|
|
console.print(table)
|
|
else:
|
|
console.print("No snapshots available. Create one with 'api-snapshot record'")
|
|
|
|
raise click.Abort()
|
|
except Exception as e:
|
|
console.print(f"[red]Error loading snapshot: {e}[/red]")
|
|
raise click.Abort()
|
|
|
|
resolved_host = host or os.environ.get("API_SNAPSHOT_HOST", "127.0.0.1")
|
|
resolved_port = port or int(os.environ.get("API_SNAPSHOT_PORT", "8080"))
|
|
|
|
if latency == "fixed" and latency_ms is None:
|
|
latency_ms = 100
|
|
console.print("[yellow]Using default latency of 100ms[/yellow]")
|
|
|
|
if latency == "random" and (latency_min is None or latency_max is None):
|
|
latency_min = 50
|
|
latency_max = 500
|
|
console.print(f"[yellow]Using random latency range: {latency_min}-{latency_max}ms[/yellow]")
|
|
|
|
random_range = None
|
|
if latency == "random":
|
|
random_range = (latency_min or 50, latency_max or 500)
|
|
|
|
if not no_header:
|
|
console.print("[cyan]Starting API Mock Server[/cyan]")
|
|
console.print("=" * 40)
|
|
console.print(f"[green]Snapshot:[/green] {name}")
|
|
desc = snapshot.metadata.description or "No description"
|
|
console.print(f"[green]Description:[/green] {desc}")
|
|
console.print(f"[green]Endpoints:[/green] {len(snapshot.requests)}")
|
|
console.print(f"[green]Address:[/green] http://{resolved_host}:{resolved_port}")
|
|
console.print(f"[green]Latency:[/green] {latency}")
|
|
if latency == "fixed":
|
|
console.print(f"[green]Latency (ms):[/green] {latency_ms}")
|
|
elif latency == "random" and random_range:
|
|
console.print(f"[green]Latency range:[/green] {random_range[0]}-{random_range[1]}ms")
|
|
console.print("-" * 40)
|
|
console.print("Press [yellow]Ctrl+C[/yellow] to stop")
|
|
console.print("-" * 40)
|
|
|
|
server = MockServer(
|
|
snapshot=snapshot,
|
|
host=resolved_host,
|
|
port=resolved_port,
|
|
latency_mode=latency,
|
|
fixed_latency_ms=latency_ms,
|
|
random_latency_range=random_range
|
|
)
|
|
|
|
def signal_handler(sig, frame):
|
|
console.print("\n[yellow]Shutting down mock server...[/yellow]")
|
|
sys.exit(0)
|
|
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
|
|
try:
|
|
server.run(debug=False)
|
|
except Exception as e:
|
|
console.print(f"[red]Error running server: {e}[/red]")
|
|
raise click.Abort()
|
|
|
|
|
|
@click.command(name="serve-info")
|
|
@click.argument("name")
|
|
@click.option("--snapshot-dir", default=None, help="Override snapshot directory")
|
|
@click.pass_context
|
|
def serve_info_command(
|
|
ctx: click.Context,
|
|
name: str,
|
|
snapshot_dir: Optional[str]
|
|
) -> None:
|
|
"""Show mock server configuration for a snapshot without starting it."""
|
|
import os
|
|
|
|
snapshot_dir = snapshot_dir or ctx.obj.get("snapshot_dir", "./snapshots")
|
|
|
|
manager = SnapshotManager(snapshot_dir)
|
|
|
|
try:
|
|
snapshot = manager.load_snapshot(name)
|
|
except FileNotFoundError:
|
|
console.print(f"[red]Error: Snapshot '{name}' not found[/red]")
|
|
raise click.Abort()
|
|
except Exception as e:
|
|
console.print(f"[red]Error loading snapshot: {e}[/red]")
|
|
raise click.Abort()
|
|
|
|
host = os.environ.get("API_SNAPSHOT_HOST", "127.0.0.1")
|
|
port = int(os.environ.get("API_SNAPSHOT_PORT", "8080"))
|
|
|
|
console.print(f"[cyan]Mock Server Configuration: {name}[/cyan]")
|
|
console.print("=" * 40)
|
|
console.print(f"Host: {host}")
|
|
console.print(f"Port: {port}")
|
|
console.print(f"URL: http://{host}:{port}")
|
|
console.print(f"Endpoints: {len(snapshot.requests)}")
|
|
|
|
table = Table(title="Available Endpoints")
|
|
table.add_column("Method", style="cyan")
|
|
table.add_column("Path", style="green")
|
|
table.add_column("Status", style="magenta")
|
|
|
|
for pair in snapshot.requests:
|
|
url = pair.request.url
|
|
if url.startswith("http"):
|
|
from urllib.parse import urlparse
|
|
path = urlparse(url).path
|
|
else:
|
|
path = url
|
|
|
|
table.add_row(
|
|
pair.request.method,
|
|
path,
|
|
str(pair.response.status_code)
|
|
)
|
|
|
|
console.print(table)
|