tests/test_record_rehome.py
Rama master · réplica de lectura del repositorio original.
"""已消失行的个人记录接到新文件（ADR-0087 决策二）。
登记那一侧走真的 `scan.scan_location` 与 `scan.ingest_path`：全量扫描与推送发现是两条
登记路径，接回要在两条上都发生。数据库路径前缀使用 `R:\\media`，挂载表指向临时目录，支持跨平台验证。
"""
from __future__ import annotations
import importlib.util
import io
import json
import sqlite3
import tempfile
import unittest
from contextlib import closing, redirect_stdout
from pathlib import Path
from peach import personal_records, record_rehome, scan
from peach.migrations import upgrade
ROOT = Path(__file__).resolve().parents[1]
DECLARED = {"local": ("R:\\media",)}
class RecordRehomeTests(unittest.TestCase):
def test_follow_reference_moves_with_the_asset_and_reverts_with_its_batch(self):
self.add_vanished(7, r'R:\media\old\EXM-001.mp4', code='EXM-001', records=False)
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("INSERT INTO asset(id,location,path,name,medium) "
"VALUES(8,'local','new.mp4','new.mp4','video')")
connection.execute("INSERT INTO follow_source(id,provider,ref,label,url,created_at,updated_at) "
"VALUES(1,'test','sample','sample','https://example.test/','now','now')")
connection.execute("INSERT INTO follow_item(id,source_id,external_id,title,release_key,asset_id,"
"first_seen_at,last_seen_at) VALUES(1,1,'item','item','item',7,'now','now')")
baseline = connection.execute('PRAGMA foreign_key_check').fetchall()
batch = record_rehome.move(connection, 7, 8, source='test', rule='same-release')
self.assertEqual(connection.execute('SELECT asset_id FROM follow_item').fetchall(), [(8,)])
self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(), baseline)
record_rehome.revert_one(connection, batch)
self.assertEqual(connection.execute('SELECT asset_id FROM follow_item').fetchall(), [(7,)])
self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(), baseline)
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.db = self.root / "ledger.db"
upgrade(self.db, ROOT / "migrations")
self.media = self.root / "media"
self.media.mkdir()
self.mounts = {"local": (str(self.media),)}
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("INSERT INTO playlist(id,profile_id,name,current_asset_id,created_at,updated_at) "
"VALUES(1,'default','收藏',NULL,'2026-01-01','2026-01-01')")
connection.commit()
def add_vanished(self, asset_id, path, *, code=None, creator=None, duration=None, records=True):
"""一条已消失的旧行；`records` 时带上六张表里的记录，各一条。"""
with closing(sqlite3.connect(self.db)) as connection:
connection.execute(
"INSERT INTO asset(id,location,path,name,medium,size,code,creator,duration,rating,disposal) "
"VALUES(?,'local',?,?,'video',10,?,?,?,4,'vanished')",
(asset_id, path, path.rsplit("\\", 1)[-1], code, creator, duration))
connection.execute("INSERT INTO asset_tag(asset_id,tag,source) VALUES(?,'旧标签','name')", (asset_id,))
if records:
self.add_records(connection, asset_id)
connection.commit()
@staticmethod
def add_records(connection, asset_id, *, liked=1):
connection.execute("INSERT INTO asset_preference(profile_id,asset_id,liked,reason,updated_at) "
"VALUES('default',?,?,'旧的','2026-01-01')", (asset_id, liked))
connection.execute("INSERT INTO watch_queue(profile_id,asset_id,added_at) VALUES('default',?,'2026-01-01')",
(asset_id,))
connection.execute("INSERT INTO playlist_item(playlist_id,asset_id,position,added_at) "
"VALUES(1,?,?,'2026-01-01')", (asset_id, asset_id))
connection.execute("UPDATE playlist SET current_asset_id=? WHERE id=1", (asset_id,))
connection.execute("INSERT INTO asset_quality_goal(profile_id,asset_id,wanted,reason,updated_at) "
"VALUES('default',?,1,'要高清','2026-01-01')", (asset_id,))
connection.execute("INSERT INTO activity_event(asset_id,profile_id,kind,occurred_at,position_seconds,source) "
"VALUES(?,'default','climax','2026-01-01',61.5,'web')", (asset_id,))
connection.execute("INSERT INTO asset_tag_preference(profile_id,asset_id,normalized_tag,hidden,updated_at) "
"VALUES('default',?,'旧标签',1,'2026-01-01')", (asset_id,))
def put(self, relative: str) -> str:
path = self.media.joinpath(*relative.split("/"))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"0" * 10)
return "R:\\media\\" + relative.replace("/", "\\")
def scan_all(self):
with redirect_stdout(io.StringIO()):
return scan.scan_location(self.db, "local", "R:\\media", declared_roots=DECLARED,
mounts=self.mounts, windows=False)
def ingest(self, path):
return scan.ingest_path(self.db, "local", path, declared_roots=DECLARED, mounts=self.mounts,
windows=False)
def query(self, sql, params=()):
with closing(sqlite3.connect(self.db)) as connection:
return connection.execute(sql, params).fetchall()
def owner(self, table):
return sorted({row[0] for row in self.query(f"SELECT asset_id FROM {table}")})
def id_of(self, path):
return self.query("SELECT id FROM asset WHERE path=?", (path,))[0][0]
def test_a_file_back_at_its_path_clears_vanished_on_the_same_row(self):
path = self.put("ABC-123.mp4")
self.add_vanished(7, path, code="ABC-123")
result = self.scan_all()
self.assertEqual(self.query("SELECT id,disposal FROM asset"), [(7, None)])
self.assertEqual(result.new_ids, ())
self.assertEqual(self.query("SELECT count(*) FROM record_rehome"), [(0,)])
def test_a_full_scan_moves_all_records_to_the_one_new_file_with_the_same_code(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123", duration=3600)
new_path = self.put("hd/ABC-123.mkv")
result = self.scan_all()
new_id = self.id_of(new_path)
self.assertEqual(result.reattached, 1)
self.assertEqual(self.query("SELECT id FROM asset"), [(new_id,)], "旧行搬完就删")
for table in ("asset_preference", "watch_queue", "playlist_item", "asset_quality_goal",
"activity_event", "asset_tag_preference"):
self.assertEqual(self.owner(table), [new_id], table)
self.assertEqual(self.query("SELECT current_asset_id FROM playlist"), [(new_id,)])
self.assertEqual(self.query("SELECT position_seconds FROM activity_event"), [(61.5,)])
goal = self.query("SELECT wanted,replaced_at IS NOT NULL FROM asset_quality_goal")
self.assertEqual(goal, [(0, 1)], "寻找更好版本随记录搬过去后关闭，标已替换")
self.assertEqual(self.owner("asset_tag"), [], "资料不在搬运范围内，留在快照里")
batch = self.query("SELECT source,rule,old_asset_id,new_asset_id FROM record_rehome")
self.assertEqual(batch, [(record_rehome.AUTO_SOURCE, "code", 7, new_id)])
def test_push_discovery_ingest_moves_records_too(self):
self.add_vanished(7, "R:\\media\\old\\SSIS-950.mp4", code="SSIS-950")
new_path = self.put("SSIS-950.mkv")
result = self.ingest(new_path)
self.assertTrue(result.new)
self.assertEqual(result.reattached, 1)
self.assertEqual(self.owner("asset_preference"), [result.asset_id])
self.assertEqual(self.query("SELECT count(*) FROM asset WHERE id=7"), [(0,)])
def test_new_row_records_win_over_the_moved_ones(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("INSERT INTO asset(id,location,path,name,medium,size) "
"VALUES(9,'local','R:\\media\\ABC-123.mkv','ABC-123.mkv','video',10)")
connection.execute("INSERT INTO asset_preference(profile_id,asset_id,liked,reason,updated_at) "
"VALUES('default',9,0,'新的','2026-02-01')")
record_rehome.move(connection, 7, 9, source=record_rehome.MANUAL_SOURCE, rule="manual")
connection.commit()
self.assertEqual(self.query("SELECT asset_id,reason FROM asset_preference"), [(9, "新的")])
self.assertEqual(self.owner("watch_queue"), [9])
def test_several_new_files_pick_the_closest_duration_or_wait_for_a_person(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123", duration=3600)
near, far = self.put("a/ABC-123.mp4"), self.put("b/ABC-123.mp4")
result = self.scan_all()
self.assertEqual(result.reattached, 0, "登记那一刻新文件还没有时长，分不出")
self.assertEqual(self.owner("asset_preference"), [7])
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("UPDATE asset SET duration=3598 WHERE path=?", (near,))
connection.execute("UPDATE asset SET duration=1800 WHERE path=?", (far,))
connection.commit()
self.assertEqual(record_rehome.reattach_in(self.db, list(result.new_ids)), 1)
self.assertEqual(self.owner("asset_preference"), [self.id_of(near)])
def test_two_vanished_rows_with_one_code_are_left_for_a_person(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123-1.mp4", code="ABC-123")
self.add_vanished(8, "R:\\media\\old\\ABC-123-2.mp4", code="ABC-123")
self.put("ABC-123.mp4")
self.assertEqual(self.scan_all().reattached, 0)
self.assertEqual(self.owner("asset_preference"), [7, 8])
def test_a_creator_video_without_code_matches_creator_stem_and_duration(self):
self.add_vanished(7, "R:\\media\\桃子\\old\\海边.mp4", creator="桃子", duration=600)
path = self.put("桃子/2026/海边.mp4")
result = self.scan_all()
self.assertEqual(result.reattached, 0, "登记时还没有时长")
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("UPDATE asset SET duration=601 WHERE path=?", (path,))
connection.commit()
self.assertEqual(record_rehome.reattach_in(self.db, list(result.new_ids)), 1)
self.assertEqual(self.owner("watch_queue"), [self.id_of(path)])
def test_a_creator_video_under_another_creator_is_not_matched(self):
self.add_vanished(7, "R:\\media\\桃子\\海边.mp4", creator="桃子", duration=600)
path = self.put("别人/海边.mp4")
result = self.scan_all()
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("UPDATE asset SET duration=600 WHERE path=?", (path,))
connection.commit()
self.assertEqual(record_rehome.reattach_in(self.db, list(result.new_ids)), 0)
def test_revert_rebuilds_the_old_row_and_points_records_back(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123", duration=3600)
new_path = self.put("ABC-123.mp4")
self.scan_all()
new_id = self.id_of(new_path)
with closing(sqlite3.connect(self.db)) as connection:
planned = record_rehome.planned_revert(connection, record_rehome.AUTO_SOURCE)
self.assertEqual([item["old_asset_id"] for item in planned], [7])
with closing(sqlite3.connect(self.db)) as connection:
self.assertEqual(record_rehome.planned_revert(
connection, record_rehome.MANUAL_SOURCE, planned[0]["batch"]), [],
"别的来源的批次号一条都不认")
spec = importlib.util.spec_from_file_location(
"revert_auto_landing_under_test", ROOT / "scripts" / "revert_auto_landing.py")
script = importlib.util.module_from_spec(spec)
spec.loader.exec_module(script)
with redirect_stdout(io.StringIO()) as printed:
self.assertEqual(script.main([
"--db", str(self.db), "--logo-root", str(self.root / "logos"),
"--source", record_rehome.AUTO_SOURCE, "--batch", planned[0]["batch"],
"--apply", "--backup", str(self.root / "backup.db")]), 0)
self.assertIn("'撤回接回': 1", printed.getvalue())
self.assertEqual(self.query("SELECT id,disposal,rating FROM asset ORDER BY id"),
[(7, "vanished", 4), (new_id, None, None)])
for table in ("asset_preference", "watch_queue", "playlist_item", "asset_quality_goal",
"activity_event", "asset_tag_preference", "asset_tag"):
self.assertEqual(self.owner(table), [7], table)
self.assertEqual(self.query("SELECT wanted,replaced_at FROM asset_quality_goal"), [(1, None)])
self.assertEqual(self.query("SELECT current_asset_id FROM playlist"), [(7,)])
with closing(sqlite3.connect(self.db)) as connection:
self.assertEqual(record_rehome.planned_revert(connection, record_rehome.AUTO_SOURCE), [])
def test_revert_refuses_when_the_old_id_is_taken(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("INSERT INTO asset(id,location,path,name,medium,size) "
"VALUES(9,'local','R:\\media\\ABC-123.mkv','ABC-123.mkv','video',10)")
batch = record_rehome.move(connection, 7, 9, source=record_rehome.MANUAL_SOURCE, rule="manual")
connection.execute("INSERT INTO asset(id,location,path,name,medium,size) "
"VALUES(7,'local','R:\\media\\other.mp4','other.mp4','video',10)")
with self.assertRaisesRegex(ValueError, "已被占用"):
record_rehome.revert_one(connection, batch)
snapshot = json.loads(connection.execute(
"SELECT snapshot_json FROM record_rehome WHERE id=?", (batch,)).fetchone()[0])
self.assertEqual(snapshot["asset"]["rating"], 4, "行上的打分留在快照里")
def add_live(self, asset_id, path, *, code=None, duration=None):
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("INSERT INTO asset(id,location,path,name,medium,size,code,duration) "
"VALUES(?,'local',?,?,'video',10,?,?)",
(asset_id, path, path.rsplit("\\", 1)[-1], code, duration))
connection.commit()
def reattach_vanished(self, *old_ids):
with closing(sqlite3.connect(self.db)) as connection:
moved = record_rehome.reattach_vanished(connection, old_ids)
connection.commit()
return moved
def test_a_row_that_just_vanished_moves_to_the_one_version_already_in_the_library(self):
self.add_live(9, "R:\\media\\hd\\ABC-123.mkv", code="ABC-123")
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
self.assertEqual(self.reattach_vanished(7), [7])
self.assertEqual(self.owner("asset_preference"), [9])
self.assertEqual(self.query("SELECT source,rule,old_asset_id,new_asset_id FROM record_rehome"),
[(record_rehome.AUTO_SOURCE, "code", 7, 9)])
def test_a_row_that_just_vanished_with_two_versions_and_no_duration_stays_an_orphan(self):
self.add_live(9, "R:\\media\\a\\ABC-123.mkv", code="ABC-123")
self.add_live(10, "R:\\media\\b\\ABC-123.mp4", code="ABC-123")
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
self.assertEqual(self.reattach_vanished(7), [])
self.assertEqual(self.owner("asset_preference"), [7])
def test_only_the_rows_that_just_vanished_are_moved(self):
self.add_live(9, "R:\\media\\ABC-123.mkv", code="ABC-123")
self.add_live(10, "R:\\media\\XYZ-001.mkv", code="XYZ-001")
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
self.add_vanished(8, "R:\\media\\old\\XYZ-001.mp4", code="XYZ-001")
self.assertEqual(self.reattach_vanished(8), [8])
self.assertEqual(self.owner("watch_queue"), [7, 10], "不是这一轮标的孤儿不跟着搬")
def test_a_creator_video_that_just_vanished_matches_creator_stem_and_duration(self):
self.add_live(9, "R:\\media\\桃子\\2026\\海边.mp4", duration=601)
self.add_live(10, "R:\\media\\别人\\海边.mp4", duration=600)
self.add_vanished(7, "R:\\media\\桃子\\old\\海边.mp4", creator="桃子", duration=600)
self.assertEqual(self.reattach_vanished(7), [7])
self.assertEqual(self.owner("watch_queue"), [9])
def test_each_of_the_four_row_columns_alone_counts_as_records(self):
with closing(sqlite3.connect(self.db)) as connection:
connection.executemany(
"INSERT INTO asset(id,location,path,name,medium,size,rating,o_count,play_count,last_played) "
"VALUES(?,'local',?,'x.mp4','video',10,?,?,?,?)",
[(1, "R:\\media\\1.mp4", 3, None, 0, None), (2, "R:\\media\\2.mp4", None, 2, 0, None),
(3, "R:\\media\\3.mp4", None, None, 5, None), (4, "R:\\media\\4.mp4", None, None, 0, 1700.0),
(5, "R:\\media\\5.mp4", None, 0, 0, None)])
self.assertEqual(personal_records.record_holders(connection, [1, 2, 3, 4, 5]), {1, 2, 3, 4})
self.assertEqual(personal_records.record_counts(connection, [1, 2, 3, 4, 5]), {
1: {"rating": 3}, 2: {"o_count": 2}, 3: {"play_count": 5}, 4: {"last_played": 1}, 5: {}})
def row_records(self, asset_id):
"""行上四列；`last_played` 是 TEXT 列，写进去的 epoch 秒读回来是文本。"""
return self.query("SELECT rating,o_count,play_count,last_played FROM asset WHERE id=?", (asset_id,))[0]
def test_row_records_fold_into_the_new_row_and_revert_restores_both(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("UPDATE asset SET o_count=2,play_count=3,last_played=1000.0 WHERE id=7")
connection.execute(
"INSERT INTO asset(id,location,path,name,medium,size,rating,o_count,play_count,last_played) "
"VALUES(9,'local','R:\\media\\ABC-123.mkv','ABC-123.mkv','video',10,5,NULL,1,2000.0)")
batch = record_rehome.move(connection, 7, 9, source=record_rehome.MANUAL_SOURCE, rule="manual")
connection.commit()
self.assertEqual(self.row_records(9), (5, 2, 4, "2000.0"),
"评分新行有就留新行的；次数相加；最近播放取较晚的")
with closing(sqlite3.connect(self.db)) as connection:
record_rehome.revert_one(connection, batch)
connection.commit()
self.assertEqual(self.row_records(9), (5, None, 1, "2000.0"), "新行还原成接回前的值")
self.assertEqual(self.row_records(7), (4, 2, 3, "1000.0"), "旧行按快照重建")
def test_an_unrated_new_row_takes_the_old_rating_and_the_later_play(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("UPDATE asset SET play_count=2,last_played=3000.0 WHERE id=7")
connection.execute("INSERT INTO asset(id,location,path,name,medium,size,last_played) "
"VALUES(9,'local','R:\\media\\ABC-123.mkv','ABC-123.mkv','video',10,1000.0)")
record_rehome.move(connection, 7, 9, source=record_rehome.MANUAL_SOURCE, rule="manual")
connection.commit()
self.assertEqual(self.row_records(9), (4, None, 2, "3000.0"))
def test_move_only_takes_a_vanished_row_onto_an_in_library_row(self):
self.add_vanished(7, "R:\\media\\old\\ABC-123.mp4", code="ABC-123")
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("INSERT INTO asset(id,location,path,name,medium,size,disposal) "
"VALUES(9,'local','R:\\media\\x.mp4','x.mp4','video',10,'trash')")
with self.assertRaisesRegex(ValueError, "在库"):
record_rehome.move(connection, 7, 9, source=record_rehome.MANUAL_SOURCE, rule="manual")
with self.assertRaisesRegex(ValueError, "已消失"):
record_rehome.move(connection, 9, 7, source=record_rehome.MANUAL_SOURCE, rule="manual")
# 以下：旧行还在库、文件已不在盘上，新文件同番号，登记前把旧行改指过去（ADR-0131）。
OLD = "R:\\media\\演员\\FNS-257\\FNS-257.mp4"
def add_old(self, asset_id=7, path=OLD):
"""一条在库旧行，盘上没有它的文件；带刮削标签、探测结果与一次播放。"""
self.add_live(asset_id, path, code="FNS-257", duration=7208.23)
with closing(sqlite3.connect(self.db)) as connection:
connection.execute("UPDATE asset SET width=1920,height=1080,play_count=1 WHERE id=?", (asset_id,))
connection.execute("INSERT INTO asset_tag(asset_id,tag,source) VALUES(?,'刮削标签','javdb')", (asset_id,))
connection.commit()
def scan_new(self):
with redirect_stdout(io.StringIO()):
return scan.scan_location(self.db, "local", "R:\\media", declared_roots=DECLARED,
mounts=self.mounts, windows=False, new_only=True)
def test_a_renamed_and_reencoded_folder_keeps_its_row(self):
"""#312：目录改名又换了编码，增量扫描不新建行，标签与播放留在原行，探测列等重探。"""
self.add_old()
new_path = self.put("演员/FNS-257 新后缀/FNS-257.mp4")
result = self.scan_new()
self.assertEqual((result.new_ids, result.relocated_ids), ((), (7,)))
self.assertEqual(self.query("SELECT id,path,code,duration,width,height,play_count FROM asset"),
[(7, new_path, "FNS-257", None, None, None, 1)])
self.assertEqual(self.owner("asset_tag"), [7])
self.assertEqual(self.query("SELECT count(*) FROM record_rehome"), [(0,)])
def test_push_discovery_keeps_the_row_too(self):
self.add_old()
new_path = self.put("演员/FNS-257 新后缀/FNS-257.mp4")
result = self.ingest(new_path)
self.assertEqual((result.asset_id, result.new, result.relocated), (7, False, 7))
self.assertEqual(self.query("SELECT id,path FROM asset"), [(7, new_path)])
def test_an_old_file_still_on_disk_is_a_second_copy(self):
old_path = self.put("演员/FNS-257/FNS-257.mp4")
self.scan_all()
new_path = self.put("演员/FNS-257 新后缀/FNS-257.mp4")
result = self.scan_new()
self.assertEqual(result.relocated_ids, ())
self.assertEqual(sorted(row[0] for row in self.query("SELECT path FROM asset")), sorted([old_path, new_path]))
def test_two_old_rows_with_the_code_are_left_alone(self):
self.add_old(7)
self.add_old(8, "R:\\media\\演员\\FNS-257 4K\\FNS-257.mp4")
new_path = self.put("演员/FNS-257 新后缀/FNS-257.mp4")
result = self.scan_new()
self.assertEqual(result.relocated_ids, ())
self.assertEqual(self.query("SELECT path FROM asset WHERE id IN (7,8) ORDER BY id"),
[(self.OLD,), ("R:\\media\\演员\\FNS-257 4K\\FNS-257.mp4",)])
self.assertEqual(len(result.new_ids), 1)
self.assertEqual(self.query("SELECT path FROM asset WHERE id=?", (result.new_ids[0],)), [(new_path,)])
def test_two_new_videos_with_the_code_in_one_folder_are_left_alone(self):
self.add_old()
self.put("演员/FNS-257 新后缀/FNS-257.mp4")
self.put("演员/FNS-257 新后缀/FNS-257.mkv")
result = self.scan_new()
self.assertEqual((result.relocated_ids, len(result.new_ids)), ((), 2))
self.assertEqual(self.query("SELECT path FROM asset WHERE id=7"), [(self.OLD,)])
def test_an_old_row_under_another_media_folder_is_left_alone(self):
"""另一个媒体目录可能只是没挂上：它下面的行答「不在」不能作数。"""
other = self.root / "other"
other.mkdir()
self.mounts["local"] = (str(self.media), str(other))
self.add_old(path="R:\\other\\FNS-257\\FNS-257.mp4")
self.put("演员/FNS-257 新后缀/FNS-257.mp4")
declared = {"local": ("R:\\media", "R:\\other")}
with redirect_stdout(io.StringIO()):
result = scan.scan_location(self.db, "local", "R:\\media", declared_roots=declared,
mounts=self.mounts, windows=False, new_only=True)
self.assertEqual((result.relocated_ids, len(result.new_ids)), ((), 1))
self.assertEqual(self.query("SELECT path FROM asset WHERE id=7"), [("R:\\other\\FNS-257\\FNS-257.mp4",)])
if __name__ == "__main__":
unittest.main()