From 6fb7c87340fd48182eeac9235ae22ea1b84171c3 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 14:16:12 +0000 Subject: [PATCH] fix: resolve CI/CD test, lint, and type-check failures --- app/api_snapshot/cli/cli.py | 155 ++++++++++++++++++++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 app/api_snapshot/cli/cli.py diff --git a/app/api_snapshot/cli/cli.py b/app/api_snapshot/cli/cli.py new file mode 100644 index 0000000..dfbc41d --- /dev/null +++ b/app/api_snapshot/cli/cli.py @@ -0,0 +1,155 @@ +"""Main CLI module for api-snapshot-cli.""" + +import os +from typing import Optional + +import click +from rich.console import Console +from rich.table import Table + +from api_snapshot.snapshot.manager import SnapshotManager + +console = Console() + + +def get_snapshot_dir() -> str: + """Get the snapshot directory from environment or default.""" + return os.environ.get("API_SNAPSHOT_DIR", "./snapshots") + + +def get_default_host() -> str: + """Get the default host from environment.""" + return os.environ.get("API_SNAPSHOT_HOST", "127.0.0.1") + + +def get_default_port() -> int: + """Get the default port from environment.""" + try: + return int(os.environ.get("API_SNAPSHOT_PORT", "8080")) + except ValueError: + return 8080 + + +@click.group() +@click.version_option(version="0.1.0", prog_name="api-snapshot") +@click.option("--verbose", "-v", is_flag=True, help="Enable verbose output") +@click.option("--snapshot-dir", default=None, help="Override snapshot directory") +@click.pass_context +def main(ctx: click.Context, verbose: bool, snapshot_dir: Optional[str]) -> None: + """API Snapshot CLI - Record HTTP API traffic and generate local mock servers. + + This tool allows you to: + - Record HTTP requests and responses from any API + - Save them as JSON snapshots + - Spin up a local mock server to replay recorded responses + """ + ctx.ensure_object(dict) + ctx.obj["verbose"] = verbose + ctx.obj["snapshot_dir"] = snapshot_dir or get_snapshot_dir() + + if verbose: + console.print("[cyan]Verbose mode enabled[/cyan]") + console.print(f"[cyan]Snapshot directory: {ctx.obj['snapshot_dir']}[/cyan]") + + +@main.command(name="list") +@click.pass_context +def list_snapshots(ctx: click.Context) -> None: + """List all available snapshots.""" + snapshot_dir = ctx.obj["snapshot_dir"] + manager = SnapshotManager(snapshot_dir) + + snapshots = manager.list_snapshots() + + if not snapshots: + console.print("[yellow]No snapshots found.[/yellow]") + return + + table = Table(title="Available Snapshots") + table.add_column("Name", style="cyan") + table.add_column("Created", style="green") + table.add_column("Endpoints", style="magenta") + table.add_column("Size", style="blue") + + for snapshot in snapshots: + table.add_row( + snapshot["name"], + snapshot["created"], + str(snapshot["endpoint_count"]), + snapshot["size"] + ) + + console.print(table) + + +@main.command(name="info") +@click.argument("name") +@click.pass_context +def snapshot_info(ctx: click.Context, name: str) -> None: + """Show detailed information about a snapshot.""" + snapshot_dir = ctx.obj["snapshot_dir"] + 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() + + table = Table(title=f"Snapshot: {name}") + table.add_column("Property", style="yellow") + table.add_column("Value", style="cyan") + + version = getattr(snapshot.metadata, 'version', '1.0') + timestamp = getattr(snapshot.metadata, 'timestamp', 'Unknown') + description = getattr(snapshot.metadata, 'description', 'No description') + + table.add_row("Version", version) + table.add_row("Created", timestamp) + table.add_row("Description", description) + table.add_row("Endpoints", str(len(snapshot.requests))) + + console.print(table) + + if snapshot.requests: + endpoint_table = Table(title="Recorded Endpoints") + endpoint_table.add_column("Method", style="cyan") + endpoint_table.add_column("Path", style="green") + endpoint_table.add_column("Status", style="magenta") + + for req in snapshot.requests: + endpoint_table.add_row( + req.request.method, + req.request.url, + str(req.response.status_code) + ) + + console.print(endpoint_table) + + +@main.command(name="delete") +@click.argument("name") +@click.option("--force", "-f", is_flag=True, help="Skip confirmation") +@click.pass_context +def delete_snapshot(ctx: click.Context, name: str, force: bool) -> None: + """Delete a snapshot.""" + snapshot_dir = ctx.obj["snapshot_dir"] + manager = SnapshotManager(snapshot_dir) + + if not force: + if not click.confirm(f"Delete snapshot '{name}'?"): + console.print("[yellow]Cancelled[/yellow]") + return + + try: + manager.delete_snapshot(name) + console.print(f"[green]Deleted snapshot '{name}'[/green]") + except FileNotFoundError: + console.print(f"[red]Error: Snapshot '{name}' not found[/red]") + raise click.Abort() + except Exception as e: + console.print(f"[red]Error deleting snapshot: {e}[/red]") + raise click.Abort()