feat: add StorageAdapter abstraction and InMemoryAdapter with full test suite
- 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
This commit is contained in:
@@ -0,0 +1,314 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user