MsgCenterPy/tests/test_ros_to_json_schema_conversion.py

import os
import sys

import pytest

# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

# 依赖检查
try:
    from geometry_msgs.msg import Point, Pose, Vector3
    from std_msgs.msg import Bool, Float64MultiArray, Int32, String

    # 只有在 ROS2 消息包可用时才导入 ROS2MessageInstance
    from msgcenterpy.instances.ros2_instance import ROS2MessageInstance

    HAS_ROS2 = True
except ImportError:
    HAS_ROS2 = False


import jsonschema

from msgcenterpy.core.types import MessageType
from msgcenterpy.instances.json_schema_instance import JSONSchemaMessageInstance


class TestROS2ToJSONSchemaConversion:
    """ROS2 转 JSON Schema 转换测试"""

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_string_message_to_json_schema(self):
        """测试 String 消息转 JSON Schema"""
        # 创建 ROS2 String 消息
        string_msg = String()
        string_msg.data = "Hello JSON Schema"
        ros2_inst = ROS2MessageInstance(string_msg)

        # 生成 JSON Schema
        schema = ros2_inst.get_json_schema()

        # 验证 Schema 结构
        assert schema["type"] == "object"
        assert "properties" in schema
        assert "data" in schema["properties"]
        assert schema["properties"]["data"]["type"] == "string"
        assert schema["title"] == "ROS2MessageInstance Schema"
        assert "ros2" in schema["description"]

        # 验证与原始数据的一致性
        ros2_dict = ros2_inst.get_python_dict()
        assert "data" in ros2_dict
        assert ros2_dict["data"] == "Hello JSON Schema"

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_float_array_to_json_schema(self):
        """测试 Float64MultiArray 转 JSON Schema"""
        # 创建 Float64MultiArray 消息
        array_msg = Float64MultiArray()
        array_msg.data = [1.1, 2.2, 3.3, 4.4, 5.5]
        ros2_inst = ROS2MessageInstance(array_msg)

        # 生成 JSON Schema
        schema = ros2_inst.get_json_schema()

        # 验证数组字段的 Schema
        assert "data" in schema["properties"]
        data_prop = schema["properties"]["data"]
        assert data_prop["type"] == "array"
        assert "items" in data_prop
        assert data_prop["items"]["type"] == "number"

        # 验证约束条件
        assert "minItems" in data_prop or "maxItems" in data_prop or data_prop["type"] == "array"

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_pose_message_to_json_schema(self):
        """测试复杂 Pose 消息转 JSON Schema"""
        # 创建 Pose 消息
        pose_msg = Pose()
        pose_msg.position.x = 1.0
        pose_msg.position.y = 2.0
        pose_msg.position.z = 3.0
        pose_msg.orientation.w = 1.0
        ros2_inst = ROS2MessageInstance(pose_msg)

        # 生成 JSON Schema
        schema = ros2_inst.get_json_schema()

        # 验证嵌套对象结构
        properties = schema["properties"]
        assert "position" in properties
        assert "orientation" in properties

        # 验证对象类型
        position_prop = properties["position"]
        assert position_prop["type"] == "object"

        orientation_prop = properties["orientation"]
        assert orientation_prop["type"] == "object"


class TestROS2ToJSONSchemaInstanceConversion:
    """ROS2 转 JSONSchemaMessageInstance 转换测试"""

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_ros2_to_json_schema_instance_string(self):
        """测试 ROS2 String 转 JSONSchemaMessageInstance"""
        # 创建 ROS2 实例
        string_msg = String()
        string_msg.data = "Test conversion"
        ros2_inst = ROS2MessageInstance(string_msg)

        # 转换为 JSONSchemaMessageInstance
        json_schema_inst = ros2_inst.to_json_schema()

        # 验证转换结果
        assert isinstance(json_schema_inst, JSONSchemaMessageInstance)
        assert json_schema_inst.message_type == MessageType.JSON_SCHEMA

        # 验证数据一致性
        original_data = ros2_inst.get_python_dict()
        converted_data = json_schema_inst.get_python_dict()
        assert original_data == converted_data

        # 验证 Schema 存在
        schema = json_schema_inst.json_schema
        assert schema is not None
        assert schema["type"] == "object"
        assert "properties" in schema

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_ros2_to_json_schema_instance_array(self):
        """测试 ROS2 数组转 JSONSchemaMessageInstance"""
        # 创建 ROS2 数组实例
        array_msg = Float64MultiArray()
        array_msg.data = [10.5, 20.3, 30.7]
        ros2_inst = ROS2MessageInstance(array_msg)

        # 转换为 JSONSchemaMessageInstance
        json_schema_inst = ros2_inst.to_json_schema()

        # 验证转换结果
        assert isinstance(json_schema_inst, JSONSchemaMessageInstance)

        # 验证数据一致性
        original_data = ros2_inst.get_python_dict()
        converted_data = json_schema_inst.get_python_dict()
        assert original_data == converted_data
        assert converted_data["data"] == [10.5, 20.3, 30.7]

        # 验证 Schema 的数组类型
        schema = json_schema_inst.json_schema
        if "data" in schema["properties"]:
            data_prop = schema["properties"]["data"]
            assert data_prop["type"] == "array"

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_ros2_to_json_schema_instance_pose(self):
        """测试 ROS2 Pose 转 JSONSchemaMessageInstance"""
        # 创建复杂的 Pose 消息
        pose_msg = Pose()
        pose_msg.position.x = 5.0
        pose_msg.position.y = 10.0
        pose_msg.position.z = 15.0
        pose_msg.orientation.x = 0.0
        pose_msg.orientation.y = 0.0
        pose_msg.orientation.z = 0.0
        pose_msg.orientation.w = 1.0
        ros2_inst = ROS2MessageInstance(pose_msg)

        # 转换为 JSONSchemaMessageInstance
        json_schema_inst = ros2_inst.to_json_schema()

        # 验证转换结果
        assert isinstance(json_schema_inst, JSONSchemaMessageInstance)

        # 验证嵌套数据一致性
        original_data = ros2_inst.get_python_dict()
        converted_data = json_schema_inst.get_python_dict()
        assert original_data == converted_data

        # 验证嵌套结构
        assert "position" in converted_data
        assert "orientation" in converted_data
        assert converted_data["position"]["x"] == 5.0
        assert converted_data["position"]["y"] == 10.0
        assert converted_data["position"]["z"] == 15.0


class TestConversionConsistency:
    """转换一致性测试"""

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_schema_data_consistency(self):
        """测试 Schema 与数据的一致性"""
        # 创建多种类型的数据
        test_cases = []

        # String 消息
        string_msg = String()
        string_msg.data = "consistency_test"
        test_cases.append(("String", ROS2MessageInstance(string_msg)))

        # Float64MultiArray 消息
        array_msg = Float64MultiArray()
        array_msg.data = [1.0, 2.0, 3.0]
        test_cases.append(("Float64MultiArray", ROS2MessageInstance(array_msg)))

        for test_name, ros2_inst in test_cases:
            # 生成 Schema
            schema = ros2_inst.get_json_schema()
            original_data = ros2_inst.get_python_dict()

            # 验证 Schema 属性与数据字段一致
            schema_props = set(schema["properties"].keys())
            data_keys = set(original_data.keys())

            assert schema_props == data_keys, f"{test_name}: Schema properties and data keys don't match"

            # 验证每个字段的类型一致性
            for field_name, field_value in original_data.items():
                if field_name in schema["properties"]:
                    prop_schema = schema["properties"][field_name]

                    # 基本类型检查
                    if isinstance(field_value, str):
                        assert prop_schema["type"] == "string", f"{test_name}.{field_name}: Type mismatch"
                    elif isinstance(field_value, bool):
                        assert prop_schema["type"] == "boolean", f"{test_name}.{field_name}: Type mismatch"
                    elif isinstance(field_value, int):
                        assert prop_schema["type"] == "integer", f"{test_name}.{field_name}: Type mismatch"
                    elif isinstance(field_value, float):
                        assert prop_schema["type"] == "number", f"{test_name}.{field_name}: Type mismatch"
                    elif isinstance(field_value, list):
                        assert prop_schema["type"] == "array", f"{test_name}.{field_name}: Type mismatch"
                    elif isinstance(field_value, dict):
                        assert prop_schema["type"] == "object", f"{test_name}.{field_name}: Type mismatch"

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_conversion_roundtrip_data(self):
        """测试转换往返的数据一致性"""
        # 创建原始 ROS2 消息
        string_msg = String()
        string_msg.data = "roundtrip_test"
        original_ros2_inst = ROS2MessageInstance(string_msg)
        original_data = original_ros2_inst.get_python_dict()

        # ROS2 -> JSONSchema
        json_schema_inst = original_ros2_inst.to_json_schema()
        converted_data = json_schema_inst.get_python_dict()

        # 验证数据在转换过程中保持一致
        assert original_data == converted_data

        # 验证 Schema 的有效性
        schema = json_schema_inst.json_schema
        assert schema is not None

        # 使用 jsonschema 库验证数据符合生成的 Schema
        try:
            jsonschema.validate(converted_data, schema)
        except jsonschema.ValidationError as e:
            pytest.fail(f"Generated data doesn't match generated schema: {e}")

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_multiple_message_types_conversion(self):
        """测试多种消息类型的转换"""
        test_messages = []

        # String
        string_msg = String()
        string_msg.data = "multi_test_string"
        test_messages.append(("String", string_msg))

        # Float64MultiArray
        array_msg = Float64MultiArray()
        array_msg.data = [1.5, 2.5, 3.5]
        test_messages.append(("Float64MultiArray", array_msg))

        # Point
        point_msg = Point()
        point_msg.x = 1.0
        point_msg.y = 2.0
        point_msg.z = 3.0
        test_messages.append(("Point", point_msg))

        for msg_type_name, msg in test_messages:
            # 创建 ROS2 实例
            ros2_inst = ROS2MessageInstance(msg)

            # 转换为 JSONSchema 实例
            json_schema_inst = ros2_inst.to_json_schema()

            # 验证转换成功
            assert isinstance(json_schema_inst, JSONSchemaMessageInstance)
            assert json_schema_inst.message_type == MessageType.JSON_SCHEMA

            # 验证数据一致性
            original_data = ros2_inst.get_python_dict()
            converted_data = json_schema_inst.get_python_dict()
            assert original_data == converted_data, f"{msg_type_name}: Data inconsistency after conversion"

            # 验证 Schema 生成
            schema = json_schema_inst.json_schema
            assert schema is not None
            assert schema["type"] == "object"
            assert len(schema["properties"]) > 0, f"{msg_type_name}: No properties in generated schema"


class TestConversionErrorHandling:
    """转换错误处理测试"""

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_empty_message_conversion(self):
        """测试空消息的转换"""
        # 创建空的 String 消息
        empty_string = String()
        ros2_inst = ROS2MessageInstance(empty_string)

        # 转换应该成功，即使数据为空
        json_schema_inst = ros2_inst.to_json_schema()

        assert isinstance(json_schema_inst, JSONSchemaMessageInstance)

        # 验证 Schema 仍然有效
        schema = json_schema_inst.json_schema
        assert schema["type"] == "object"
        assert "properties" in schema

    @pytest.mark.skipif(not HAS_ROS2, reason="Missing dependencies")
    def test_large_data_conversion(self):
        """测试大数据量的转换"""
        # 创建包含大量数据的数组消息
        large_array = Float64MultiArray()
        large_array.data = [float(i) for i in range(1000)]  # 1000 个浮点数
        ros2_inst = ROS2MessageInstance(large_array)

        # 转换应该能处理大数据量
        json_schema_inst = ros2_inst.to_json_schema()

        assert isinstance(json_schema_inst, JSONSchemaMessageInstance)

        # 验证数据完整性
        original_data = ros2_inst.get_python_dict()
        converted_data = json_schema_inst.get_python_dict()
        assert len(original_data["data"]) == 1000
        assert len(converted_data["data"]) == 1000
        assert original_data == converted_data


# 运行测试的便捷函数
def run_conversion_tests():
    """运行转换测试"""
    if not HAS_ROS2:
        print("❌ Required dependencies not available, skipping tests")
        print(f"  ROS2: {HAS_ROS2}")
        return False

    import subprocess

    result = subprocess.run(
        [
            "python",
            "-m",
            "pytest",
            "tests/test_ros_to_json_schema_conversion.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_conversion_tests()