This commit is contained in:
@@ -1,482 +1,225 @@
|
||||
import json
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.core.parser import OpenAPIParser, ParseError, SpecValidationError
|
||||
from src.core.models import OpenAPISpec
|
||||
from src.core.models import Schema
|
||||
from src.core.parser import OpenAPIParser
|
||||
|
||||
|
||||
class TestOpenAPIParser:
|
||||
"""Test cases for OpenAPIParser."""
|
||||
|
||||
def test_parse_valid_json_spec(self):
|
||||
"""Test parsing a valid JSON OpenAPI spec."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse()
|
||||
|
||||
assert isinstance(result, OpenAPISpec)
|
||||
assert result.openapi == "3.0.0"
|
||||
assert result.info.title == "Test API"
|
||||
assert result.info.version == "1.0.0"
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_valid_yaml_spec(self):
|
||||
"""Test parsing a valid YAML OpenAPI spec."""
|
||||
spec_content = """
|
||||
openapi: 3.0.0
|
||||
info:
|
||||
title: Test API
|
||||
version: 1.0.0
|
||||
paths: {}
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write(spec_content)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse()
|
||||
|
||||
assert isinstance(result, OpenAPISpec)
|
||||
assert result.openapi == "3.0.0"
|
||||
assert result.info.title == "Test API"
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_missing_file(self):
|
||||
"""Test parsing a non-existent file."""
|
||||
parser = OpenAPIParser("nonexistent.json")
|
||||
|
||||
with pytest.raises(ParseError) as exc_info:
|
||||
parser.parse()
|
||||
|
||||
assert "not found" in str(exc_info.value).lower()
|
||||
|
||||
def test_parse_invalid_json(self):
|
||||
"""Test parsing invalid JSON file."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
f.write("{ invalid json }")
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
|
||||
with pytest.raises(ParseError):
|
||||
parser.parse()
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_invalid_yaml(self):
|
||||
"""Test parsing invalid YAML file."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write(" - invalid: yaml: content:")
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
|
||||
with pytest.raises(ParseError):
|
||||
parser.parse()
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_unsupported_format(self):
|
||||
"""Test parsing unsupported file format."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
|
||||
f.write("test content")
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
|
||||
with pytest.raises(ParseError) as exc_info:
|
||||
parser.parse()
|
||||
|
||||
assert "unsupported" in str(exc_info.value).lower()
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_spec_with_endpoints(self):
|
||||
"""Test parsing spec with endpoints."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Pet Store API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {
|
||||
"/pets": {
|
||||
"get": {
|
||||
"summary": "List pets",
|
||||
"description": "Get all pets",
|
||||
"responses": {
|
||||
"200": {"description": "Success"}
|
||||
}
|
||||
VALID_OPENAPI_SPEC = {
|
||||
"openapi": "3.0.3",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0",
|
||||
"description": "A test API"
|
||||
},
|
||||
"servers": [
|
||||
{"url": "https://api.example.com/v1", "description": "Production"}
|
||||
],
|
||||
"paths": {
|
||||
"/users": {
|
||||
"get": {
|
||||
"summary": "List users",
|
||||
"description": "Get a list of all users",
|
||||
"tags": ["Users"],
|
||||
"parameters": [
|
||||
{
|
||||
"name": "limit", "in": "query", "schema": {"type": "integer"},
|
||||
"required": False
|
||||
},
|
||||
"post": {
|
||||
"summary": "Create pet",
|
||||
"responses": {
|
||||
"201": {"description": "Created"}
|
||||
{
|
||||
"name": "offset", "in": "query", "schema": {"type": "integer"},
|
||||
"required": False
|
||||
}
|
||||
],
|
||||
"responses": {
|
||||
"200": {"description": "Success"},
|
||||
"400": {"description": "Bad Request"}
|
||||
}
|
||||
},
|
||||
"post": {
|
||||
"summary": "Create user",
|
||||
"description": "Create a new user",
|
||||
"tags": ["Users"],
|
||||
"requestBody": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"email": {"type": "string", "format": "email"}
|
||||
},
|
||||
"required": ["name", "email"]
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/pets/{petId}": {
|
||||
"get": {
|
||||
"summary": "Get pet",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "petId",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"}
|
||||
}
|
||||
],
|
||||
"responses": {
|
||||
"200": {"description": "Success"},
|
||||
"404": {"description": "Not found"}
|
||||
}
|
||||
}
|
||||
"responses": {
|
||||
"201": {"description": "Created"}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/users/{id}": {
|
||||
"get": {
|
||||
"summary": "Get user",
|
||||
"description": "Get a user by ID",
|
||||
"tags": ["Users"],
|
||||
"parameters": [
|
||||
{"name": "id", "in": "path", "required": True, "schema": {"type": "string"}}
|
||||
],
|
||||
"responses": {
|
||||
"200": {"description": "Success"},
|
||||
"404": {"description": "Not Found"}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse()
|
||||
|
||||
endpoints = result.get_endpoints()
|
||||
assert len(endpoints) == 3
|
||||
|
||||
endpoint_paths = [e.path for e in endpoints]
|
||||
assert "/pets" in endpoint_paths
|
||||
assert "/pets/{petId}" in endpoint_paths
|
||||
|
||||
endpoint_methods = [e.method for e in endpoints]
|
||||
assert "GET" in endpoint_methods
|
||||
assert "POST" in endpoint_methods
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_spec_with_schemas(self):
|
||||
"""Test parsing spec with schema definitions."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"User": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "string"},
|
||||
"name": {"type": "string"},
|
||||
"email": {"type": "string", "format": "email"}
|
||||
},
|
||||
"required": ["id", "name", "email"]
|
||||
},
|
||||
"paths": {},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"Pet": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "string"},
|
||||
"name": {"type": "string"}
|
||||
}
|
||||
},
|
||||
"User": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "integer"},
|
||||
"username": {"type": "string"},
|
||||
"email": {"type": "string", "format": "email"}
|
||||
}
|
||||
}
|
||||
"Error": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"code": {"type": "integer"},
|
||||
"message": {"type": "string"}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse()
|
||||
class TestOpenAPIParser:
|
||||
def test_parse_valid_spec(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
assert spec.info.title == "Test API"
|
||||
assert spec.info.version == "1.0.0"
|
||||
assert spec.openapi == "3.0.3"
|
||||
|
||||
schemas = result.get_schemas()
|
||||
assert "Pet" in schemas
|
||||
assert "User" in schemas
|
||||
def test_parse_paths(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
assert "/users" in spec.paths
|
||||
assert "/users/{id}" in spec.paths
|
||||
|
||||
Path(f.name).unlink()
|
||||
def test_parse_operations(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
users_path = spec.paths["/users"]
|
||||
assert users_path.get is not None
|
||||
assert users_path.post is not None
|
||||
assert users_path.get.summary == "List users"
|
||||
|
||||
def test_parse_spec_with_tags(self):
|
||||
"""Test parsing spec with tag definitions."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
def test_parse_parameters(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
users_path = spec.paths["/users"]
|
||||
get_op = users_path.get
|
||||
assert get_op is not None
|
||||
params = get_op.parameters or []
|
||||
assert len(params) == 2
|
||||
param_names = [p.name for p in params]
|
||||
assert "limit" in param_names
|
||||
assert "offset" in param_names
|
||||
|
||||
def test_parse_request_body(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
users_path = spec.paths["/users"]
|
||||
post_op = users_path.post
|
||||
assert post_op is not None
|
||||
assert post_op.request_body is not None
|
||||
|
||||
def test_parse_responses(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
users_path = spec.paths["/users"]
|
||||
get_op = users_path.get
|
||||
assert get_op is not None
|
||||
assert "200" in get_op.responses
|
||||
assert "400" in get_op.responses
|
||||
|
||||
def test_parse_components(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
assert spec.components is not None
|
||||
assert "User" in (spec.components.schemas or {})
|
||||
assert "Error" in (spec.components.schemas or {})
|
||||
|
||||
def test_parse_servers(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
assert spec.servers is not None
|
||||
assert len(spec.servers) == 1
|
||||
assert spec.servers[0].url == "https://api.example.com/v1"
|
||||
|
||||
def test_parse_tags(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
spec = parser.parse()
|
||||
users_path = spec.paths["/users"]
|
||||
get_op = users_path.get
|
||||
assert get_op is not None
|
||||
assert "Users" in (get_op.tags or [])
|
||||
|
||||
def test_validation_valid_spec(self):
|
||||
parser = OpenAPIParser(VALID_OPENAPI_SPEC)
|
||||
errors = parser.validate()
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_validation_invalid_spec(self):
|
||||
invalid_spec = {"openapi": "3.0.0", "info": {}}
|
||||
parser = OpenAPIParser(invalid_spec)
|
||||
errors = parser.validate()
|
||||
assert len(errors) > 0
|
||||
|
||||
|
||||
class TestSchemaParsing:
|
||||
def test_parse_string_schema(self):
|
||||
schema_data = {"type": "string", "format": "email", "description": "User email"}
|
||||
schema = Schema(**schema_data)
|
||||
assert schema.type == "string"
|
||||
assert schema.format == "email"
|
||||
|
||||
def test_parse_object_schema(self):
|
||||
schema_data = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"age": {"type": "integer"}
|
||||
},
|
||||
"paths": {},
|
||||
"tags": [
|
||||
{"name": "pets", "description": "Pet operations"},
|
||||
{"name": "users", "description": "User operations"}
|
||||
]
|
||||
"required": ["name"]
|
||||
}
|
||||
schema = Schema(**schema_data)
|
||||
assert schema.type == "object"
|
||||
assert schema.properties is not None
|
||||
assert "name" in schema.properties
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse()
|
||||
|
||||
tags = result.get_tags()
|
||||
assert len(tags) == 2
|
||||
tag_names = [t.name for t in tags]
|
||||
assert "pets" in tag_names
|
||||
assert "users" in tag_names
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_parse_spec_with_parameters(self):
|
||||
"""Test parsing spec with endpoint parameters."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"get": {
|
||||
"summary": "List items",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "limit",
|
||||
"in": "query",
|
||||
"schema": {"type": "integer"},
|
||||
"description": "Max items to return"
|
||||
},
|
||||
{
|
||||
"name": "offset",
|
||||
"in": "query",
|
||||
"schema": {"type": "integer"},
|
||||
"description": "Items to skip"
|
||||
}
|
||||
],
|
||||
"responses": {"200": {"description": "Success"}}
|
||||
}
|
||||
}
|
||||
}
|
||||
def test_parse_array_schema(self):
|
||||
schema_data = {
|
||||
"type": "array",
|
||||
"items": {"type": "string"}
|
||||
}
|
||||
schema = Schema(**schema_data)
|
||||
assert schema.type == "array"
|
||||
assert schema.items is not None
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse()
|
||||
|
||||
endpoints = result.get_endpoints()
|
||||
assert len(endpoints) == 1
|
||||
|
||||
params = endpoints[0].parameters or []
|
||||
assert len(params) == 2
|
||||
param_names = [p.name for p in params]
|
||||
assert "limit" in param_names
|
||||
assert "offset" in param_names
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
|
||||
class TestSpecValidation:
|
||||
"""Test cases for OpenAPI spec validation."""
|
||||
|
||||
def test_validate_valid_spec(self):
|
||||
"""Test validating a valid spec."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {}
|
||||
def test_parse_enum_schema(self):
|
||||
schema_data = {
|
||||
"type": "string",
|
||||
"enum": ["active", "inactive", "pending"]
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
is_valid, errors = parser.validate()
|
||||
|
||||
assert is_valid is True
|
||||
assert len(errors) == 0
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_validate_spec_missing_info(self):
|
||||
"""Test validating spec without info section."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"paths": {}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
is_valid, errors = parser.validate()
|
||||
|
||||
assert is_valid is False
|
||||
assert len(errors) > 0
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_validate_spec_missing_paths(self):
|
||||
"""Test validating spec without paths section."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
is_valid, errors = parser.validate()
|
||||
|
||||
assert is_valid is False
|
||||
|
||||
Path(f.name).unlink()
|
||||
schema = Schema(**schema_data)
|
||||
assert schema.enum is not None
|
||||
assert len(schema.enum) == 3
|
||||
|
||||
|
||||
class TestParseWithExamples:
|
||||
"""Test cases for parsing with generated examples."""
|
||||
|
||||
def test_parse_with_examples(self):
|
||||
"""Test parsing spec and generating examples."""
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {
|
||||
"/users": {
|
||||
"get": {
|
||||
"summary": "List users",
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Success",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "integer"},
|
||||
"name": {"type": "string"}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"User": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "integer"},
|
||||
"name": {"type": "string", "format": "email"}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
parser = OpenAPIParser(f.name)
|
||||
result = parser.parse_with_examples()
|
||||
|
||||
assert "spec" in result
|
||||
assert "endpoints" in result
|
||||
assert "schemas" in result
|
||||
|
||||
assert len(result["endpoints"]) == 1
|
||||
assert len(result["schemas"]) == 1
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
|
||||
class TestConvenienceFunctions:
|
||||
"""Test cases for convenience functions."""
|
||||
|
||||
def test_parse_spec_file(self):
|
||||
"""Test parse_spec_file convenience function."""
|
||||
from src.core.parser import parse_spec_file
|
||||
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
result = parse_spec_file(f.name)
|
||||
|
||||
assert isinstance(result, OpenAPISpec)
|
||||
assert result.info.title == "Test API"
|
||||
|
||||
Path(f.name).unlink()
|
||||
|
||||
def test_validate_spec_file(self):
|
||||
"""Test validate_spec_file convenience function."""
|
||||
from src.core.parser import validate_spec_file
|
||||
|
||||
spec = {
|
||||
"openapi": "3.0.0",
|
||||
"info": {
|
||||
"title": "Test API",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"paths": {}
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
||||
json.dump(spec, f)
|
||||
f.flush()
|
||||
|
||||
is_valid, errors = validate_spec_file(f.name)
|
||||
|
||||
assert is_valid is True
|
||||
|
||||
Path(f.name).unlink()
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
|
||||
Reference in New Issue
Block a user