From df1c041343e3054f90282b2cba5e05de6bf6e0fd Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 14:16:13 +0000 Subject: [PATCH] fix: resolve CI/CD test, lint, and type-check failures --- app/api_snapshot/cli/serve.py | 188 ++++++++++++++++++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 app/api_snapshot/cli/serve.py diff --git a/app/api_snapshot/cli/serve.py b/app/api_snapshot/cli/serve.py new file mode 100644 index 0000000..d360aa4 --- /dev/null +++ b/app/api_snapshot/cli/serve.py @@ -0,0 +1,188 @@ +"""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)