"""清理逻辑回归测试(不依赖 pytest,且**不触碰真实目录**) python tests/test_cleanup.py 做法:用 XWLB_CONFIG_FILE 指向临时 config.yml,把 paths 指向临时目录, 验证 cleanup 段各开关的组合行为。 """ import os import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) def build_env(tmp, after_daily_run=1, only_on_success=1, remove_video=1, remove_audio=1): """在临时目录生成 config.yml,并让 config 模块加载它;返回 (video_dir, audio_dir)""" video_dir = Path(tmp) / 'video' audio_dir = Path(tmp) / 'audio' video_dir.mkdir(parents=True, exist_ok=True) audio_dir.mkdir(parents=True, exist_ok=True) for name in ('a.mp3', 'b.mp4', 'keep.txt'): (video_dir / name).write_bytes(b'x' * 10) for name in ('c.wav', 'd.wav', 'keep.txt'): (audio_dir / name).write_bytes(b'x' * 10) cfg = Path(tmp) / 'config.yml' cfg.write_text(f""" paths: video_dir: {video_dir} audio_dir: {audio_dir} cleanup: after_daily_run: {after_daily_run} only_on_success: {only_on_success} remove_video: {remove_video} remove_audio: {remove_audio} """, encoding='utf-8') os.environ['XWLB_CONFIG_FILE'] = str(cfg) import config config.reload_config() return video_dir, audio_dir def snapshot(video_dir, audio_dir): return sorted(p.name for p in list(video_dir.iterdir()) + list(audio_dir.iterdir())) def main(): failed = 0 def check(desc, got, want): nonlocal failed if got == want: print(f" ✓ {desc}: {got}") else: failed += 1 print(f" ✗ {desc}: 期望 {want},实际 {got}") import cleanup # 1) 成功 + 开关全开 → 删除 mp3/mp4/wav,保留其他文件 with tempfile.TemporaryDirectory() as tmp: video_dir, audio_dir = build_env(tmp) result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=True) check('成功时删除 4 个中间产物', result['removed'], 4) check('非中间产物保留', snapshot(video_dir, audio_dir), ['keep.txt', 'keep.txt']) check('清理字节数>0', result['bytes'] > 0, True) # 2) 失败 + only_on_success=1(默认)→ 保留文件便于重跑 with tempfile.TemporaryDirectory() as tmp: video_dir, audio_dir = build_env(tmp, only_on_success=1) result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=False) check('失败时保留文件', result['skipped'], True) check('文件未被删除', len(snapshot(video_dir, audio_dir)), 6) # 3) 失败 + only_on_success=0 → 仍然清理 with tempfile.TemporaryDirectory() as tmp: video_dir, audio_dir = build_env(tmp, only_on_success=0) result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=False) check('only_on_success=0 时失败也清理', result['removed'], 4) # 4) after_daily_run=0 → 完全不动 with tempfile.TemporaryDirectory() as tmp: video_dir, audio_dir = build_env(tmp, after_daily_run=0) result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=True) check('after_daily_run=0 时跳过', result['skipped'], True) check('文件未被删除', len(snapshot(video_dir, audio_dir)), 6) # 5) 只关 wav 清理 → 只删 mp3/mp4 with tempfile.TemporaryDirectory() as tmp: video_dir, audio_dir = build_env(tmp, remove_audio=0) result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=True) check('remove_audio=0 时只删视频', result['removed'], 2) check('wav 保留、mp3/mp4 已删', snapshot(video_dir, audio_dir), ['c.wav', 'd.wav', 'keep.txt', 'keep.txt']) os.environ.pop('XWLB_CONFIG_FILE', None) import config config.reload_config() print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}") return 1 if failed else 0 if __name__ == '__main__': sys.exit(main())