# -*- coding: utf-8 -*- """ CSV Pipeline 功能测试 测试内容： 1. 基础功能测试 2. 异常处理测试 3. 边界条件测试 4. 兼容性测试 Created on 2025-10-16 @author: 道长 @email: ctrlf4@yeah.net """ import csv import os import sys import shutil from pathlib import Path # 添加项目路径 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from feapder.pipelines.csv_pipeline import CsvPipeline class FunctionalityTester: """CSV Pipeline 功能测试器""" def __init__(self, test_dir="test_output"): """初始化测试器""" self.test_dir = test_dir self.pipeline = None self.passed = 0 self.failed = 0 def setup(self): """测试前准备""" if os.path.exists(self.test_dir): shutil.rmtree(self.test_dir) os.makedirs(self.test_dir, exist_ok=True) csv_dir = os.path.join(self.test_dir, "csv") self.pipeline = CsvPipeline(csv_dir=csv_dir) print(f"✅ 测试环境准备完成") def teardown(self): """测试后清理""" if self.pipeline: self.pipeline.close() def assert_true(self, condition, message): """断言真""" if condition: print(f" ✅ {message}") self.passed += 1 else: print(f" ❌ {message}") self.failed += 1 def assert_false(self, condition, message): """断言假""" self.assert_true(not condition, message) def assert_equal(self, actual, expected, message): """断言相等""" if actual == expected: print(f" ✅ {message}") self.passed += 1 else: print(f" ❌ {message} (期望: {expected}, 实际: {actual})") self.failed += 1 def test_basic_save(self): """测试基础保存功能""" print("\n" + "=" * 80) print("测试 1: 基础保存功能") print("=" * 80) # 测试保存单条数据 item = {"id": 1, "name": "Test Product", "price": 99.99} result = self.pipeline.save_items("product", [item]) self.assert_true(result, "保存单条数据") # 检查文件是否创建 csv_file = os.path.join(self.pipeline.csv_dir, "product.csv") self.assert_true(os.path.exists(csv_file), "CSV 文件已创建") # 检查数据是否正确 with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) self.assert_equal(len(rows), 1, "文件中有 1 条数据") if rows: self.assert_equal(rows[0]["id"], "1", "数据 ID 正确") self.assert_equal(rows[0]["name"], "Test Product", "数据名称正确") def test_batch_save(self): """测试批量保存""" print("\n" + "=" * 80) print("测试 2: 批量保存功能") print("=" * 80) # 生成测试数据 items = [] for i in range(10): items.append({ "id": i + 1, "name": f"Product_{i + 1}", "price": 100 + i, }) result = self.pipeline.save_items("batch_test", items) self.assert_true(result, "批量保存 10 条数据") # 检查数据行数 csv_file = os.path.join(self.pipeline.csv_dir, "batch_test.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) self.assert_equal(len(rows), 10, "批量保存数据行数正确") def test_empty_items(self): """测试空数据处理""" print("\n" + "=" * 80) print("测试 3: 空数据处理") print("=" * 80) result = self.pipeline.save_items("empty_test", []) self.assert_true(result, "空数据列表返回 True") def test_special_characters(self): """测试特殊字符处理""" print("\n" + "=" * 80) print("测试 4: 特殊字符处理") print("=" * 80) items = [ { "id": 1, "name": "产品名称", "description": 'Contains "quotes" and, commas', "emoji": "😀🎉🚀", "newline": "Line1\nLine2", } ] result = self.pipeline.save_items("special_chars", items) self.assert_true(result, "保存包含特殊字符的数据") # 读取并检查 csv_file = os.path.join(self.pipeline.csv_dir, "special_chars.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) if rows: self.assert_equal(rows[0]["name"], "产品名称", "中文字符正确") self.assert_equal( rows[0].get("emoji", ""), "😀🎉🚀", "Emoji 正确" ) def test_multiple_tables(self): """测试多表存储""" print("\n" + "=" * 80) print("测试 5: 多表存储") print("=" * 80) tables = ["product", "user", "order"] for table in tables: item = {"id": 1, "name": f"Test {table}"} result = self.pipeline.save_items(table, [item]) self.assert_true(result, f"保存到表 {table}") # 检查所有文件 for table in tables: csv_file = os.path.join(self.pipeline.csv_dir, f"{table}.csv") self.assert_true(os.path.exists(csv_file), f"表 {table} 的 CSV 文件存在") def test_header_only_once(self): """测试表头只写一次""" print("\n" + "=" * 80) print("测试 6: 表头只写一次") print("=" * 80) table = "header_test" # 第一次写入 items1 = [{"id": 1, "name": "Product 1"}] self.pipeline.save_items(table, items1) # 第二次写入 items2 = [{"id": 2, "name": "Product 2"}] self.pipeline.save_items(table, items2) # 检查表头行数 csv_file = os.path.join(self.pipeline.csv_dir, f"{table}.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: lines = f.readlines() # 应该是：1 个表头 + 2 条数据 self.assert_equal(len(lines), 3, "文件中只有 1 行表头和 2 行数据") def test_numeric_values(self): """测试数值类型""" print("\n" + "=" * 80) print("测试 7: 数值类型处理") print("=" * 80) items = [ { "id": 1, "price": 99.99, "stock": 100, "rating": 4.5, "active": True, } ] result = self.pipeline.save_items("numeric_test", items) self.assert_true(result, "保存包含各类数值的数据") # 读取并检查 csv_file = os.path.join(self.pipeline.csv_dir, "numeric_test.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) if rows: self.assert_equal(rows[0]["price"], "99.99", "浮点数正确") self.assert_equal(rows[0]["stock"], "100", "整数正确") self.assert_equal(rows[0]["rating"], "4.5", "小数正确") def test_large_values(self): """测试大值处理""" print("\n" + "=" * 80) print("测试 8: 大值处理") print("=" * 80) large_text = "x" * 10000 # 10KB 的文本 items = [ { "id": 1, "name": "Large Content", "content": large_text, } ] result = self.pipeline.save_items("large_test", items) self.assert_true(result, "保存大内容数据") # 检查数据完整性 csv_file = os.path.join(self.pipeline.csv_dir, "large_test.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) if rows: self.assert_equal( len(rows[0]["content"]), len(large_text), "大内容数据完整" ) def test_update_items_fallback(self): """测试 update_items 降级为 save""" print("\n" + "=" * 80) print("测试 9: update_items 降级为 save") print("=" * 80) items = [{"id": 1, "name": "Product 1", "price": 100}] result = self.pipeline.update_items("update_test", items, ("price",)) self.assert_true(result, "update_items 返回 True") # 检查数据是否存在 csv_file = os.path.join(self.pipeline.csv_dir, "update_test.csv") self.assert_true(os.path.exists(csv_file), "update_items 创建了 CSV 文件") def test_file_operations(self): """测试文件操作""" print("\n" + "=" * 80) print("测试 10: 文件操作") print("=" * 80) items = [{"id": 1, "name": "Test"}] table = "file_test" result = self.pipeline.save_items(table, items) self.assert_true(result, "保存数据") csv_file = os.path.join(self.pipeline.csv_dir, f"{table}.csv") # 检查文件是否可读 try: with open(csv_file, 'r', encoding='utf-8') as f: f.read() self.assert_true(True, "CSV 文件可读") except Exception as e: self.assert_true(False, f"CSV 文件可读 ({e})") # 检查文件大小 file_size = os.path.getsize(csv_file) self.assert_true(file_size > 0, f"CSV 文件大小 > 0 ({file_size} 字节)") def test_concurrent_same_table(self): """测试同表并发写入""" print("\n" + "=" * 80) print("测试 11: 同表并发写入（Per-Table Lock）") print("=" * 80) import threading table = "concurrent_same_table" errors = [] def write_data(thread_id): try: items = [{"id": thread_id, "name": f"Item_{thread_id}"}] result = self.pipeline.save_items(table, items) if not result: errors.append(f"线程{thread_id}写入失败") except Exception as e: errors.append(f"线程{thread_id}异常: {e}") # 创建多个线程 threads = [] for i in range(5): t = threading.Thread(target=write_data, args=(i,)) t.start() threads.append(t) # 等待所有线程完成 for t in threads: t.join() self.assert_equal(len(errors), 0, "并发写入无错误") # 检查数据完整性 csv_file = os.path.join(self.pipeline.csv_dir, f"{table}.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) self.assert_true(len(rows) > 0, "并发写入产生了数据") def test_directory_creation(self): """测试目录自动创建""" print("\n" + "=" * 80) print("测试 12: 目录自动创建") print("=" * 80) # 创建新的 pipeline 实例，指定不存在的目录 new_csv_dir = os.path.join(self.test_dir, "new_csv_dir") self.assert_false(os.path.exists(new_csv_dir), "新目录不存在") new_pipeline = CsvPipeline(csv_dir=new_csv_dir) self.assert_true(os.path.exists(new_csv_dir), "目录自动创建") new_pipeline.close() def test_none_values(self): """测试 None 值处理""" print("\n" + "=" * 80) print("测试 13: None 值处理") print("=" * 80) items = [ { "id": 1, "name": "Product", "description": None, "optional_field": "", } ] result = self.pipeline.save_items("none_test", items) self.assert_true(result, "保存包含 None 值的数据") # 检查文件 csv_file = os.path.join(self.pipeline.csv_dir, "none_test.csv") with open(csv_file, 'r', encoding='utf-8', newline='') as f: reader = csv.DictReader(f) rows = list(reader) if rows: # None 会被转换为字符串 "None" self.assert_true("None" in rows[0]["description"], "None 值被正确处理") def run_all_tests(self): """运行所有测试""" print("\n") print("╔" + "═" * 78 + "╗") print("║" + " CSV Pipeline 功能测试 ".center(78) + "║") print("║" + " 作者: 道长 | 日期: 2025-10-16 ".center(78) + "║") print("╚" + "═" * 78 + "╝") try: self.setup() # 运行所有测试 self.test_basic_save() self.test_batch_save() self.test_empty_items() self.test_special_characters() self.test_multiple_tables() self.test_header_only_once() self.test_numeric_values() self.test_large_values() self.test_update_items_fallback() self.test_file_operations() self.test_concurrent_same_table() self.test_directory_creation() self.test_none_values() # 打印总结 self.print_summary() return self.failed == 0 except Exception as e: print(f"\n❌ 测试过程中出错: {e}") import traceback traceback.print_exc() return False finally: self.teardown() def print_summary(self): """打印测试总结""" print("\n" + "=" * 80) print("测试总结") print("=" * 80) print(f"✅ 通过: {self.passed}") print(f"❌ 失败: {self.failed}") print(f"总计: {self.passed + self.failed}") if self.failed == 0: print("\n🎉 所有测试通过！") else: print(f"\n⚠️ 有 {self.failed} 个测试失败") print("=" * 80) def main(): """主函数""" tester = FunctionalityTester(test_dir="tests/test_csv_pipeline/test_output_func") success = tester.run_all_tests() return 0 if success else 1 if __name__ == "__main__": sys.exit(main())