video_decode/test/functions/test_json_utils.py

"""
JSON 安全解析功能组件的单元测试
"""

import pytest
import json
from src.components.functions.json_utils import (
    JsonSafeParseFunction,
    JSONParseError,
    safe_json_parse,
    batch_json_parse,
    validate_json_structure
)


class TestJsonSafeParseFunction:
    """测试 JsonSafeParseFunction 类"""
    
    def setup_method(self):
        """设置测试方法"""
        self.parser = JsonSafeParseFunction()
    
    def test_execute_basic(self):
        """测试基本 JSON 解析"""
        # 测试简单对象
        json_str = '{"name": "Alice", "age": 30}'
        result = self.parser.execute(json_str)
        assert result == {"name": "Alice", "age": 30}
        
        # 测试数组
        json_str = '[1, 2, 3, "test"]'
        result = self.parser.execute(json_str)
        assert result == [1, 2, 3, "test"]
    
    def test_execute_wrapped_json(self):
        """测试包装的 JSON 解析"""
        # 测试 ```json``` 包装
        json_str = '```json\n{"key": "value"}\n```'
        result = self.parser.execute(json_str)
        assert result == {"key": "value"}
        
        # 测试 ``` 包装
        json_str = '```\n{"key": "value"}\n```'
        result = self.parser.execute(json_str)
        assert result == {"key": "value"}
        
        # 测试带额外空白的包装
        json_str = '   ```json   \n  {"key": "value"}  \n   ```   '
        result = self.parser.execute(json_str)
        assert result == {"key": "value"}
    
    def test_execute_with_context(self):
        """测试带上下文的解析"""
        json_str = '{"name": "Alice", "age": 30}'
        context = {"options": {"strict": False}}
        result = self.parser.execute(json_str, context)
        assert result == {"name": "Alice", "age": 30}
    
    def test_pure_function_property(self):
        """测试纯函数特性"""
        json_str = '{"test": "data"}'
        
        # 多次调用应返回相同结果
        result1 = self.parser.execute(json_str)
        result2 = self.parser.execute(json_str)
        assert result1 == result2
        
        # 不同实例应返回相同结果
        parser2 = JsonSafeParseFunction()
        result3 = parser2.execute(json_str)
        assert result1 == result3
    
    def test_invalid_input_type(self):
        """测试无效输入类型"""
        with pytest.raises(JSONParseError) as exc_info:
            self.parser.execute(123)  # 非字符串输入
        assert "Expected string input" in str(exc_info.value)
    
    def test_invalid_json(self):
        """测试无效 JSON"""
        with pytest.raises(JSONParseError) as exc_info:
            self.parser.execute('{"invalid": json}')  # 无效 JSON
        assert "JSON decode error" in str(exc_info.value)
    
    def test_empty_input(self):
        """测试空输入"""
        with pytest.raises(JSONParseError) as exc_info:
            self.parser.execute('')
        assert "Empty or invalid JSON content" in str(exc_info.value)
    
    def test_extract_json_from_wrapped_string(self):
        """测试 JSON 提取方法"""
        # 测试各种格式
        test_cases = [
            ('```json\n{"key": "value"}\n```', '{"key": "value"}'),
            ('```\n[1, 2, 3]\n```', '[1, 2, 3]'),
            ('{"key": "value"}', '{"key": "value"}'),
            ('[1, 2, 3]', '[1, 2, 3]'),
            ('   ```JSON   \n  {"test": true}  \n   ```   ', '{"test": true}'),
        ]
        
        for input_str, expected in test_cases:
            result = self.parser._extract_json_from_wrapped_string(input_str)
            assert result == expected
    
    def test_complex_nested_json(self):
        """测试复杂嵌套 JSON"""
        complex_json = {
            "users": [
                {"name": "Alice", "age": 30, "active": True},
                {"name": "Bob", "age": 25, "active": False}
            ],
            "metadata": {
                "total": 2,
                "timestamp": "2024-01-01T00:00:00Z"
            }
        }
        
        json_str = f'```json\n{json.dumps(complex_json, indent=2)}\n```'
        result = self.parser.execute(json_str)
        assert result == complex_json


class TestSafeJsonParse:
    """测试 safe_json_parse 便捷函数"""
    
    def test_basic_parsing(self):
        """测试基本解析功能"""
        result = safe_json_parse('{"name": "test"}')
        assert result == {"name": "test"}
    
    def test_wrapped_parsing(self):
        """测试包装格式解析"""
        result = safe_json_parse('```json\n{"wrapped": true}\n```')
        assert result == {"wrapped": True}
    
    def test_error_handling(self):
        """测试错误处理"""
        with pytest.raises(JSONParseError):
            safe_json_parse('invalid json')


class TestBatchJsonParse:
    """测试批量 JSON 解析功能"""
    
    def test_batch_parsing_success(self):
        """测试成功的批量解析"""
        json_strings = [
            '{"name": "Alice"}',
            '```json\n{"name": "Bob"}\n```',
            '[1, 2, 3]'
        ]
        
        results = batch_json_parse(json_strings)
        expected = [
            {"name": "Alice"},
            {"name": "Bob"},
            [1, 2, 3]
        ]
        assert results == expected
    
    def test_batch_parsing_error(self):
        """测试批量解析中的错误"""
        json_strings = [
            '{"name": "Alice"}',
            'invalid json',  # 这个会导致错误
            '[1, 2, 3]'
        ]
        
        with pytest.raises(JSONParseError) as exc_info:
            batch_json_parse(json_strings)
        assert "Failed to parse JSON at index 1" in str(exc_info.value)
    
    def test_invalid_input_type(self):
        """测试无效输入类型"""
        with pytest.raises(JSONParseError) as exc_info:
            batch_json_parse("not a list")
        assert "Expected list input" in str(exc_info.value)


class TestValidateJsonStructure:
    """测试 JSON 结构验证功能"""
    
    def test_validate_string_input(self):
        """测试字符串输入验证"""
        result = validate_json_structure('{"name": "Alice", "age": 30}')
        assert result['valid'] is True
        assert result['data'] == {"name": "Alice", "age": 30}
        assert len(result['errors']) == 0
    
    def test_validate_dict_input(self):
        """测试字典输入验证"""
        data = {"name": "Alice", "age": 30}
        result = validate_json_structure(data)
        assert result['valid'] is True
        assert result['data'] == data
        assert len(result['errors']) == 0
    
    def test_validate_with_required_keys(self):
        """测试必需字段验证"""
        context = {
            'validation_rules': {
                'required_keys': ['name', 'age', 'email']
            }
        }
        
        # 缺少字段的情况
        result = validate_json_structure('{"name": "Alice", "age": 30}', context)
        assert result['valid'] is False
        assert any("Missing required keys" in error for error in result['errors'])
        
        # 包含所有字段的情况
        result = validate_json_structure(
            '{"name": "Alice", "age": 30, "email": "alice@test.com"}', 
            context
        )
        assert result['valid'] is True
    
    def test_validate_expected_type(self):
        """测试期望类型验证"""
        # 期望对象但得到数组
        context = {'validation_rules': {'expected_type': 'object'}}
        result = validate_json_structure('[1, 2, 3]', context)
        assert result['valid'] is False
        assert any("Expected JSON object" in error for error in result['errors'])
        
        # 期望数组但得到对象
        context = {'validation_rules': {'expected_type': 'array'}}
        result = validate_json_structure('{"key": "value"}', context)
        assert result['valid'] is False
        assert any("Expected JSON array" in error for error in result['errors'])
    
    def test_validate_invalid_json(self):
        """测试无效 JSON 验证"""
        result = validate_json_structure('invalid json')
        assert result['valid'] is False
        assert result['data'] is None
        assert len(result['errors']) > 0
    
    def test_validate_non_json_types(self):
        """测试非 JSON 类型验证"""
        # 测试无法解析为 JSON 的字符串
        result = validate_json_structure("just a string")
        assert result['valid'] is False
        assert any("JSON decode error" in error for error in result['errors'])
        
        # 测试解析成功但不是 JSON 对象或数组的情况（如字符串、数字等）
        result = validate_json_structure('"just a string"')  # 这是有效的 JSON 字符串
        assert result['valid'] is False
        assert any("not a JSON object or array" in error for error in result['errors'])


class TestPipelineComposition:
    """测试管道组合能力"""
    
    def test_json_parse_in_pipeline(self):
        """测试 JSON 解析在管道中的使用"""
        from src.components.functions import create_simple_function, create_pipeline
        
        # 创建一个简单的数据转换函数
        def extract_name(data, context=None):
            if isinstance(data, dict) and 'name' in data:
                return data['name']
            raise ValueError("No name field found")
        
        # 创建管道
        parse_func = JsonSafeParseFunction()
        extract_func = create_simple_function("extract_name", "提取姓名", extract_name)
        pipeline = create_pipeline("parse_and_extract", "解析并提取", [parse_func, extract_func])
        
        # 测试管道执行
        json_str = '```json\n{"name": "Alice", "age": 30}\n```'
        result = pipeline(json_str)
        assert result == "Alice"
    
    def test_error_propagation_in_pipeline(self):
        """测试管道中的错误传播"""
        from src.components.functions import create_simple_function, create_pipeline
        
        def dummy_transform(data, context=None):
            return data
        
        parse_func = JsonSafeParseFunction()
        transform_func = create_simple_function("dummy", "虚拟转换", dummy_transform)
        pipeline = create_pipeline("parse_pipeline", "解析管道", [parse_func, transform_func])
        
        # 测试无效 JSON 在管道中的错误传播
        with pytest.raises(JSONParseError):
            pipeline('invalid json')


class TestContextHandling:
    """测试上下文处理"""
    
    def test_context_options(self):
        """测试上下文选项"""
        parser = JsonSafeParseFunction()
        
        # 测试严格模式
        context_strict = {'options': {'strict': True}}
        result = parser.execute('{"name": "Alice"}', context_strict)
        assert result == {"name": "Alice"}
        
        # 测试非严格模式
        context_non_strict = {'options': {'strict': False}}
        result = parser.execute('{"name": "Alice"}', context_non_strict)
        assert result == {"name": "Alice"}
    
    def test_context_preservation(self):
        """测试上下文保持"""
        # 验证函数不会修改传入的上下文
        original_context = {'options': {'strict': True}, 'other': 'data'}
        context_copy = original_context.copy()
        
        safe_json_parse('{"test": true}', context_copy)
        assert context_copy == original_context


if __name__ == "__main__":
    pytest.main([__file__])