DataOps-platform/test_parse_pic_fix.py

#!/usr/bin/env python3
"""
测试parse_pic.py的修复
验证batch_process_images函数能正确处理字典格式的输入
"""

import sys
import os
import logging

# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(funcName)s - %(lineno)d - %(message)s')

def test_batch_process_images_fix():
    """测试batch_process_images函数的修复"""
    print("测试batch_process_images函数的修复...")
    
    try:
        from app.core.data_parse.parse_pic import batch_process_images
        
        # 测试数据 - 模拟错误日志中的输入格式
        test_input = [
            {
                'minio_path': 'http://192.168.3.143:9000/dataops-bucket/misc_files/misc_file_20250801_212256_b61ddd94.png',
                'original_filename': '杂项表格样例.png',
                'status': '正常'
            }
        ]
        
        print(f"测试输入: {test_input}")
        
        # 调用函数（这里只是测试参数处理，不会真正处理图片）
        # 由于没有真实的图片文件，我们期望函数能正确处理参数格式而不抛出类型错误
        result = batch_process_images(test_input, process_type='table')
        
        print(f"函数执行结果: {result}")
        print("✓ batch_process_images函数修复验证通过")
        
    except TypeError as e:
        if "expected str, bytes or os.PathLike object, not dict" in str(e):
            print(f"❌ 仍然存在类型错误: {e}")
            return False
        else:
            print(f"其他类型错误: {e}")
            return False
    except Exception as e:
        print(f"其他错误: {e}")
        return False
    
    return True

def test_invalid_input_handling():
    """测试无效输入的处理"""
    print("\n测试无效输入处理...")
    
    try:
        from app.core.data_parse.parse_pic import batch_process_images
        
        # 测试各种无效输入
        invalid_inputs = [
            [{'minio_path': None}],  # minio_path为None
            [{'minio_path': {}}],    # minio_path为字典
            [{'minio_path': 123}],   # minio_path为数字
            [{'wrong_field': 'value'}],  # 缺少minio_path字段
            [{'minio_path': ''}],    # minio_path为空字符串
        ]
        
        for i, invalid_input in enumerate(invalid_inputs):
            print(f"测试无效输入 {i+1}: {invalid_input}")
            try:
                result = batch_process_images(invalid_input, process_type='table')
                print(f"  处理结果: {result.get('success', False)}")
                # 期望处理失败但不抛出类型错误
                if not result.get('success', False):
                    print(f"  ✓ 正确处理了无效输入")
                else:
                    print(f"  ⚠ 意外成功处理了无效输入")
            except TypeError as e:
                if "expected str, bytes or os.PathLike object, not dict" in str(e):
                    print(f"  ❌ 仍然存在类型错误: {e}")
                    return False
                else:
                    print(f"  ✓ 正确处理了类型错误: {e}")
            except Exception as e:
                print(f"  ✓ 正确处理了其他错误: {e}")
        
        print("✓ 无效输入处理验证通过")
        return True
        
    except Exception as e:
        print(f"❌ 测试失败: {e}")
        return False

if __name__ == "__main__":
    print("开始测试parse_pic.py的修复...")
    
    success = True
    
    # 测试基本功能
    if not test_batch_process_images_fix():
        success = False
    
    # 测试无效输入处理
    if not test_invalid_input_handling():
        success = False
    
    if success:
        print("\n🎉 所有测试通过！parse_pic.py修复成功。")
    else:
        print("\n❌ 部分测试失败，需要进一步修复。")
        sys.exit(1)