mirror of
https://github.com/ZGCA-Forge/MsgCenterPy.git
synced 2025-12-14 13:04:34 +00:00
464 lines
17 KiB
Python
464 lines
17 KiB
Python
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
# JSON Schema 依赖检查
|
|
|
|
from msgcenterpy.core.type_info import ConstraintType
|
|
from msgcenterpy.core.types import MessageType
|
|
from msgcenterpy.instances.json_schema_instance import JSONSchemaMessageInstance
|
|
|
|
|
|
class TestJSONSchemaMessageInstance:
|
|
"""JSONSchemaMessageInstance 基本功能测试"""
|
|
|
|
@pytest.fixture
|
|
def simple_schema(self):
|
|
"""简单的 JSON Schema 示例"""
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"age": {"type": "integer", "minimum": 0, "maximum": 150},
|
|
"active": {"type": "boolean"},
|
|
},
|
|
"required": ["name"],
|
|
}
|
|
|
|
@pytest.fixture
|
|
def complex_schema(self):
|
|
"""复杂的 JSON Schema 示例"""
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"user": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "string"},
|
|
"profile": {
|
|
"type": "object",
|
|
"properties": {
|
|
"email": {"type": "string", "format": "email"},
|
|
"phone": {
|
|
"type": "string",
|
|
"pattern": "^\\+?[1-9]\\d{1,14}$",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
"required": ["id"],
|
|
},
|
|
"tags": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"minItems": 1,
|
|
"maxItems": 10,
|
|
},
|
|
"scores": {
|
|
"type": "array",
|
|
"items": {"type": "number", "minimum": 0, "maximum": 100},
|
|
},
|
|
"metadata": {"type": "object", "additionalProperties": True},
|
|
},
|
|
"required": ["user", "tags"],
|
|
}
|
|
|
|
@pytest.fixture
|
|
def simple_data(self):
|
|
"""匹配简单 schema 的数据"""
|
|
return {"name": "John Doe", "age": 30, "active": True}
|
|
|
|
@pytest.fixture
|
|
def complex_data(self):
|
|
"""匹配复杂 schema 的数据"""
|
|
return {
|
|
"user": {
|
|
"id": "user_123",
|
|
"profile": {"email": "john@example.com", "phone": "+1234567890"},
|
|
},
|
|
"tags": ["developer", "python", "testing"],
|
|
"scores": [85.5, 92.0, 78.3],
|
|
"metadata": {"created_at": "2024-01-01", "version": 1},
|
|
}
|
|
|
|
def test_basic_creation(self, simple_schema, simple_data):
|
|
"""测试基本创建功能"""
|
|
instance = JSONSchemaMessageInstance(simple_data, simple_schema)
|
|
|
|
assert instance.message_type == MessageType.JSON_SCHEMA
|
|
assert instance.inner_data == simple_data
|
|
assert instance.json_schema == simple_schema
|
|
assert len(instance._validation_errors) == 0
|
|
|
|
def test_data_validation_success(self, simple_schema, simple_data):
|
|
"""测试数据验证成功"""
|
|
instance = JSONSchemaMessageInstance(simple_data, simple_schema)
|
|
|
|
# 验证成功，没有错误
|
|
assert len(instance._validation_errors) == 0
|
|
|
|
def test_data_validation_failure(self, simple_schema):
|
|
"""测试数据验证失败"""
|
|
invalid_data = {
|
|
"age": -5, # 违反 minimum 约束
|
|
"active": "not_boolean", # 类型错误
|
|
# 缺少必需的 "name" 字段
|
|
}
|
|
|
|
instance = JSONSchemaMessageInstance(invalid_data, simple_schema)
|
|
|
|
# 应该有验证错误
|
|
assert len(instance._validation_errors) > 0
|
|
|
|
def test_get_python_dict(self, simple_schema, simple_data):
|
|
"""测试获取 Python 字典"""
|
|
instance = JSONSchemaMessageInstance(simple_data, simple_schema)
|
|
|
|
result = instance.get_python_dict()
|
|
|
|
assert isinstance(result, dict)
|
|
assert result == simple_data
|
|
assert result is not simple_data # 应该是副本
|
|
|
|
def test_set_python_dict(self, simple_schema, simple_data):
|
|
"""测试设置 Python 字典"""
|
|
instance = JSONSchemaMessageInstance(simple_data, simple_schema)
|
|
|
|
new_data = {"name": "Jane Smith", "age": 25}
|
|
|
|
result = instance.set_python_dict(new_data)
|
|
|
|
assert result is True
|
|
assert instance.get_python_dict()["name"] == "Jane Smith"
|
|
assert instance.get_python_dict()["age"] == 25
|
|
|
|
def test_export_to_envelope(self, simple_schema, simple_data):
|
|
"""测试导出信封"""
|
|
instance = JSONSchemaMessageInstance(simple_data, simple_schema)
|
|
|
|
envelope = instance.export_to_envelope()
|
|
|
|
assert "content" in envelope
|
|
assert "metadata" in envelope
|
|
assert envelope["content"] == simple_data
|
|
|
|
metadata = envelope["metadata"]
|
|
assert metadata["current_format"] == "json_schema"
|
|
assert "properties" in metadata
|
|
|
|
def test_import_from_envelope(self, simple_schema, simple_data):
|
|
"""测试从信封导入"""
|
|
# 创建原始实例
|
|
original = JSONSchemaMessageInstance(simple_data, simple_schema)
|
|
envelope = original.export_to_envelope()
|
|
|
|
# 从信封导入新实例
|
|
new_instance = JSONSchemaMessageInstance.import_from_envelope(envelope)
|
|
|
|
assert isinstance(new_instance, JSONSchemaMessageInstance)
|
|
assert new_instance.get_python_dict() == simple_data
|
|
assert new_instance.json_schema == simple_schema
|
|
|
|
|
|
class TestJSONSchemaFieldTypeInfo:
|
|
"""JSON Schema 字段类型信息测试"""
|
|
|
|
@pytest.fixture
|
|
def typed_schema(self):
|
|
"""包含各种类型的 schema"""
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"string_field": {"type": "string", "minLength": 3, "maxLength": 50},
|
|
"integer_field": {"type": "integer", "minimum": 0, "maximum": 100},
|
|
"number_field": {"type": "number", "multipleOf": 0.5},
|
|
"boolean_field": {"type": "boolean"},
|
|
"array_field": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"minItems": 1,
|
|
"maxItems": 5,
|
|
},
|
|
"object_field": {
|
|
"type": "object",
|
|
"properties": {"nested_string": {"type": "string"}},
|
|
},
|
|
"enum_field": {
|
|
"type": "string",
|
|
"enum": ["option1", "option2", "option3"],
|
|
},
|
|
"format_field": {"type": "string", "format": "email"},
|
|
},
|
|
"required": ["string_field", "integer_field"],
|
|
}
|
|
|
|
@pytest.fixture
|
|
def typed_data(self):
|
|
"""匹配类型化 schema 的数据"""
|
|
return {
|
|
"string_field": "hello",
|
|
"integer_field": 42,
|
|
"number_field": 3.5,
|
|
"boolean_field": True,
|
|
"array_field": ["item1", "item2"],
|
|
"object_field": {"nested_string": "nested_value"},
|
|
"enum_field": "option1",
|
|
"format_field": "test@example.com",
|
|
}
|
|
|
|
def test_string_field_type_info(self, typed_schema, typed_data):
|
|
"""测试字符串字段类型信息"""
|
|
instance = JSONSchemaMessageInstance(typed_data, typed_schema)
|
|
type_info = instance.fields.get_sub_type_info("string_field")
|
|
|
|
assert type_info is not None
|
|
assert type_info.field_name == "string_field"
|
|
assert type_info.standard_type.value == "string"
|
|
assert type_info.current_value == "hello"
|
|
|
|
# 检查约束
|
|
assert type_info.has_constraint(ConstraintType.MIN_LENGTH)
|
|
assert type_info.get_constraint_value(ConstraintType.MIN_LENGTH) == 3
|
|
assert type_info.has_constraint(ConstraintType.MAX_LENGTH)
|
|
assert type_info.get_constraint_value(ConstraintType.MAX_LENGTH) == 50
|
|
assert type_info.has_constraint(ConstraintType.REQUIRED)
|
|
|
|
def test_integer_field_type_info(self, typed_schema, typed_data):
|
|
"""测试整数字段类型信息"""
|
|
instance = JSONSchemaMessageInstance(typed_data, typed_schema)
|
|
type_info = instance.fields.get_sub_type_info("integer_field")
|
|
|
|
assert type_info is not None
|
|
assert type_info.standard_type.value == "integer"
|
|
assert type_info.current_value == 42
|
|
|
|
# 检查数值约束
|
|
assert type_info.has_constraint(ConstraintType.MIN_VALUE)
|
|
assert type_info.get_constraint_value(ConstraintType.MIN_VALUE) == 0
|
|
assert type_info.has_constraint(ConstraintType.MAX_VALUE)
|
|
assert type_info.get_constraint_value(ConstraintType.MAX_VALUE) == 100
|
|
assert type_info.has_constraint(ConstraintType.REQUIRED)
|
|
|
|
def test_array_field_type_info(self, typed_schema, typed_data):
|
|
"""测试数组字段类型信息"""
|
|
instance = JSONSchemaMessageInstance(typed_data, typed_schema)
|
|
type_info = instance.fields.get_sub_type_info("array_field")
|
|
|
|
assert type_info is not None
|
|
assert type_info.is_array is True
|
|
assert type_info.current_value == ["item1", "item2"]
|
|
|
|
# 检查数组约束
|
|
assert type_info.has_constraint(ConstraintType.MIN_ITEMS)
|
|
assert type_info.get_constraint_value(ConstraintType.MIN_ITEMS) == 1
|
|
assert type_info.has_constraint(ConstraintType.MAX_ITEMS)
|
|
assert type_info.get_constraint_value(ConstraintType.MAX_ITEMS) == 5
|
|
|
|
# 检查元素类型信息
|
|
assert type_info.element_type_info is not None
|
|
assert type_info.element_type_info.standard_type.value == "string"
|
|
|
|
def test_object_field_type_info(self, typed_schema, typed_data):
|
|
"""测试对象字段类型信息"""
|
|
instance = JSONSchemaMessageInstance(typed_data, typed_schema)
|
|
type_info = instance.fields.get_sub_type_info("object_field")
|
|
|
|
assert type_info is not None
|
|
assert type_info.is_object is True
|
|
assert type_info.current_value == {"nested_string": "nested_value"}
|
|
|
|
# 检查对象字段定义
|
|
assert len(type_info.object_fields) > 0
|
|
assert "nested_string" in type_info.object_fields
|
|
|
|
nested_field_info = type_info.object_fields["nested_string"]
|
|
assert nested_field_info.standard_type.value == "string"
|
|
|
|
def test_enum_field_type_info(self, typed_schema, typed_data):
|
|
"""测试枚举字段类型信息"""
|
|
instance = JSONSchemaMessageInstance(typed_data, typed_schema)
|
|
type_info = instance.fields.get_sub_type_info("enum_field")
|
|
|
|
assert type_info is not None
|
|
assert type_info.has_constraint(ConstraintType.ENUM_VALUES)
|
|
enum_values = type_info.get_constraint_value(ConstraintType.ENUM_VALUES)
|
|
assert enum_values == ["option1", "option2", "option3"]
|
|
|
|
def test_format_field_type_info(self, typed_schema, typed_data):
|
|
"""测试格式字段类型信息"""
|
|
instance = JSONSchemaMessageInstance(typed_data, typed_schema)
|
|
type_info = instance.fields.get_sub_type_info("format_field")
|
|
|
|
assert type_info is not None
|
|
assert type_info.has_constraint(ConstraintType.FORMAT)
|
|
format_value = type_info.get_constraint_value(ConstraintType.FORMAT)
|
|
assert format_value == "email"
|
|
|
|
|
|
class TestJSONSchemaInstanceJSONSchema:
|
|
"""JSONSchemaMessageInstance 自身的 JSON Schema 生成测试"""
|
|
|
|
def test_get_json_schema_simple(self):
|
|
"""测试简单数据的 JSON Schema 生成"""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"name": {"type": "string"}, "count": {"type": "integer"}},
|
|
}
|
|
data = {"name": "test", "count": 5}
|
|
|
|
instance = JSONSchemaMessageInstance(data, schema)
|
|
generated_schema = instance.get_json_schema()
|
|
|
|
assert generated_schema["type"] == "object"
|
|
assert "properties" in generated_schema
|
|
assert "name" in generated_schema["properties"]
|
|
assert "count" in generated_schema["properties"]
|
|
assert generated_schema["title"] == "JSONSchemaMessageInstance Schema"
|
|
|
|
def test_get_json_schema_with_constraints(self):
|
|
"""测试包含约束的 JSON Schema 生成"""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"email": {"type": "string", "format": "email", "minLength": 5},
|
|
"age": {"type": "integer", "minimum": 0, "maximum": 120},
|
|
"tags": {"type": "array", "items": {"type": "string"}, "minItems": 1},
|
|
},
|
|
"required": ["email"],
|
|
}
|
|
data = {"email": "test@example.com", "age": 25, "tags": ["tag1", "tag2"]}
|
|
|
|
instance = JSONSchemaMessageInstance(data, schema)
|
|
generated_schema = instance.get_json_schema()
|
|
|
|
# 检查约束是否保留在生成的 schema 中
|
|
properties = generated_schema["properties"]
|
|
|
|
# email 字段约束
|
|
email_prop = properties["email"]
|
|
assert email_prop["type"] == "string"
|
|
|
|
# age 字段约束
|
|
age_prop = properties["age"]
|
|
assert age_prop["type"] == "integer"
|
|
|
|
# tags 数组约束
|
|
tags_prop = properties["tags"]
|
|
assert tags_prop["type"] == "array"
|
|
|
|
def test_get_json_schema_nested_objects(self):
|
|
"""测试嵌套对象的 JSON Schema 生成"""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"user": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "string"},
|
|
"settings": {
|
|
"type": "object",
|
|
"properties": {"theme": {"type": "string"}},
|
|
},
|
|
},
|
|
}
|
|
},
|
|
}
|
|
data = {"user": {"id": "user123", "settings": {"theme": "dark"}}}
|
|
|
|
instance = JSONSchemaMessageInstance(data, schema)
|
|
generated_schema = instance.get_json_schema()
|
|
|
|
assert "user" in generated_schema["properties"]
|
|
user_prop = generated_schema["properties"]["user"]
|
|
assert user_prop["type"] == "object"
|
|
|
|
|
|
class TestJSONSchemaValidation:
|
|
"""JSON Schema 验证功能测试"""
|
|
|
|
def test_constraint_validation(self):
|
|
"""测试约束验证"""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"age": {"type": "integer", "minimum": 0, "maximum": 150},
|
|
"name": {"type": "string", "minLength": 2, "maxLength": 50},
|
|
},
|
|
"required": ["name"],
|
|
}
|
|
|
|
# 有效数据
|
|
valid_data = {"name": "John", "age": 30}
|
|
valid_instance = JSONSchemaMessageInstance(valid_data, schema)
|
|
assert len(valid_instance._validation_errors) == 0
|
|
|
|
# 无效数据 - 年龄超出范围
|
|
invalid_data1 = {"name": "John", "age": 200}
|
|
invalid_instance1 = JSONSchemaMessageInstance(invalid_data1, schema)
|
|
assert len(invalid_instance1._validation_errors) > 0
|
|
|
|
# 无效数据 - 缺少必需字段
|
|
invalid_data2 = {"age": 30}
|
|
invalid_instance2 = JSONSchemaMessageInstance(invalid_data2, schema)
|
|
assert len(invalid_instance2._validation_errors) > 0
|
|
|
|
def test_type_validation(self):
|
|
"""测试类型验证"""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"count": {"type": "integer"},
|
|
"active": {"type": "boolean"},
|
|
"items": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
}
|
|
|
|
# 类型正确的数据
|
|
valid_data = {"count": 42, "active": True, "items": ["a", "b", "c"]}
|
|
valid_instance = JSONSchemaMessageInstance(valid_data, schema)
|
|
assert len(valid_instance._validation_errors) == 0
|
|
|
|
# 类型错误的数据
|
|
invalid_data = {
|
|
"count": "not_integer",
|
|
"active": "not_boolean",
|
|
"items": "not_array",
|
|
}
|
|
invalid_instance = JSONSchemaMessageInstance(invalid_data, schema)
|
|
assert len(invalid_instance._validation_errors) > 0
|
|
|
|
|
|
# 运行测试的便捷函数
|
|
def run_json_schema_tests():
|
|
"""运行 JSON Schema 相关测试"""
|
|
|
|
import subprocess
|
|
|
|
result = subprocess.run(
|
|
[
|
|
"python",
|
|
"-m",
|
|
"pytest",
|
|
"tests/test_json_schema_instance.py",
|
|
"-v",
|
|
"--tb=short",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
print(result.stdout)
|
|
if result.stderr:
|
|
print("STDERR:", result.stderr)
|
|
|
|
return result.returncode == 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 直接运行测试
|
|
run_json_schema_tests()
|