aazerra
5c8310b837
- Introduce StorageAdapter protocol and InMemoryAdapter (thread-safe, in-memory) in engine/adapter.py to decouple storage from sync logic - Refactor SyncEngine to accept a StorageAdapter, replacing the previous stub that returned an empty list - Wire server to use SyncEngine + InMemoryAdapter, removing hardcoded SYNC_OPERATIONS fixtures - Add tests for engine, adapter, generator, and types (tests/) - Add pre-commit config (.pre-commit-config.yaml) with linting/formatting hooks - Update pyproject.toml and uv.lock with new dev dependencies - Clean up type annotations and CLI help strings across generator, server, and utils
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Tests for generator.languages.python — unit and integration."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from generator.languages.python import PythonGenerator
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
OPERATION_SCHEMA = {
|
|
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
|
"$id": "https://example.com/operation.schema.json",
|
|
"title": "Operation",
|
|
"description": "A single operation.",
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "string", "description": "Unique id."},
|
|
"type": {"type": "string", "description": "Op type."},
|
|
"count": {"type": "integer", "description": "Count."},
|
|
"score": {"type": "number", "description": "Score."},
|
|
"active": {"type": "boolean", "description": "Active flag."},
|
|
"meta": {"type": "object", "description": "Metadata."},
|
|
},
|
|
}
|
|
|
|
ITEM_SCHEMA = {
|
|
"title": "Item",
|
|
"description": "An item.",
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string", "description": "Item name."},
|
|
},
|
|
"required": ["name"],
|
|
}
|
|
|
|
CONTAINER_SCHEMA = {
|
|
"title": "Container",
|
|
"description": "A container of items.",
|
|
"type": "object",
|
|
"properties": {
|
|
"items": {
|
|
"type": "array",
|
|
"items": {"$ref": "item.schema.json"},
|
|
"description": "Items list.",
|
|
},
|
|
"label": {"type": "string", "description": "Label."},
|
|
},
|
|
"required": ["label"],
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def schema_dir(tmp_path: Path) -> Path:
|
|
"""Write a minimal set of schemas to a temp directory and return its path."""
|
|
(tmp_path / "operation.schema.json").write_text(json.dumps(OPERATION_SCHEMA))
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture()
|
|
def ref_schema_dir(tmp_path: Path) -> Path:
|
|
"""Write item + container schemas (cross-ref) to a temp directory."""
|
|
(tmp_path / "item.schema.json").write_text(json.dumps(ITEM_SCHEMA))
|
|
(tmp_path / "container.schema.json").write_text(json.dumps(CONTAINER_SCHEMA))
|
|
return tmp_path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _schema_to_module
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSchemaToModule:
|
|
@pytest.fixture(autouse=True)
|
|
def gen(self, schema_dir: Path) -> None:
|
|
self.gen = PythonGenerator(schema_dir)
|
|
|
|
def test_removes_schema_json_suffix(self) -> None:
|
|
assert self.gen._schema_to_module("operation.schema.json") == "operation"
|
|
|
|
def test_replaces_hyphens_with_underscores(self) -> None:
|
|
assert self.gen._schema_to_module("sync-request.schema.json") == "sync_request"
|
|
|
|
def test_combined(self) -> None:
|
|
assert self.gen._schema_to_module("my-type.schema.json") == "my_type"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _ref_to_filename
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRefToFilename:
|
|
@pytest.fixture(autouse=True)
|
|
def gen(self, schema_dir: Path) -> None:
|
|
self.gen = PythonGenerator(schema_dir)
|
|
|
|
def test_bare_filename(self) -> None:
|
|
result = self.gen._ref_to_filename("operation.schema.json")
|
|
assert result == "operation.schema.json"
|
|
|
|
def test_relative_prefix(self) -> None:
|
|
result = self.gen._ref_to_filename("./operation.schema.json")
|
|
assert result == "operation.schema.json"
|
|
|
|
def test_nested_path(self) -> None:
|
|
result = self.gen._ref_to_filename("schemas/operation.schema.json")
|
|
assert result == "operation.schema.json"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _find_refs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFindRefs:
|
|
@pytest.fixture(autouse=True)
|
|
def gen(self, schema_dir: Path) -> None:
|
|
self.gen = PythonGenerator(schema_dir)
|
|
|
|
def test_no_refs(self) -> None:
|
|
schema = {"type": "object", "properties": {"id": {"type": "string"}}}
|
|
assert self.gen._find_refs(schema) == []
|
|
|
|
def test_top_level_ref(self) -> None:
|
|
schema = {"$ref": "other.schema.json"}
|
|
assert self.gen._find_refs(schema) == ["other.schema.json"]
|
|
|
|
def test_nested_ref_in_properties(self) -> None:
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"child": {"$ref": "child.schema.json"},
|
|
},
|
|
}
|
|
refs = self.gen._find_refs(schema)
|
|
assert "child.schema.json" in refs
|
|
|
|
def test_ref_inside_array_items(self) -> None:
|
|
schema = {
|
|
"type": "array",
|
|
"items": {"$ref": "item.schema.json"},
|
|
}
|
|
refs = self.gen._find_refs(schema)
|
|
assert "item.schema.json" in refs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_type
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResolveType:
|
|
@pytest.fixture(autouse=True)
|
|
def gen(self, schema_dir: Path) -> None:
|
|
self.gen = PythonGenerator(schema_dir)
|
|
|
|
@pytest.mark.parametrize(
|
|
"json_type,expected",
|
|
[
|
|
("string", "str"),
|
|
("integer", "int"),
|
|
("number", "float"),
|
|
("boolean", "bool"),
|
|
("object", "dict"),
|
|
],
|
|
)
|
|
def test_primitive_types(self, json_type: str, expected: str) -> None:
|
|
info = self.gen._resolve_type({"type": json_type})
|
|
assert info.type == expected
|
|
assert info.imports == set()
|
|
|
|
def test_array_of_string(self) -> None:
|
|
info = self.gen._resolve_type({"type": "array", "items": {"type": "string"}})
|
|
assert info.type == "list[str]"
|
|
|
|
def test_format_int64(self) -> None:
|
|
info = self.gen._resolve_type({"type": "integer", "format": "int64"})
|
|
assert info.type == "int"
|
|
|
|
def test_unsupported_type_raises(self) -> None:
|
|
with pytest.raises(ValueError, match="Unsupported schema"):
|
|
self.gen._resolve_type({"type": "unknown_type"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _parse_model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestParseModel:
|
|
@pytest.fixture(autouse=True)
|
|
def gen(self, schema_dir: Path) -> None:
|
|
self.gen = PythonGenerator(schema_dir)
|
|
|
|
def test_model_name_from_title(self) -> None:
|
|
model = self.gen._parse_model(OPERATION_SCHEMA)
|
|
assert model.name == "Operation"
|
|
|
|
def test_model_description(self) -> None:
|
|
model = self.gen._parse_model(OPERATION_SCHEMA)
|
|
assert model.description == "A single operation."
|
|
|
|
def test_all_fields_optional_when_no_required(self) -> None:
|
|
model = self.gen._parse_model(OPERATION_SCHEMA)
|
|
assert all(f.optional for f in model.fields)
|
|
|
|
def test_required_fields_not_optional(self) -> None:
|
|
schema = {
|
|
"title": "Req",
|
|
"description": "desc",
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"age": {"type": "integer"},
|
|
},
|
|
"required": ["name"],
|
|
}
|
|
model = self.gen._parse_model(schema)
|
|
name_field = next(f for f in model.fields if f.name == "name")
|
|
age_field = next(f for f in model.fields if f.name == "age")
|
|
assert not name_field.optional
|
|
assert age_field.optional
|
|
|
|
def test_optional_fields_have_none_default_in_type(self) -> None:
|
|
model = self.gen._parse_model(OPERATION_SCHEMA)
|
|
for field in model.fields:
|
|
assert "| None = None" in field.type
|
|
|
|
def test_required_fields_sorted_before_optional(self) -> None:
|
|
schema = {
|
|
"title": "T",
|
|
"description": "",
|
|
"type": "object",
|
|
"properties": {
|
|
"opt_field": {"type": "string"},
|
|
"req_field": {"type": "integer"},
|
|
},
|
|
"required": ["req_field"],
|
|
}
|
|
model = self.gen._parse_model(schema)
|
|
assert model.fields[0].name == "req_field"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration — generate single schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGenerateIntegration:
|
|
def test_generates_python_file(self, schema_dir: Path, tmp_path: Path) -> None:
|
|
gen = PythonGenerator(schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=False, schema_file="operation.schema.json")
|
|
assert (out / "operation.py").exists()
|
|
|
|
def test_generated_file_contains_class(
|
|
self, schema_dir: Path, tmp_path: Path
|
|
) -> None:
|
|
gen = PythonGenerator(schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=False, schema_file="operation.schema.json")
|
|
content = (out / "operation.py").read_text()
|
|
assert "class Operation(BaseModel):" in content
|
|
|
|
def test_generated_file_has_encode_decode(
|
|
self, schema_dir: Path, tmp_path: Path
|
|
) -> None:
|
|
gen = PythonGenerator(schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=False, schema_file="operation.schema.json")
|
|
content = (out / "operation.py").read_text()
|
|
assert "def encode" in content
|
|
assert "def decode" in content
|
|
|
|
def test_generate_all_creates_all_files(
|
|
self, schema_dir: Path, tmp_path: Path
|
|
) -> None:
|
|
gen = PythonGenerator(schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=True, schema_file=None)
|
|
assert (out / "operation.py").exists()
|
|
|
|
def test_cross_ref_generates_dependency_first(
|
|
self, ref_schema_dir: Path, tmp_path: Path
|
|
) -> None:
|
|
"""Generating container.schema.json must also produce item.py."""
|
|
gen = PythonGenerator(ref_schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=False, schema_file="container.schema.json")
|
|
assert (out / "item.py").exists()
|
|
assert (out / "container.py").exists()
|
|
|
|
def test_cross_ref_import_present_in_generated_file(
|
|
self, ref_schema_dir: Path, tmp_path: Path
|
|
) -> None:
|
|
gen = PythonGenerator(ref_schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=False, schema_file="container.schema.json")
|
|
content = (out / "container.py").read_text()
|
|
assert "from .item import Item" in content
|
|
|
|
def test_generate_all_is_idempotent(self, schema_dir: Path, tmp_path: Path) -> None:
|
|
"""Running generate twice must not raise or produce duplicate content."""
|
|
gen = PythonGenerator(schema_dir)
|
|
out = tmp_path / "out"
|
|
gen.generate(out, make_all=True, schema_file=None)
|
|
content_first = (out / "operation.py").read_text()
|
|
gen.generate(out, make_all=True, schema_file=None)
|
|
content_second = (out / "operation.py").read_text()
|
|
assert content_first == content_second
|