309 lines
9.2 KiB
Python
309 lines
9.2 KiB
Python
"""Integration tests for JSON to OpenAPI Generator."""
|
|
import pytest
|
|
import json
|
|
import yaml
|
|
from click.testing import CliRunner
|
|
from pathlib import Path
|
|
from json_to_openapi.cli import main
|
|
from json_to_openapi.schema_generator import OpenAPIGenerator
|
|
|
|
|
|
class TestIntegration:
|
|
"""Integration test cases."""
|
|
|
|
@pytest.fixture
|
|
def runner(self):
|
|
return CliRunner()
|
|
|
|
@pytest.fixture
|
|
def user_schema_file(self, tmp_path):
|
|
data = {
|
|
"users": [
|
|
{
|
|
"id": 1,
|
|
"name": "Alice",
|
|
"email": "alice@example.com",
|
|
"age": 30,
|
|
"active": True,
|
|
"created_at": "2024-01-15T10:30:00Z",
|
|
"tags": ["admin", "user"],
|
|
"metadata": {
|
|
"department": "Engineering",
|
|
"level": 5
|
|
}
|
|
},
|
|
{
|
|
"id": 2,
|
|
"name": "Bob",
|
|
"email": "bob@example.com",
|
|
"age": 25,
|
|
"active": False,
|
|
"created_at": "2024-02-20T14:00:00Z",
|
|
"tags": ["user"],
|
|
"metadata": {
|
|
"department": "Sales",
|
|
"level": 3
|
|
}
|
|
}
|
|
],
|
|
"total": 2
|
|
}
|
|
file_path = tmp_path / "users.json"
|
|
file_path.write_text(json.dumps(data))
|
|
return file_path
|
|
|
|
@pytest.fixture
|
|
def product_schema_file(self, tmp_path):
|
|
data = {
|
|
"products": [
|
|
{
|
|
"id": 101,
|
|
"name": "Laptop",
|
|
"price": 999.99,
|
|
"in_stock": True,
|
|
"category": "Electronics",
|
|
"variants": [
|
|
{"color": "black", "sku": "LAP-BLK-001"},
|
|
{"color": "silver", "sku": "LAP-SLV-001"}
|
|
]
|
|
},
|
|
{
|
|
"id": 102,
|
|
"name": "Mouse",
|
|
"price": 29.99,
|
|
"in_stock": True,
|
|
"category": "Electronics",
|
|
"variants": [
|
|
{"color": "white", "sku": "MSE-WHT-001"}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
file_path = tmp_path / "products.json"
|
|
file_path.write_text(json.dumps(data))
|
|
return file_path
|
|
|
|
def test_end_to_end_user_schema(self, runner, user_schema_file, tmp_path):
|
|
"""Test end-to-end conversion of user schema."""
|
|
output_file = tmp_path / "users_openapi.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(user_schema_file),
|
|
"-o", str(output_file),
|
|
"-t", "Users API",
|
|
"-v", "1.0.0",
|
|
"-d", "API for managing users"
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert output_file.exists()
|
|
|
|
content = output_file.read_text()
|
|
spec = yaml.safe_load(content)
|
|
|
|
assert spec["openapi"] == "3.0.3"
|
|
assert spec["info"]["title"] == "Users API"
|
|
assert spec["info"]["version"] == "1.0.0"
|
|
assert spec["info"]["description"] == "API for managing users"
|
|
|
|
def test_end_to_end_product_schema(self, runner, product_schema_file, tmp_path):
|
|
"""Test end-to-end conversion of product schema."""
|
|
output_file = tmp_path / "products_openapi.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(product_schema_file),
|
|
"-o", str(output_file),
|
|
"-t", "Products API"
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert output_file.exists()
|
|
|
|
content = output_file.read_text()
|
|
spec = yaml.safe_load(content)
|
|
|
|
assert spec["openapi"] == "3.0.3"
|
|
assert spec["info"]["title"] == "Products API"
|
|
|
|
def test_batch_processing(self, runner, user_schema_file, product_schema_file, tmp_path):
|
|
"""Test batch processing of multiple JSON files."""
|
|
output_dir = tmp_path / "output"
|
|
output_dir.mkdir()
|
|
|
|
result = runner.invoke(main, [
|
|
"batch",
|
|
str(tmp_path / "*.json"),
|
|
"-o", str(output_dir),
|
|
"-f", "yaml"
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Successfully processed 2 files" in result.output
|
|
|
|
assert (output_dir / "users.yaml").exists()
|
|
assert (output_dir / "products.yaml").exists()
|
|
|
|
def test_combined_batch_processing(self, runner, user_schema_file, product_schema_file, tmp_path):
|
|
"""Test combined batch processing."""
|
|
output_dir = tmp_path / "combined_output"
|
|
output_dir.mkdir()
|
|
|
|
result = runner.invoke(main, [
|
|
"batch",
|
|
str(tmp_path / "*.json"),
|
|
"-o", str(output_dir),
|
|
"-c",
|
|
"-f", "yaml"
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Generated combined spec" in result.output or "combined.yaml" in result.output
|
|
|
|
def test_spec_validity(self, runner, user_schema_file, tmp_path):
|
|
"""Test that generated spec is valid OpenAPI 3.0.3."""
|
|
output_file = tmp_path / "valid_spec.yaml"
|
|
|
|
runner.invoke(main, [
|
|
"convert",
|
|
str(user_schema_file),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
result = runner.invoke(main, ["validate", str(output_file)])
|
|
assert result.exit_code == 0
|
|
assert "Validation passed" in result.output
|
|
|
|
def test_json_output_format(self, runner, user_schema_file, tmp_path):
|
|
"""Test JSON output format."""
|
|
output_file = tmp_path / "spec.json"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(user_schema_file),
|
|
"-o", str(output_file),
|
|
"-f", "json"
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
content = output_file.read_text()
|
|
spec = json.loads(content)
|
|
|
|
assert spec["openapi"] == "3.0.3"
|
|
assert "info" in spec
|
|
assert "paths" in spec
|
|
|
|
def test_custom_endpoint_path(self, runner, user_schema_file, tmp_path):
|
|
"""Test custom endpoint path."""
|
|
output_file = tmp_path / "custom_path.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(user_schema_file),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
content = output_file.read_text()
|
|
spec = yaml.safe_load(content)
|
|
|
|
assert "/" in spec["paths"] or "/users" in spec["paths"]
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Test edge cases and error handling."""
|
|
|
|
@pytest.fixture
|
|
def runner(self):
|
|
return CliRunner()
|
|
|
|
def test_empty_object(self, runner, tmp_path):
|
|
"""Test handling of empty object."""
|
|
data = {}
|
|
file_path = tmp_path / "empty.json"
|
|
file_path.write_text(json.dumps(data))
|
|
|
|
output_file = tmp_path / "output.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(file_path),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
assert output_file.exists()
|
|
|
|
def test_empty_array(self, runner, tmp_path):
|
|
"""Test handling of empty array."""
|
|
data = []
|
|
file_path = tmp_path / "empty_array.json"
|
|
file_path.write_text(json.dumps(data))
|
|
|
|
output_file = tmp_path / "output.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(file_path),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_nested_deep_structure(self, runner, tmp_path):
|
|
"""Test handling of deeply nested structures."""
|
|
data = {"level1": {"level2": {"level3": {"level4": {"level5": "deep"}}}}}
|
|
file_path = tmp_path / "deep.json"
|
|
file_path.write_text(json.dumps(data))
|
|
|
|
output_file = tmp_path / "output.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(file_path),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_special_characters_in_strings(self, runner, tmp_path):
|
|
"""Test handling of special characters."""
|
|
data = {
|
|
"message": "Hello \"World\"!",
|
|
"path": "/api/v1/users",
|
|
"email": "user@example.com"
|
|
}
|
|
file_path = tmp_path / "special.json"
|
|
file_path.write_text(json.dumps(data))
|
|
|
|
output_file = tmp_path / "output.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(file_path),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_unicode_characters(self, runner, tmp_path):
|
|
"""Test handling of unicode characters."""
|
|
data = {
|
|
"name": "日本語 Test",
|
|
"emoji": "🚀",
|
|
"chinese": "你好"
|
|
}
|
|
file_path = tmp_path / "unicode.json"
|
|
file_path.write_text(json.dumps(data))
|
|
|
|
output_file = tmp_path / "output.yaml"
|
|
|
|
result = runner.invoke(main, [
|
|
"convert",
|
|
str(file_path),
|
|
"-o", str(output_file)
|
|
])
|
|
|
|
assert result.exit_code == 0
|