# -*- coding: utf-8 -*- """核心层单元测试：数据库 CRUD、过滤、导入导出往返。 运行: python -m unittest discover tests """ import json import shutil import tempfile import unittest from pathlib import Path from comic_asset_manager.db import Database from comic_asset_manager.library import ( ImportReport, import_folder, import_library, import_file, export_library, scan_files, ) from comic_asset_manager.models import Asset, AssetType, Chapter, Series from comic_asset_manager import thumbnail class DBTestBase(unittest.TestCase): def setUp(self): self.tmp = Path(tempfile.mkdtemp(prefix="cam_test_")) self.db = Database(self.tmp / "lib.db") def tearDown(self): self.db.close() shutil.rmtree(self.tmp, ignore_errors=True) class SeriesChapterTest(DBTestBase): def test_create_and_list_series(self): s = self.db.create_series(Series(name="星际旅人", author="张三", status="ongoing")) self.assertIsNotNone(s.id) self.assertEqual(s.status_label, "连载中") rows = self.db.list_series() self.assertEqual(len(rows), 1) self.assertEqual(rows[0].name, "星际旅人") def test_update_series(self): s = self.db.create_series(Series(name="A")) s.status = "complete" s.author = "李四" self.db.update_series(s) loaded = self.db.get_series(s.id) self.assertEqual(loaded.status, "complete") self.assertEqual(loaded.author, "李四") def test_delete_series_cascades_chapters(self): s = self.db.create_series(Series(name="A")) c = self.db.create_chapter(Chapter(series_id=s.id, number=1, title="序章")) self.db.delete_series(s.id) self.assertIsNone(self.db.get_series(s.id)) self.assertEqual(self.db.list_chapters(series_id=s.id), []) def test_chapter_label(self): self.assertEqual(Chapter(volume=3, number=12).label, "第3卷 · 第12话") self.assertEqual(Chapter(number=5).label, "第5话") self.assertEqual(Chapter(title="番外").label, "番外") class AssetTest(DBTestBase): def _mk_series(self): return self.db.create_series(Series(name="测试作品")) def test_create_asset_requires_title(self): with self.assertRaises(ValueError): self.db.create_asset(Asset(title=" ", type=AssetType.COMIC_PAGE, file_path="/tmp/x.png")) def test_crud_asset(self): s = self._mk_series() a = self.db.create_asset(Asset( series_id=s.id, type=AssetType.COMIC_PAGE, title="第1页", file_path="/tmp/p1.png", tags=["战斗", "热血"], page_index=1, )) self.assertIsNotNone(a.id) self.assertIn("战斗", a.tags) a.title = "第1页-修改" a.description = "开场战斗" a.tags = ["战斗", "热血", "主角"] self.db.update_asset(a) got = self.db.get_asset(a.id) self.assertEqual(got.title, "第1页-修改") self.assertEqual(got.description, "开场战斗") self.assertIn("主角", got.tags) self.assertIsInstance(got.type, AssetType) def test_filter_by_type_and_search(self): s = self._mk_series() self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title="01 封面", file_path="/tmp/a")) self.db.create_asset(Asset(series_id=s.id, type=AssetType.CHARACTER, title="艾伦·王", file_path="/tmp/b")) self.db.create_asset(Asset(series_id=s.id, type=AssetType.AUDIO, title="主题曲", file_path="/tmp/c")) pages = self.db.list_assets(asset_type=AssetType.COMIC_PAGE) self.assertEqual(len(pages), 1) self.assertEqual(pages[0].title, "01 封面") found = self.db.list_assets(search="王") self.assertEqual(len(found), 1) self.assertEqual(found[0].title, "艾伦·王") # 搜索内容 self.db.create_asset(Asset(series_id=s.id, type=AssetType.SCRIPT, title="剧本1", content="这是热血战斗剧情")) found = self.db.list_assets(search="热血") self.assertTrue(any("热血" in (x.content or "") or "热血" in (x.title or "") for x in found)) def test_filter_by_tag(self): s = self._mk_series() self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title="A", tags=["boss"])) self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title="B", tags=["日常"])) rows = self.db.list_assets(tag="boss") self.assertEqual(len(rows), 1) self.assertEqual(rows[0].title, "A") def test_pagination(self): s = self._mk_series() for i in range(25): self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title=f"p{i:02d}", file_path="/tmp/x", page_index=i)) self.assertEqual(self.db.count_assets(), 25) page1 = self.db.list_assets(limit=10, offset=0, order_by="id", reverse=False) page2 = self.db.list_assets(limit=10, offset=10, order_by="id", reverse=False) self.assertEqual(len(page1), 10) self.assertEqual(len(page2), 10) self.assertNotEqual(page1[0].id, page2[0].id) def test_delete_asset(self): s = self._mk_series() a = self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title="X", file_path="/tmp/x")) self.db.delete_asset(a.id) self.assertIsNone(self.db.get_asset(a.id)) self.assertEqual(self.db.count_assets(), 0) def test_order_by_whitelist(self): """非白名单列不应被当列名注入。""" s = self._mk_series() self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title="a", file_path="/tmp")) rows = self.db.list_assets(order_by="title; DROP TABLE assets") self.assertEqual(len(rows), 1) self.assertEqual(self.db.count_assets(), 1) # 表还在 class TagTest(DBTestBase): def test_tag_dedupe(self): t1 = self.db.create_tag("战斗", "#E74C3C") t2 = self.db.create_tag("战斗", "#00FF00") self.assertEqual(t1.id, t2.id) # 同名复用 self.assertEqual(len(self.db.list_tags()), 1) def test_empty_name_rejected(self): with self.assertRaises(ValueError): self.db.create_tag(" ") def test_asset_lazy_registers_tag(self): s = self.db.create_series(Series(name="S")) self.db.create_asset(Asset(series_id=s.id, type=AssetType.COMIC_PAGE, title="x", file_path="/tmp", tags=["新标签", ""])) names = {t.name for t in self.db.list_tags()} self.assertIn("新标签", names) self.assertNotIn("", names) class BackupExportTest(DBTestBase): def _seed(self): s = self.db.create_series(Series(name="库1", author="A")) c = self.db.create_chapter(Chapter(series_id=s.id, number=1, title="序章")) self.db.create_tag("测试") self.db.create_asset(Asset(series_id=s.id, chapter_id=c.id, type=AssetType.COMIC_PAGE, title="页1", file_path="/tmp/x.png", tags=["测试"], page_index=1)) self.db.create_asset(Asset(series_id=s.id, type=AssetType.SCRIPT, title="剧本", content="正文内容")) def test_backup_produces_file(self): self._seed() out = self.db.backup(self.tmp / "backup") self.assertTrue(Path(out).exists()) self.assertGreater(Path(out).stat().st_size, 0) def test_export_import_roundtrip(self): self._seed() jpath = export_library(self.db, self.tmp / "export.json") self.assertTrue(Path(jpath).exists()) with open(jpath, encoding="utf-8") as f: data = json.load(f) self.assertEqual(len(data["series"]), 1) self.assertEqual(len(data["chapters"]), 1) self.assertEqual(len(data["assets"]), 2) # 导入到第二个库 db2 = Database(self.tmp / "lib2.db") try: report = import_library(db2, jpath) self.assertEqual(report.imported, 2) self.assertEqual(db2.count_assets(), 2) self.assertEqual(len(db2.list_series()), 1) self.assertEqual(len(db2.list_chapters()), 1) # 内容与标签保留 got = db2.list_assets(asset_type=AssetType.SCRIPT)[0] self.assertEqual(got.content, "正文内容") self.assertIn("测试", db2.list_assets(asset_type=AssetType.COMIC_PAGE)[0].tags) finally: db2.close() class ScanTest(unittest.TestCase): def test_scan_and_import_folder(self): tmp = Path(tempfile.mkdtemp(prefix="cam_scan_")) try: d = tmp / "ep1" d.mkdir() for i in (1, 2, 10, 11): # 故意乱序，验证自然排序 (d / f"page{i:02d}.png").write_bytes(b"x") db = Database(tmp / "lib.db") try: found = scan_files(d) self.assertEqual(len(found), 4) self.assertEqual(found[0][1], AssetType.COMIC_PAGE) report = import_folder(db, d, series_name="某作", copy_files=False) self.assertEqual(report.imported, 4) pages = db.list_assets(order_by="page_index", reverse=False) idxs = [p.page_index for p in pages] self.assertEqual(idxs, sorted(idxs)) # 自然排序 # page 序号来自文件名末位数字 self.assertEqual([p.page_index for p in pages], [1, 2, 10, 11]) finally: db.close() finally: shutil.rmtree(tmp, ignore_errors=True) def test_import_file_missing(self): tmp = Path(tempfile.mkdtemp()) try: db = Database(tmp / "lib.db") try: self.assertIsNone(import_file(db, tmp / "nope.png")) finally: db.close() finally: shutil.rmtree(tmp, ignore_errors=True) class ThumbnailTest(unittest.TestCase): def test_thumb_generate_and_cache(self): tmp = Path(tempfile.mkdtemp()) try: from PIL import Image src = tmp / "big.png" Image.new("RGB", (1200, 800), (200, 30, 30)).save(src) out = thumbnail.make_thumb(src, 42, tmp) self.assertIsNotNone(out) self.assertTrue(Path(out).exists()) with Image.open(out) as im: self.assertLessEqual(max(im.size), thumbnail.THUMB_MAX) # 二次调用命中缓存 m1 = Path(out).stat().st_mtime out2 = thumbnail.make_thumb(src, 42, tmp) self.assertEqual(out, out2) self.assertEqual(Path(out2).stat().st_mtime, m1) finally: shutil.rmtree(tmp, ignore_errors=True) def test_thumb_rejects_missing(self): tmp = Path(tempfile.mkdtemp()) try: self.assertIsNone(thumbnail.make_thumb(tmp / "nope.png", 1, tmp)) self.assertIsNone(thumbnail.audio_duration(tmp / "nope.wav")) finally: shutil.rmtree(tmp, ignore_errors=True) def test_describe_file(self): self.assertEqual(thumbnail.describe_file("/no/such/file"), {"status": "文件不存在"}) class ImportReportTest(unittest.TestCase): def test_defaults(self): r = ImportReport() self.assertFalse(r.ok) self.assertEqual(r.imported, 0) self.assertEqual(r.failed, []) r2 = ImportReport(imported=3) self.assertTrue(r2.ok) class TextContentTest(unittest.TestCase): """文本类资产导入时必须提取正文（回归测试）。""" def _run(self, copy_files): tmp = Path(tempfile.mkdtemp(prefix="cam_text_")) try: txt = tmp / "剧本.txt" txt.write_text("# 第1话\n\n**船长**：出发。", encoding="utf-8") db = Database(tmp / "lib.db") try: a = import_file(db, txt, asset_type=AssetType.SCRIPT, copy_files=copy_files) self.assertIsNotNone(a) self.assertEqual(a.type, AssetType.SCRIPT) self.assertIn("船长", a.content or "") # 数据库持久化后可读回 got = db.get_asset(a.id) self.assertIn("船长", got.content or "") finally: db.close() finally: shutil.rmtree(tmp, ignore_errors=True) def test_import_file_copies(self): self._run(copy_files=True) def test_import_file_links(self): self._run(copy_files=False) def test_import_folder_reads_content(self): """批量导入模式下文本正文也应提取。""" tmp = Path(tempfile.mkdtemp(prefix="cam_text2_")) try: d = tmp / "scripts" d.mkdir() (d / "第1话.txt").write_text("场景一：舰桥", encoding="utf-8") db = Database(tmp / "lib.db") try: rep = import_folder(db, d, series_name="T") self.assertEqual(rep.imported, 1) a = db.list_assets()[0] self.assertEqual(a.type, AssetType.SCRIPT) self.assertIn("舰桥", a.content or "") finally: db.close() finally: shutil.rmtree(tmp, ignore_errors=True) def test_binary_ext_not_read_as_text(self): """非文本扩展名即使内容像文本也不读取。""" tmp = Path(tempfile.mkdtemp(prefix="cam_text3_")) try: p = tmp / "fake.txt.exe" p.write_text("not really text", encoding="utf-8") db = Database(tmp / "lib.db") try: a = import_file(db, p, asset_type=AssetType.SCRIPT, copy_files=False) self.assertEqual(a.content, "") finally: db.close() finally: shutil.rmtree(tmp, ignore_errors=True) if __name__ == "__main__": unittest.main(verbosity=2)