diff --git a/.github/workflows/ci-tests-mkdocs-deployment.yml b/.github/workflows/ci-tests-mkdocs-deployment.yml index 39d08b52..aece728a 100644 --- a/.github/workflows/ci-tests-mkdocs-deployment.yml +++ b/.github/workflows/ci-tests-mkdocs-deployment.yml @@ -38,7 +38,7 @@ jobs: run: ruff check aws_resource_validator tests - name: Mypy - run: mypy aws_resource_validator/core aws_resource_validator/generator + run: mypy aws_resource_validator/core aws_resource_validator/generator aws_resource_validator/cli test: runs-on: ubuntu-latest diff --git a/.gitignore b/.gitignore index ae5d3124..30e797bd 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,4 @@ Thumbs.db # Local scratch *.tmp /tmp_* +.agents/ diff --git a/README.md b/README.md index bcafe53e..8e621add 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Features +- **Developer CLI (`arv`)**: Validate names, inspect AWS constraints, and generate synthetic test values directly in your terminal with Rich UI output. - **Validation**: Check if a given AWS resource name meets the AWS naming constraints. - **Constraint Display**: Display constraints for different AWS resource names. - **Pattern Generation**: Generate compatible patterns for AWS resource names for testing purposes. @@ -53,7 +54,37 @@ pip install 'aws-resource-validator[generator]' See [`docs/packaging.md`](docs/packaging.md) for the full list of standalone service packages, shard membership, and the detailed install matrix. -## Usage Example +## Developer CLI (`arv`) + +The package includes a command-line tool, `arv`, providing instant terminal validation, constraint inspection, and test data generation powered by [Rich](https://github.com/Textualize/rich). + +```sh +# Validate an AWS resource name against AWS constraints +arv validate lambda FunctionName "my-lambda-function" + +# Enforce strict end-to-end regex matching (rather than botocore prefix matching) +arv validate lambda FunctionName "invalid func! name" --strict + +# Inspect naming limits, regex patterns, and sample values +arv inspect lambda FunctionName + +# Inspect all patterned shapes available for a service +arv inspect dynamodb + +# Generate compliant synthetic test names +arv generate lambda FunctionName --count 3 + +# Raw output for shell scripts and pipelines +arv generate lambda FunctionName --plain + +# Machine-readable JSON output for CI/CD checks +arv validate lambda FunctionName "my-func" --json + +# List registered AWS services and shape counts +arv list --search s3 +``` + +## Python Usage Example Here's a simple example demonstrating how to use `aws_resource_validator`: diff --git a/aws_resource_validator/cli/__init__.py b/aws_resource_validator/cli/__init__.py new file mode 100644 index 00000000..8621ba3c --- /dev/null +++ b/aws_resource_validator/cli/__init__.py @@ -0,0 +1,7 @@ +"""Developer CLI package for aws_resource_validator.""" + +from __future__ import annotations + +from aws_resource_validator.cli.app import app + +__all__ = ["app"] diff --git a/aws_resource_validator/cli/__main__.py b/aws_resource_validator/cli/__main__.py new file mode 100644 index 00000000..d2402c4d --- /dev/null +++ b/aws_resource_validator/cli/__main__.py @@ -0,0 +1,14 @@ +"""Entry point for the ``arv`` CLI command.""" + +from __future__ import annotations + +from aws_resource_validator.cli.app import app + + +def main() -> None: + """Execute the arv CLI.""" + app() + + +if __name__ == "__main__": + main() diff --git a/aws_resource_validator/cli/app.py b/aws_resource_validator/cli/app.py new file mode 100644 index 00000000..86513f4f --- /dev/null +++ b/aws_resource_validator/cli/app.py @@ -0,0 +1,233 @@ +"""Typer application for the ``arv`` command-line interface.""" + +from __future__ import annotations + +import json +import platform +import sys +from typing import Any + +import rich +import typer + +from aws_resource_validator import __version__ +from aws_resource_validator.cli.diagnostics import diagnose_validation +from aws_resource_validator.cli.formatters import ( + get_console, + print_error, + print_generated_values, + print_inspect_service, + print_inspect_shape, + print_services_list, + print_validation_result, +) +from aws_resource_validator.cli.resolver import ( + get_registry, + list_services, + resolve_service, + resolve_shape, +) + +app = typer.Typer( + name="arv", + help="AWS Resource Validator: inspect AWS constraints, validate resource names, and generate test data.", + no_args_is_help=True, + add_completion=True, +) + + +def _output_json(data: Any) -> None: + """Print machine-readable JSON directly to standard output without terminal markup.""" + print(json.dumps(data, indent=2)) + + +@app.command("validate") +def validate( + service: str = typer.Argument(..., help="AWS service name (e.g. lambda, dynamodb, s3control)."), + shape: str = typer.Argument(..., help="AWS shape or resource name (e.g. FunctionName, TableName)."), + value: str = typer.Argument(..., help="Candidate value to test against AWS naming constraints."), + strict: bool = typer.Option( + False, + "--strict", + "-s", + help="Enforce end-to-end regex matching instead of botocore prefix matching.", + ), + json_output: bool = typer.Option(False, "--json", help="Output result as JSON."), +) -> None: + """Validate an AWS resource name against botocore patterns and length limits.""" + console = get_console() + svc, svc_suggestions = resolve_service(service) + + if svc is None: + if json_output: + _output_json({"error": f"Service '{service}' not found", "suggestions": svc_suggestions}) + else: + print_error(console, f"Service '{service}' not found.", suggestions=svc_suggestions) + raise typer.Exit(code=2) + + obj, shape_suggestions = resolve_shape(svc, shape) + if obj is None: + if json_output: + _output_json( + {"error": f"Shape '{shape}' not found in {svc.service_name}", "suggestions": shape_suggestions} + ) + else: + print_error( + console, + f"Shape '{shape}' not found in service '{svc.service_name}'.", + suggestions=shape_suggestions, + ) + raise typer.Exit(code=2) + + result = diagnose_validation(obj, svc.service_name, value, strict=strict) + + if json_output: + _output_json(result.to_dict()) + else: + print_validation_result(console, result) + + if not result.is_valid: + raise typer.Exit(code=1) + + +@app.command("inspect") +def inspect( + service: str = typer.Argument(..., help="AWS service name (e.g. lambda, dynamodb, s3control)."), + shape: str | None = typer.Argument(None, help="Optional shape name to inspect specifically."), + json_output: bool = typer.Option(False, "--json", help="Output details as JSON."), +) -> None: + """Inspect regex pattern constraints, length limits, and sample values.""" + console = get_console() + svc, svc_suggestions = resolve_service(service) + + if svc is None: + if json_output: + _output_json({"error": f"Service '{service}' not found", "suggestions": svc_suggestions}) + else: + print_error(console, f"Service '{service}' not found.", suggestions=svc_suggestions) + raise typer.Exit(code=2) + + if shape is not None: + obj, shape_suggestions = resolve_shape(svc, shape) + if obj is None: + if json_output: + _output_json( + {"error": f"Shape '{shape}' not found in {svc.service_name}", "suggestions": shape_suggestions} + ) + else: + print_error( + console, + f"Shape '{shape}' not found in service '{svc.service_name}'.", + suggestions=shape_suggestions, + ) + raise typer.Exit(code=2) + + if json_output: + data = { + "service": svc.service_name, + "shape": obj.name, + "type": obj.type, + "pattern": obj.pattern, + "min_length": obj.min_length, + "max_length": obj.max_length, + } + _output_json(data) + else: + print_inspect_shape(console, svc.service_name, obj) + return + + # Inspect all shapes in service + if json_output: + shapes_data = [ + { + "shape": obj.name, + "type": obj.type, + "pattern": obj.pattern, + "min_length": obj.min_length, + "max_length": obj.max_length, + } + for obj in svc.api_objects.values() + ] + _output_json({"service": svc.service_name, "shapes": shapes_data}) + else: + print_inspect_service(console, svc) + + +@app.command("generate") +def generate( + service: str = typer.Argument(..., help="AWS service name (e.g. lambda, dynamodb)."), + shape: str = typer.Argument(..., help="Shape name (e.g. FunctionName)."), + count: int = typer.Option(1, "--count", "-c", min=1, max=100, help="Number of synthetic values to generate."), + plain: bool = typer.Option( + False, + "--plain", + "-p", + help="Print raw values one per line (ideal for shell scripts and pipelines).", + ), + json_output: bool = typer.Option(False, "--json", help="Output values as a JSON list."), +) -> None: + """Generate compliant synthetic test values satisfying AWS constraints.""" + console = get_console() + svc, svc_suggestions = resolve_service(service) + + if svc is None: + if json_output: + _output_json({"error": f"Service '{service}' not found", "suggestions": svc_suggestions}) + else: + print_error(console, f"Service '{service}' not found.", suggestions=svc_suggestions) + raise typer.Exit(code=2) + + obj, shape_suggestions = resolve_shape(svc, shape) + if obj is None: + if json_output: + _output_json( + {"error": f"Shape '{shape}' not found in {svc.service_name}", "suggestions": shape_suggestions} + ) + else: + print_error( + console, + f"Shape '{shape}' not found in service '{svc.service_name}'.", + suggestions=shape_suggestions, + ) + raise typer.Exit(code=2) + + values = [obj.generate() for _ in range(count)] + + if json_output: + _output_json({"service": svc.service_name, "shape": obj.name, "values": values}) + else: + print_generated_values(console, svc.service_name, obj.name, values, plain=plain) + + +@app.command("list") +def list_cmd( + service: str | None = typer.Argument(None, help="Optional service name to list shapes for."), + search: str | None = typer.Option(None, "--search", "-s", help="Filter services by name substring."), + json_output: bool = typer.Option(False, "--json", help="Output as JSON."), +) -> None: + """List supported AWS services and their validated shape counts.""" + console = get_console() + + if service is not None: + # Delegate to inspect + inspect(service=service, shape=None, json_output=json_output) + return + + services = list_services(query=search) + total_registry_count = len(get_registry()) + + if json_output: + data = [{"service": s, "shape_count": c} for s, c in services] + _output_json(data) + else: + print_services_list(console, services, total_count=total_registry_count) + + +@app.command("version") +def version() -> None: + """Display version information for arv and its runtime environment.""" + console = get_console() + console.print(f"[bold cyan]aws-resource-validator (arv)[/bold cyan] version [bold green]{__version__}[/bold green]") + console.print(f" Python: {platform.python_version()} ({sys.executable})") + console.print(f" Rich: {rich.__file__}") + console.print(f" Typer: {typer.__version__}") diff --git a/aws_resource_validator/cli/diagnostics.py b/aws_resource_validator/cli/diagnostics.py new file mode 100644 index 00000000..0acfa8bd --- /dev/null +++ b/aws_resource_validator/cli/diagnostics.py @@ -0,0 +1,98 @@ +"""Detailed diagnostic analysis for AWS resource name validation.""" + +from __future__ import annotations + +import re +from dataclasses import asdict, dataclass, field +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from aws_resource_validator.core.api_object import APIObject + + +@dataclass(frozen=True, slots=True) +class ValidationResult: + """Structured validation outcome containing diagnostics, bounds, and warnings.""" + + is_valid: bool + value: str + service_name: str + shape_name: str + pattern: str + length: int + min_length: int | None = None + max_length: int | None = None + errors: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + strict: bool = False + prefix_matched: bool = False + full_matched: bool = False + + def to_dict(self) -> dict[str, Any]: + """Serialize result to a dictionary for JSON output.""" + return asdict(self) + + +def diagnose_validation( + api_object: APIObject, + service_name: str, + value: str, + *, + strict: bool = False, +) -> ValidationResult: + """Perform detailed validation with diagnostic messages on failure. + + Checks: + 1. Minimum length constraint. + 2. Maximum length constraint. + 3. Botocore regex match (prefix-anchored via :func:`re.match`). + 4. End-to-end match (:func:`re.fullmatch`), enforced in strict mode or flagged as a warning. + """ + length = len(value) + errors: list[str] = [] + warnings: list[str] = [] + + # 1. Length constraints + if api_object.min_length is not None and length < api_object.min_length: + errors.append(f"Length {length} is shorter than minimum required length ({api_object.min_length}).") + + if api_object.max_length is not None and length > api_object.max_length: + errors.append(f"Length {length} exceeds maximum allowed length ({api_object.max_length}).") + + # 2. Regex matching + match = re.match(api_object.pattern, value) + prefix_matched = match is not None + full_matched = re.fullmatch(api_object.pattern, value) is not None + + if not prefix_matched: + errors.append(f"Value does not match required regex pattern: {api_object.pattern}") + elif not full_matched: + if strict: + matched_substr = match.group(0) if match else "" + errors.append( + f"Value only matches prefix '{matched_substr}' under botocore re.match; " + "strict mode requires full string match." + ) + else: + warnings.append( + "Value matches prefix under botocore's re.match semantics, " + "but does not match end-to-end (use --strict to enforce)." + ) + + is_valid = len(errors) == 0 + + return ValidationResult( + is_valid=is_valid, + value=value, + service_name=service_name, + shape_name=api_object.name, + pattern=api_object.pattern, + min_length=api_object.min_length, + max_length=api_object.max_length, + length=length, + errors=errors, + warnings=warnings, + strict=strict, + prefix_matched=prefix_matched, + full_matched=full_matched, + ) diff --git a/aws_resource_validator/cli/formatters.py b/aws_resource_validator/cli/formatters.py new file mode 100644 index 00000000..a78b6c55 --- /dev/null +++ b/aws_resource_validator/cli/formatters.py @@ -0,0 +1,201 @@ +"""Rich terminal formatters for validation, inspection, and synthetic generation.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from rich import box +from rich.console import Console +from rich.markup import escape +from rich.panel import Panel +from rich.table import Table +from rich.text import Text + +if TYPE_CHECKING: + from aws_resource_validator.cli.diagnostics import ValidationResult + from aws_resource_validator.core.api_object import APIObject + from aws_resource_validator.core.service import Service + + +def get_console() -> Console: + """Create a default Rich console.""" + return Console() + + +def print_validation_result(console: Console, result: ValidationResult) -> None: + """Render a formatted Rich panel displaying validation outcome and diagnostics.""" + if result.is_valid: + status_badge = "[bold white on green] PASS [/bold white on green]" + border_style = "green" + summary = ( + f"[bold green]Value satisfies AWS constraints for " + f"{escape(result.service_name)}.{escape(result.shape_name)}[/bold green]" + ) + else: + status_badge = "[bold white on red] FAIL [/bold white on red]" + border_style = "red" + summary = ( + f"[bold red]Value does NOT satisfy AWS constraints for " + f"{escape(result.service_name)}.{escape(result.shape_name)}[/bold red]" + ) + + grid = Table.grid(padding=(0, 1)) + grid.add_column(style="bold cyan", justify="right", width=14) + grid.add_column() + + grid.add_row("Status:", status_badge) + grid.add_row("Summary:", summary) + grid.add_row("Service:", f"[white]{escape(result.service_name)}[/white]") + grid.add_row("Shape:", f"[white]{escape(result.shape_name)}[/white]") + grid.add_row("Tested Value:", f"[bold white]{escape(result.value)}[/bold white]") + + # Length info + min_l = str(result.min_length) if result.min_length is not None else "0" + max_l = str(result.max_length) if result.max_length is not None else "unbounded" + grid.add_row("Length:", f"{result.length} (allowed: {min_l} - {max_l})") + grid.add_row("Regex Pattern:", f"[dim]{escape(result.pattern)}[/dim]") + + if result.errors: + error_lines = "\n".join(f"[bold red]• {escape(err)}[/bold red]" for err in result.errors) + grid.add_row("Errors:", error_lines) + + if result.warnings: + warn_lines = "\n".join(f"[bold yellow]• {escape(w)}[/bold yellow]" for w in result.warnings) + grid.add_row("Warnings:", warn_lines) + + panel = Panel( + grid, + title=f"[bold]Validation — {escape(result.service_name)}.{escape(result.shape_name)}[/bold]", + border_style=border_style, + expand=False, + ) + console.print(panel) + + +def print_inspect_shape(console: Console, service_name: str, obj: APIObject) -> None: + """Render a detailed panel showing shape metadata, constraints, and an example.""" + grid = Table.grid(padding=(0, 1)) + grid.add_column(style="bold cyan", justify="right", width=16) + grid.add_column() + + grid.add_row("Service:", f"[bold white]{escape(service_name)}[/bold white]") + grid.add_row("Shape:", f"[bold white]{escape(obj.name)}[/bold white]") + grid.add_row("Botocore Type:", f"[white]{escape(obj.type)}[/white]") + + min_l = str(obj.min_length) if obj.min_length is not None else "None" + max_l = str(obj.max_length) if obj.max_length is not None else "None" + grid.add_row("Min Length:", min_l) + grid.add_row("Max Length:", max_l) + grid.add_row("Regex Pattern:", f"[yellow]{escape(obj.pattern)}[/yellow]") + + try: + sample = obj.generate() + grid.add_row("Sample Value:", f"[bold green]{escape(sample)}[/bold green]") + except Exception as exc: + grid.add_row("Sample Value:", f"[dim italic]Generation unavailable: {escape(str(exc))}[/dim italic]") + + panel = Panel( + grid, + title=f"[bold]Inspect: {escape(service_name)}.{escape(obj.name)}[/bold]", + border_style="cyan", + expand=False, + ) + console.print(panel) + + +def print_inspect_service(console: Console, service: Service) -> None: + """Render a table of all patterned shapes available in a service.""" + if not service.api_objects: + console.print( + f"[yellow]Service [bold]{escape(service.service_name)}[/bold] " + "has no shapes with pattern constraints.[/yellow]" + ) + return + + table = Table( + title=f"Shapes in [bold cyan]{escape(service.service_name)}[/bold cyan] ({len(service.api_objects)} total)", + box=box.ROUNDED, + header_style="bold magenta", + ) + table.add_column("Shape Name", style="bold cyan") + table.add_column("Min", justify="right", style="green") + table.add_column("Max", justify="right", style="green") + table.add_column("Regex Pattern", style="yellow") + table.add_column("Sample Value", style="white") + + for obj in service.api_objects.values(): + min_l = str(obj.min_length) if obj.min_length is not None else "-" + max_l = str(obj.max_length) if obj.max_length is not None else "-" + pattern_display = obj.pattern if len(obj.pattern) <= 40 else obj.pattern[:37] + "..." + try: + sample = obj.generate() + if len(sample) > 30: + sample = sample[:27] + "..." + except Exception: + sample = "-" + table.add_row(obj.name, min_l, max_l, pattern_display, sample) + + console.print(table) + + +def print_generated_values( + console: Console, + service_name: str, + shape_name: str, + values: list[str], + *, + plain: bool = False, +) -> None: + """Output generated synthetic values.""" + if plain: + for val in values: + console.print(val) + return + + table = Table( + title=f"Synthetic [bold cyan]{escape(service_name)}.{escape(shape_name)}[/bold cyan] ({len(values)} generated)", + box=box.ROUNDED, + header_style="bold magenta", + ) + table.add_column("#", justify="right", style="dim") + table.add_column("Generated Value", style="bold green") + + for idx, val in enumerate(values, start=1): + table.add_row(str(idx), escape(val)) + + console.print(table) + + +def print_services_list( + console: Console, + services: list[tuple[str, int]], + total_count: int, +) -> None: + """Render a table of registered AWS services and their shape counts.""" + table = Table( + title=f"Registered AWS Services ({len(services)} shown of {total_count} total)", + box=box.ROUNDED, + header_style="bold magenta", + ) + table.add_column("Service", style="bold cyan") + table.add_column("Patterned Shapes", justify="right", style="green") + + for name, count in services: + table.add_row(name, str(count)) + + console.print(table) + + +def print_error( + console: Console, + message: str, + *, + suggestions: list[str] | None = None, +) -> None: + """Display a formatted red error badge with optional typo suggestions.""" + badge = Text(" ERROR ", style="bold white on red") + console.print(badge, f"[bold red]{escape(message)}[/bold red]") + + if suggestions: + suggest_str = ", ".join(f"[bold cyan]{escape(s)}[/bold cyan]" for s in suggestions) + console.print(f" [dim]Did you mean:[/dim] {suggest_str}?") diff --git a/aws_resource_validator/cli/resolver.py b/aws_resource_validator/cli/resolver.py new file mode 100644 index 00000000..282f4b2f --- /dev/null +++ b/aws_resource_validator/cli/resolver.py @@ -0,0 +1,101 @@ +"""Service and shape lookup utilities with typo suggestions and fuzzy matching.""" + +from __future__ import annotations + +import difflib +from functools import cache +from typing import TYPE_CHECKING + +from aws_resource_validator.core.naming import to_pascal_case + +if TYPE_CHECKING: + from aws_resource_validator.core.api_object import APIObject + from aws_resource_validator.core.registry import APIRegistry + from aws_resource_validator.core.service import Service + + +@cache +def get_registry() -> APIRegistry: + """Lazily load and cache the class registry from class_definitions. + + Deferring import preserves sub-50ms startup time for commands like ``--help``. + """ + from aws_resource_validator.class_definitions import class_registry + + return class_registry + + +def normalize_name(name: str) -> str: + """Strip hyphens, underscores, and lowercase for fault-tolerant comparison.""" + return name.lower().replace("-", "").replace("_", "") + + +def resolve_service(query: str) -> tuple[Service | None, list[str]]: + """Look up an AWS service in the registry by name. + + Supports exact PascalCase, snake_case, kebab-case, and case-insensitive queries. + Returns ``(service, [])`` on success, or ``(None, suggestions)`` on failure. + """ + registry = get_registry() + + # 1. Exact match + if query in registry: + return registry[query], [] + + # 2. PascalCase conversion (e.g. "s3-control" -> "S3Control", "lambda" -> "Lambda") + pascal = to_pascal_case(query) + if pascal in registry: + return registry[pascal], [] + + # 3. Normalized match across all registered services + query_norm = normalize_name(query) + for name, svc in registry.services.items(): + if normalize_name(name) == query_norm: + return svc, [] + + # 4. Fuzzy suggestions for typos + all_names = list(registry.services.keys()) + suggestions = difflib.get_close_matches(query, all_names, n=3, cutoff=0.5) + if not suggestions: + suggestions = difflib.get_close_matches(pascal, all_names, n=3, cutoff=0.5) + + return None, suggestions + + +def resolve_shape(service: Service, query: str) -> tuple[APIObject | None, list[str]]: + """Look up an API shape within a service. + + Supports exact PascalCase, snake_case aliases, and case-insensitive queries. + Returns ``(api_object, [])`` on success, or ``(None, suggestions)`` on failure. + """ + # 1. Exact index or alias (Service.__getitem__ handles PascalCase and snake_case) + try: + return service[query], [] + except KeyError: + pass + + # 2. Normalized match across available shapes + query_norm = normalize_name(query) + for name, obj in service.api_objects.items(): + if normalize_name(name) == query_norm: + return obj, [] + + # 3. Fuzzy suggestions + all_shapes = list(service.api_objects.keys()) + suggestions = difflib.get_close_matches(query, all_shapes, n=3, cutoff=0.5) + + return None, suggestions + + +def list_services(query: str | None = None) -> list[tuple[str, int]]: + """Return a sorted list of ``(service_name, shape_count)``, optionally filtered.""" + registry = get_registry() + items: list[tuple[str, int]] = [] + norm_query = query.lower() if query else None + + for name in sorted(registry.services.keys()): + if norm_query and norm_query not in name.lower(): + continue + items.append((name, len(registry[name]))) + + return items diff --git a/pyproject.toml b/pyproject.toml index 8db05297..0e8e6ff1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,9 +25,12 @@ dependencies = [ "exrex", "pydantic (>=2.13.5,<3.0)", "botocore", + "typer", + "rich", ] [project.scripts] +arv = "aws_resource_validator.cli.__main__:main" arv-generate = "aws_resource_validator.generator.__main__:main" [tool.poetry] @@ -261,6 +264,7 @@ ignore = ["PLR0913", "PLR0917", "PLR2004"] "tests/*" = ["PLR2004", "S101", "B018", "PLW0108", "PLC0415"] # Typer's documented API requires calling ``typer.Option`` in argument defaults. "aws_resource_validator/generator/cli.py" = ["B008"] +"aws_resource_validator/cli/*.py" = ["B008", "PLC0415"] # Lazy import is the whole point of this shim — see the module docstring. "aws_resource_validator/generator/__main__.py" = ["PLC0415"] # Release scripts: tolerate inline imports (for optional playwright dep), @@ -273,7 +277,7 @@ ignore = ["PLR0913", "PLR0917", "PLR2004"] [tool.mypy] python_version = "3.11" strict = true -files = ["aws_resource_validator/core", "aws_resource_validator/generator"] +files = ["aws_resource_validator/core", "aws_resource_validator/generator", "aws_resource_validator/cli"] [[tool.mypy.overrides]] module = ["exrex.*", "responses.*", "botocore.*", "requests.*"] diff --git a/tests/unit/cli/__init__.py b/tests/unit/cli/__init__.py new file mode 100644 index 00000000..1df47c90 --- /dev/null +++ b/tests/unit/cli/__init__.py @@ -0,0 +1 @@ +"""Tests for the ``aws_resource_validator.cli`` package.""" diff --git a/tests/unit/cli/test_commands.py b/tests/unit/cli/test_commands.py new file mode 100644 index 00000000..314c415c --- /dev/null +++ b/tests/unit/cli/test_commands.py @@ -0,0 +1,137 @@ +"""Integration / smoke tests for CLI subcommands via Typer CliRunner.""" + +from __future__ import annotations + +import json + +from typer.testing import CliRunner + +from aws_resource_validator import __version__ +from aws_resource_validator.cli.app import app + +runner = CliRunner() + + +def test_version_command() -> None: + result = runner.invoke(app, ["version"]) + assert result.exit_code == 0 + assert __version__ in result.stdout + assert "Python:" in result.stdout + + +def test_validate_success() -> None: + result = runner.invoke(app, ["validate", "lambda", "FunctionName", "my-function-name"]) + assert result.exit_code == 0 + assert "PASS" in result.stdout + assert "Lambda.FunctionName" in result.stdout + + +def test_validate_success_json() -> None: + result = runner.invoke(app, ["validate", "lambda", "FunctionName", "my-function-name", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.stdout) + assert data["is_valid"] is True + assert data["service_name"] == "Lambda" + assert data["shape_name"] == "FunctionName" + + +def test_validate_failure() -> None: + result = runner.invoke(app, ["validate", "lambda", "FunctionName", "invalid func! name", "--strict"]) + assert result.exit_code == 1 + assert "FAIL" in result.stdout + + +def test_validate_failure_json() -> None: + result = runner.invoke( + app, + ["validate", "lambda", "FunctionName", "invalid func! name", "--strict", "--json"], + ) + assert result.exit_code == 1 + data = json.loads(result.stdout) + assert data["is_valid"] is False + assert len(data["errors"]) > 0 + + +def test_validate_unknown_service() -> None: + result = runner.invoke(app, ["validate", "nonexistentservice", "SomeShape", "val"]) + assert result.exit_code == 2 + assert "not found" in result.stdout + + +def test_validate_unknown_shape() -> None: + result = runner.invoke(app, ["validate", "lambda", "NonExistentShape", "val"]) + assert result.exit_code == 2 + assert "not found" in result.stdout + + +def test_inspect_shape() -> None: + result = runner.invoke(app, ["inspect", "lambda", "FunctionName"]) + assert result.exit_code == 0 + assert "Inspect: Lambda.FunctionName" in result.stdout + assert "Regex Pattern:" in result.stdout + + +def test_inspect_shape_json() -> None: + result = runner.invoke(app, ["inspect", "lambda", "FunctionName", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.stdout) + assert data["service"] == "Lambda" + assert data["shape"] == "FunctionName" + assert "pattern" in data + + +def test_inspect_service_all_shapes() -> None: + result = runner.invoke(app, ["inspect", "lambda"]) + assert result.exit_code == 0 + assert "Shapes in Lambda" in result.stdout + assert "FunctionName" in result.stdout + + +def test_inspect_service_json() -> None: + result = runner.invoke(app, ["inspect", "lambda", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.stdout) + assert data["service"] == "Lambda" + assert len(data["shapes"]) > 0 + + +def test_generate_default() -> None: + result = runner.invoke(app, ["generate", "lambda", "FunctionName"]) + assert result.exit_code == 0 + assert "Synthetic" in result.stdout + assert "Lambda.FunctionName" in result.stdout + assert "Generated Value" in result.stdout + + +def test_generate_plain() -> None: + result = runner.invoke(app, ["generate", "lambda", "FunctionName", "--count", "3", "--plain"]) + assert result.exit_code == 0 + lines = [line.strip() for line in result.stdout.strip().splitlines() if line.strip()] + assert len(lines) == 3 + + +def test_generate_json() -> None: + result = runner.invoke(app, ["generate", "lambda", "FunctionName", "--count", "2", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.stdout) + assert data["service"] == "Lambda" + assert data["shape"] == "FunctionName" + assert len(data["values"]) == 2 + + +def test_list_services() -> None: + result = runner.invoke(app, ["list"]) + assert result.exit_code == 0 + assert "Registered AWS Services" in result.stdout + + +def test_list_services_search() -> None: + result = runner.invoke(app, ["list", "--search", "dynamodb"]) + assert result.exit_code == 0 + assert "Dynamodb" in result.stdout + + +def test_list_delegates_to_inspect_when_service_provided() -> None: + result = runner.invoke(app, ["list", "lambda"]) + assert result.exit_code == 0 + assert "Shapes in Lambda" in result.stdout diff --git a/tests/unit/cli/test_diagnostics.py b/tests/unit/cli/test_diagnostics.py new file mode 100644 index 00000000..56f8f7c0 --- /dev/null +++ b/tests/unit/cli/test_diagnostics.py @@ -0,0 +1,72 @@ +"""Unit tests for validation diagnostics and strict mode enforcement.""" + +from __future__ import annotations + +from aws_resource_validator.cli.diagnostics import diagnose_validation +from aws_resource_validator.core.api_object import APIObject + + +def test_diagnose_valid_value() -> None: + obj = APIObject(name="Identifier", type="string", pattern=r"[a-z0-9-]+", min_length=3, max_length=20) + result = diagnose_validation(obj, "TestService", "valid-123") + + assert result.is_valid is True + assert result.errors == [] + assert result.warnings == [] + assert result.length == 9 + assert result.min_length == 3 + assert result.max_length == 20 + assert result.full_matched is True + + +def test_diagnose_too_short() -> None: + obj = APIObject(name="Identifier", type="string", pattern=r"[a-z0-9-]+", min_length=5, max_length=20) + result = diagnose_validation(obj, "TestService", "ab") + + assert result.is_valid is False + assert any("shorter than minimum" in err for err in result.errors) + + +def test_diagnose_too_long() -> None: + obj = APIObject(name="Identifier", type="string", pattern=r"[a-z0-9-]+", min_length=1, max_length=5) + result = diagnose_validation(obj, "TestService", "toolongvalue") + + assert result.is_valid is False + assert any("exceeds maximum" in err for err in result.errors) + + +def test_diagnose_regex_mismatch() -> None: + obj = APIObject(name="Identifier", type="string", pattern=r"[0-9]+", min_length=1, max_length=10) + result = diagnose_validation(obj, "TestService", "letters_only") + + assert result.is_valid is False + assert any("does not match required regex pattern" in err for err in result.errors) + + +def test_diagnose_prefix_vs_strict_mode() -> None: + # Pattern without $ / \Z: matches prefix under re.match + obj = APIObject(name="Identifier", type="string", pattern=r"[a-z]+", min_length=1, max_length=50) + + # In non-strict mode: valid under botocore semantics, with a warning + loose_res = diagnose_validation(obj, "TestService", "prefix_with_invalid_!@#", strict=False) + assert loose_res.is_valid is True + assert len(loose_res.warnings) == 1 + assert "matches prefix" in loose_res.warnings[0] + + # In strict mode: rejected + strict_res = diagnose_validation(obj, "TestService", "prefix_with_invalid_!@#", strict=True) + assert strict_res.is_valid is False + assert any("strict mode requires full string match" in err for err in strict_res.errors) + + +def test_validation_result_to_dict() -> None: + obj = APIObject(name="Id", type="string", pattern=r".+", min_length=1, max_length=10) + result = diagnose_validation(obj, "Svc", "val") + d = result.to_dict() + + assert d["is_valid"] is True + assert d["service_name"] == "Svc" + assert d["shape_name"] == "Id" + assert d["pattern"] == ".+" + assert d["value"] == "val" + assert d["length"] == 3 diff --git a/tests/unit/cli/test_resolver.py b/tests/unit/cli/test_resolver.py new file mode 100644 index 00000000..5fad0825 --- /dev/null +++ b/tests/unit/cli/test_resolver.py @@ -0,0 +1,95 @@ +"""Unit tests for service and shape resolution and fuzzy suggestions.""" + +from __future__ import annotations + +from aws_resource_validator.cli.resolver import ( + list_services, + normalize_name, + resolve_service, + resolve_shape, +) +from aws_resource_validator.core.api_object import APIObject +from aws_resource_validator.core.service import Service + + +def test_normalize_name() -> None: + assert normalize_name("S3-Control") == "s3control" + assert normalize_name("Cognito_Idp") == "cognitoidp" + assert normalize_name("lambda") == "lambda" + + +def test_resolve_service_exact_and_case_insensitive() -> None: + svc, suggestions = resolve_service("Lambda") + assert svc is not None + assert svc.service_name == "Lambda" + assert suggestions == [] + + svc2, _ = resolve_service("lambda") + assert svc2 is not None + assert svc2.service_name == "Lambda" + + svc3, _ = resolve_service("LAMBDA") + assert svc3 is not None + assert svc3.service_name == "Lambda" + + +def test_resolve_service_with_hyphens_or_underscores() -> None: + svc, _ = resolve_service("s3-control") + assert svc is not None + assert svc.service_name == "S3control" + + svc2, _ = resolve_service("s3_control") + assert svc2 is not None + assert svc2.service_name == "S3control" + + +def test_resolve_service_missing_returns_suggestions() -> None: + svc, suggestions = resolve_service("lamda") + assert svc is None + assert "Lambda" in suggestions + + +def test_resolve_shape_exact_and_case_insensitive() -> None: + sample_svc = Service( + "Demo", + [ + APIObject(name="ResourceName", type="string", pattern="[a-z]+"), + APIObject(name="ClusterId", type="string", pattern="[0-9]+"), + ], + ) + + obj, suggestions = resolve_shape(sample_svc, "ResourceName") + assert obj is not None + assert obj.name == "ResourceName" + assert suggestions == [] + + # snake_case alias + obj2, _ = resolve_shape(sample_svc, "resource_name") + assert obj2 is not None + assert obj2.name == "ResourceName" + + # case-insensitive + obj3, _ = resolve_shape(sample_svc, "resourcename") + assert obj3 is not None + assert obj3.name == "ResourceName" + + +def test_resolve_shape_missing_returns_suggestions() -> None: + sample_svc = Service( + "Demo", + [ + APIObject(name="ResourceName", type="string", pattern="[a-z]+"), + ], + ) + obj, suggestions = resolve_shape(sample_svc, "ResourceNam") + assert obj is None + assert "ResourceName" in suggestions + + +def test_list_services_filtering() -> None: + services = list_services() + assert len(services) > 100 + + filtered = list_services(query="lambda") + assert any(s == "Lambda" for s, _ in filtered) + assert all("lambda" in s.lower() for s, _ in filtered)