通用文件读取

test/日志测试.py

@@ -14,14 +14,43 @@
 
 
 # test_log_rotation.py
-from utils.logger import log
-import time
+# from utils.logger import log
+# import time
+#
+# def generate_large_log():
+#     """快速生成超过20MB的测试日志"""
+#     for i in range(10000):
+#         log.info(f"测试日志填充数据... {i}" * 10)
+#         time.sleep(0.001)  # 避免内存暴涨
+#
+# if __name__ == "__main__":
+#     generate_large_log()
 
-def generate_large_log():
-    """快速生成超过20MB的测试日志"""
-    for i in range(10000):
-        log.info(f"测试日志填充数据... {i}" * 10)
-        time.sleep(0.001)  # 避免内存暴涨
+# 使用方法
+# my_module/main_class.py
+from utils.logger import log
+
+class MainProcessor:
+    def __init__(self):
+        self.log = log.bind(module=self.__class__.__name__)  # 动态绑定类名
+
+    def main(self):
+        """主执行方法"""
+        self.log.info("开始执行主流程")
+        try:
+            self._step1()
+            # self._step2()
+        except Exception as e:
+            self.log.error("主流程执行失败", exc_info=e)
+            raise
+
+    def _step1(self):
+        """子方法示例"""
+        self.log.debug("执行步骤1: 初始化资源")
+        # ...业务逻辑...
+        resource_count = 10
+        self.log.info("步骤1完成 | created={}", resource_count)
 
 if __name__ == "__main__":
-    generate_large_log()
+    processor = MainProcessor()
+    processor.main()

test/通用文件读取测试.py

@@ -0,0 +1,91 @@
+import pytest
+import pandas as pd
+import os
+from pathlib import Path
+from utils.file_handler import FileHandler
+
+@pytest.fixture
+def temp_dir(tmp_path):
+    """创建临时测试目录"""
+    test_dir = tmp_path / "test_files"
+    test_dir.mkdir()
+    return test_dir
+
+@pytest.fixture
+def file_handler(temp_dir):
+    """创建FileHandler实例"""
+    return FileHandler(temp_dir)
+
+@pytest.fixture
+def sample_dataframe():
+    """创建测试用DataFrame"""
+    return pd.DataFrame({
+        'id': [1, 2, 3],
+        'name': ['Alice', 'Bob', 'Charlie'],
+        'value': [10.5, 20.3, 30.1]
+    })
+
+@pytest.fixture
+def sample_text_file(temp_dir):
+    """创建测试文本文件"""
+    file_path = temp_dir / "test.txt"
+    with open(file_path, 'w') as f:
+        f.write("line1\nline2\nline3")
+    return file_path
+
+
+# 开始测试
+
+def test_read_write_csv(file_handler, temp_dir, sample_dataframe):
+    """测试CSV文件读写"""
+    test_file = temp_dir / "test.csv"
+
+    # 测试写入
+    write_result = file_handler.write_file(test_file, sample_dataframe)
+
+    # 修改断言方式
+    assert bool(write_result.iloc[0]['success']) == True  # 使用bool()转换
+    # 或者
+    assert write_result.iloc[0]['success'] == True  # 使用值比较
+
+    assert os.path.exists(test_file)
+
+    # 测试读取
+    df = file_handler.read_file(test_file)
+    assert df.shape == (3, 3)
+    assert list(df.columns) == ['id', 'name', 'value']
+
+def test_read_write_json(file_handler, temp_dir, sample_dataframe):
+    """测试JSON文件读写"""
+    test_file = temp_dir / "test.json"
+
+    # 测试写入
+    write_result = file_handler.write_file(test_file, sample_dataframe)
+    assert write_result.iloc[0]['success'] ==  True
+
+    # 测试读取
+    df = file_handler.read_file(test_file)
+    assert df.shape == (3, 3)
+
+def test_read_write_excel(file_handler, temp_dir, sample_dataframe):
+    """测试Excel文件读写"""
+    test_file = temp_dir / "test.xlsx"
+
+    # 测试写入
+    write_result = file_handler.write_file(test_file, sample_dataframe)
+    assert write_result.iloc[0]['success'] ==  True
+
+    # 测试读取
+    df = file_handler.read_file(test_file)
+    assert df.shape == (3, 3)
+
+def test_read_write_text(file_handler, sample_text_file):
+    """测试文本文件读写"""
+    # 测试读取
+    df = file_handler.read_file(sample_text_file)
+    assert df.shape == (1, 1)  # 默认单行读取
+
+    # 测试按行读取
+    lines_df = file_handler.read_lines(sample_text_file)
+    assert lines_df.shape == (3, 1)
+    assert lines_df.iloc[0, 0] == "line1"