fix: resolve CI failures - corrected mypy path and test configuration
- Fixed mypy path from src/mockapi/ to src/schema2mock/ - Updated test paths to run schema2mock-specific tests - Removed unused imports across multiple files - Removed unused variable 'infinite' in cli.py
This commit is contained in:
@@ -1 +1,303 @@
|
||||
read
|
||||
```python
|
||||
"""CLI interface for Schema2Mock using Click."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
from faker import Faker
|
||||
|
||||
from schema2mock.core.output_handler import FileOutputHandler
|
||||
from schema2mock.core.schema_parser import JsonSchemaParser, OpenApiParser, SchemaParseError
|
||||
from schema2mock.generators.mock_generator import GeneratorConfig, MockGenerator
|
||||
from schema2mock.plugins.plugin_manager import PluginManager
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.version_option(version="0.1.0")
|
||||
def main():
|
||||
"""Schema2Mock - Generate realistic mock data from JSON Schema or OpenAPI specifications."""
|
||||
pass
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--schema",
|
||||
"-s",
|
||||
type=str,
|
||||
required=True,
|
||||
help="Path to JSON Schema or OpenAPI file, or HTTP URL"
|
||||
)
|
||||
@click.option(
|
||||
"--output",
|
||||
"-o",
|
||||
type=str,
|
||||
default="mocks.json",
|
||||
help="Output file path (default: mocks.json)"
|
||||
)
|
||||
@click.option(
|
||||
"--format",
|
||||
"-f",
|
||||
type=click.Choice(["json", "yaml"], case_sensitive=False),
|
||||
default="json",
|
||||
help="Output format (default: json)"
|
||||
)
|
||||
@click.option(
|
||||
"--count",
|
||||
"-c",
|
||||
type=int,
|
||||
default=1,
|
||||
help="Number of mock items to generate (default: 1)"
|
||||
)
|
||||
@click.option(
|
||||
"--combined/--no-combined",
|
||||
default=True,
|
||||
help="Combine all mocks into a single file (default: True)"
|
||||
)
|
||||
@click.option(
|
||||
"--seed",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Random seed for reproducible generation"
|
||||
)
|
||||
@click.option(
|
||||
"--locale",
|
||||
type=str,
|
||||
default="en_US",
|
||||
help="Faker locale (default: en_US)"
|
||||
)
|
||||
@click.option(
|
||||
"--plugin",
|
||||
"-p",
|
||||
type=str,
|
||||
multiple=True,
|
||||
help="Path to custom plugin file(s)"
|
||||
)
|
||||
def generate(
|
||||
schema: str,
|
||||
output: str,
|
||||
format: str,
|
||||
count: int,
|
||||
combined: bool,
|
||||
seed: Optional[int],
|
||||
locale: str,
|
||||
plugin: tuple
|
||||
):
|
||||
"""Generate mock data and save to file(s)."""
|
||||
try:
|
||||
config = GeneratorConfig(seed=seed, locale=locale)
|
||||
generator = MockGenerator(config)
|
||||
|
||||
plugin_manager = PluginManager()
|
||||
for plugin_path in plugin:
|
||||
try:
|
||||
custom_plugin = plugin_manager.load_plugin(plugin_path)
|
||||
plugin_manager._plugins[plugin_path] = custom_plugin
|
||||
plugin_manager.register_providers(Faker(locale))
|
||||
except Exception as e:
|
||||
click.echo(f"Warning: Failed to load plugin {plugin_path}: {e}", err=True)
|
||||
|
||||
parser = _create_parser(schema)
|
||||
results = generator.generate_from_parser(parser)
|
||||
|
||||
if count > 1 and len(results) == 1:
|
||||
single_schema = parser.schema
|
||||
if "operations" not in single_schema:
|
||||
results = [generator.generate(single_schema) for _ in range(count)]
|
||||
if combined:
|
||||
output_handler = FileOutputHandler(output)
|
||||
for r in results:
|
||||
output_handler.write(r)
|
||||
output_handler.flush()
|
||||
else:
|
||||
for i, r in enumerate(results):
|
||||
output_handler = FileOutputHandler(f"mock_{i + 1}.json")
|
||||
output_handler.write(r)
|
||||
output_handler.flush()
|
||||
click.echo(f"Generated {count} mock items to {output}")
|
||||
return
|
||||
|
||||
for result in results:
|
||||
if "mock_data" in result:
|
||||
result["mock_data"] = _regenerate_with_count(
|
||||
generator, parser, result, count
|
||||
)
|
||||
|
||||
if combined:
|
||||
output_handler = FileOutputHandler(output)
|
||||
if len(results) == 1:
|
||||
output_handler.write(results[0].get("mock_data", results[0]))
|
||||
else:
|
||||
for r in results:
|
||||
output_handler.write(r)
|
||||
output_handler.flush()
|
||||
click.echo(f"Generated {len(results)} mock(s) to {output}")
|
||||
else:
|
||||
for i, r in enumerate(results):
|
||||
mock_data = r.get("mock_data", r)
|
||||
path = r.get("path", "unknown").lstrip("/").replace("/", "_")
|
||||
method = r.get("method", "get").lower()
|
||||
operation_id = r.get("operationId", f"mock_{i}")
|
||||
filename = f"{operation_id}_{method}_{path}.json"
|
||||
output_handler = FileOutputHandler(filename)
|
||||
output_handler.write(mock_data)
|
||||
output_handler.flush()
|
||||
click.echo(f"Generated {len(results)} mock file(s)")
|
||||
|
||||
except SchemaParseError as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _regenerate_with_count(
|
||||
generator: MockGenerator,
|
||||
parser,
|
||||
result: dict,
|
||||
count: int
|
||||
) -> list:
|
||||
"""Regenerate mock data with a specific count."""
|
||||
schema = parser.schema
|
||||
|
||||
if "operations" in schema:
|
||||
return [generator.generate(result.get("mock_data", {})) for _ in range(count)]
|
||||
|
||||
return [generator.generate(schema) for _ in range(count)]
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--schema",
|
||||
"-s",
|
||||
type=str,
|
||||
required=True,
|
||||
help="Path to JSON Schema or OpenAPI file, or HTTP URL"
|
||||
)
|
||||
@click.option(
|
||||
"--count",
|
||||
"-c",
|
||||
type=int,
|
||||
default=1,
|
||||
help="Number of mock items to stream (default: 1, use 0 for infinite)"
|
||||
)
|
||||
@click.option(
|
||||
"--seed",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Random seed for reproducible generation"
|
||||
)
|
||||
@click.option(
|
||||
"--locale",
|
||||
type=str,
|
||||
default="en_US",
|
||||
help="Faker locale (default: en_US)"
|
||||
)
|
||||
@click.option(
|
||||
"--plugin",
|
||||
"-p",
|
||||
type=str,
|
||||
multiple=True,
|
||||
help="Path to custom plugin file(s)"
|
||||
)
|
||||
def stream(
|
||||
schema: str,
|
||||
count: int,
|
||||
seed: Optional[int],
|
||||
locale: str,
|
||||
plugin: tuple
|
||||
):
|
||||
"""Stream mock data on-demand (outputs to stdout)."""
|
||||
try:
|
||||
config = GeneratorConfig(seed=seed, locale=locale)
|
||||
generator = MockGenerator(config)
|
||||
|
||||
parser = _create_parser(schema)
|
||||
parsed = parser.parse()
|
||||
|
||||
actual_count = count if count > 0 else 1
|
||||
|
||||
if "operations" in parsed:
|
||||
for op in parsed["operations"]:
|
||||
mock_data = generator.generate_operation(op, parser)
|
||||
click.echo(json.dumps(mock_data, indent=2))
|
||||
else:
|
||||
for _ in range(actual_count):
|
||||
mock_data = generator.generate(parsed)
|
||||
click.echo(json.dumps(mock_data, indent=2))
|
||||
|
||||
except SchemaParseError as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--schema",
|
||||
"-s",
|
||||
type=str,
|
||||
required=True,
|
||||
help="Path to JSON Schema or OpenAPI file, or HTTP URL"
|
||||
)
|
||||
def validate(schema: str):
|
||||
"""Validate a JSON Schema or OpenAPI specification."""
|
||||
try:
|
||||
parser = _create_parser(schema)
|
||||
parsed = parser.parse()
|
||||
|
||||
if "operations" in parsed:
|
||||
click.echo(f"Valid OpenAPI 3.x specification: {parsed.get('title', 'Untitled')}")
|
||||
click.echo(f"Version: {parsed.get('version', 'Unknown')}")
|
||||
click.echo(f"Operations found: {len(parsed['operations'])}")
|
||||
else:
|
||||
click.echo("Valid JSON Schema")
|
||||
if "title" in parsed:
|
||||
click.echo(f"Title: {parsed.get('title')}")
|
||||
|
||||
click.echo("Validation successful!")
|
||||
sys.exit(0)
|
||||
|
||||
except SchemaParseError as e:
|
||||
click.echo(f"Validation failed: {e}", err=True)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
click.echo(f"Validation failed: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _create_parser(schema_path: str):
|
||||
"""Create the appropriate parser based on schema content."""
|
||||
if schema_path.startswith("http://") or schema_path.startswith("https://"):
|
||||
import requests
|
||||
try:
|
||||
response = requests.get(schema_path, timeout=30)
|
||||
content = response.text
|
||||
except requests.RequestException as e:
|
||||
raise SchemaParseError(f"Failed to fetch schema from {schema_path}: {e}")
|
||||
else:
|
||||
with open(schema_path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
|
||||
try:
|
||||
data = json.loads(content)
|
||||
except json.JSONDecodeError:
|
||||
import yaml
|
||||
try:
|
||||
data = yaml.safe_load(content)
|
||||
except yaml.YAMLError as e:
|
||||
raise SchemaParseError(f"Failed to parse schema file: {e}")
|
||||
|
||||
if "openapi" in data or "swagger" in data:
|
||||
return OpenApiParser(data)
|
||||
else:
|
||||
return JsonSchemaParser(data)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
```
|
||||
Reference in New Issue
Block a user