fix: resolve CI/CD test, lint, and type-check failures
This commit is contained in:
155
app/api_snapshot/cli/cli.py
Normal file
155
app/api_snapshot/cli/cli.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user