tests/test_media_sources.py
Rama master · réplica de lectura del repositorio original.
"""挂载来源注册表、配置保护与 SQL/逐项求值的一致性。"""
from dataclasses import replace
from pathlib import Path
import sqlite3
import tempfile
from types import SimpleNamespace
import unittest
from peach import media_configuration, media_sources, settings_file
class MediaSourceCostTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.config = replace(settings_file.PeachConfig(self.root, self.root / "config.toml"),
locations={"guangya": ("G:/Movies", "G:/Movies/Direct"),
"onedrive": ("O:/Public", "O:/Private"), "mounted": ("M:/",)},
media_costs={"g:/movies": "metered", "G:/Movies/Direct": "free", "O:/Public": "free"})
def test_registry_has_mount_sources_and_proxy_defaults(self):
self.assertTrue({"onedrive", "guangya", "mounted"}.issubset(media_sources.SCANNABLE_LOCATIONS))
self.assertEqual(media_sources.SOURCE_COSTS["guangya"], "free")
self.assertEqual(media_sources.SOURCE_COSTS["115"], "free")
self.assertEqual(media_sources.SOURCE_COSTS["pikpak"], "metered")
self.assertEqual(media_sources.SOURCE_COSTS["onedrive"], "metered")
roots, _, errors = media_configuration.validate([
{"location": "onedrive", "path": "O:/", "cost": "free"},
{"location": "mounted", "path": "M:/", "cost": "metered"}], windows=True)
self.assertEqual(errors, [])
self.assertEqual(set(roots), {"onedrive", "mounted"})
def test_minimal_configuration_objects_use_source_defaults(self):
for config in (SimpleNamespace(locations={"115": ("B:/",)}), SimpleNamespace()):
self.assertEqual(media_sources.cost_for("115", "B:\\film.mp4", config), "free")
expression, params = media_sources.cost_sql(config=config)
with sqlite3.connect(":memory:") as connection:
value = connection.execute(
f"SELECT {expression} FROM (SELECT '115' AS location,'B:\\film.mp4' AS path)", params).fetchone()[0]
self.assertEqual(value, "free")
def test_longest_root_directory_boundary_and_unknown_paths(self):
cases = [
("guangya", "G:\\Movies\\film.mp4", "metered"),
("guangya", "g:\\movies\\Direct\\film.mp4", "free"),
("guangya", "G:\\MoviesOther\\film.mp4", "free"),
("onedrive", "O:\\Public\\film.mp4", "free"),
("onedrive", "O:\\Private\\film.mp4", "metered"),
("onedrive", "O:\\Outside\\film.mp4", "metered"),
("mounted", "M:\\film.mp4", "metered"),
("unknown", "X:\\film.mp4", "metered"),
("local", "R:\\film.mp4", "free"),
("115", "B:\\film.mp4", "free"),
]
expression, params = media_sources.cost_sql("a.location", "a.path", config=self.config)
with sqlite3.connect(":memory:") as connection:
for location, path, expected in cases:
with self.subTest(location=location, path=path):
actual = media_sources.cost_for(location, path, self.config)
result = connection.execute(
f"SELECT {expression} FROM (SELECT ? AS location, ? AS path) a",
(*params, location, path)).fetchone()[0]
self.assertEqual((actual, result), (expected, expected))
def test_null_and_empty_paths_aggregate_configured_root_protection(self):
config = replace(self.config, media_costs={"G:/Movies": "free", "G:/Movies/Direct": "free",
"O:/Public": "free", "O:/Private": "metered", "M:/": "free"})
expression, params = media_sources.cost_sql(config=config)
with sqlite3.connect(":memory:") as connection:
for location, expected in (("guangya", "free"), ("onedrive", "metered"), ("mounted", "free")):
for path in (None, ""):
result = connection.execute(
f"SELECT {expression} FROM (SELECT ? AS location, ? AS path)",
(*params, location, path)).fetchone()[0]
self.assertEqual(result, expected)
self.assertEqual(media_sources.cost_for(location, config=config), expected)
def test_sql_expression_can_filter_assets_with_bound_parameters(self):
expression, params = media_sources.cost_sql(config=self.config)
with sqlite3.connect(":memory:") as connection:
connection.execute("CREATE TABLE asset(location TEXT,path TEXT)")
connection.executemany("INSERT INTO asset VALUES(?,?)", [
("onedrive", "O:\\Public\\film.mp4"), ("onedrive", "O:\\Private\\film.mp4")])
rows = connection.execute(f"SELECT path FROM asset WHERE ({expression})=?",
(*params, "free")).fetchall()
self.assertEqual(rows, [("O:\\Public\\film.mp4",)])
def test_nested_roots_are_refused_before_the_file_is_written(self):
with self.assertRaisesRegex(settings_file.SettingsFileError, "互相包含"):
settings_file.write(self.config)
self.assertFalse(self.config.path.exists())
def test_config_roundtrip_preserves_root_overrides_and_effective_form_cost(self):
config = replace(self.config, locations={**self.config.locations, "guangya": ("G:/Movies",)},
media_costs={"G:/Movies": "metered", "O:/Public": "free"})
settings_file.write(config)
loaded = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
self.assertEqual(loaded.media_costs, {"G:/Movies": "metered", "O:/Public": "free"})
rows = media_configuration.rows(loaded, windows=True)
self.assertEqual([row["cost"] for row in rows], ["metered", "free", "metered", "metered"])
costs = media_configuration.costs([{"location": "guangya", "path": "G:/Movies", "cost": "free"}],
loaded, windows=True)
self.assertEqual(costs["G:\\Movies"], "free")
def test_invalid_cost_values_are_rejected_without_filesystem_work(self):
for value in (True, "proxy", [], None):
self.assertTrue(media_configuration.validate([
{"location": "onedrive", "path": "O:/", "cost": value}], windows=True)[2])
settings_file.write(replace(self.config, locations={"guangya": ("G:/Movies",)}, media_costs={"G:/": "unknown"}))
with self.assertRaisesRegex(settings_file.SettingsFileError, "free.*metered"):
settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
with self.assertRaisesRegex(ValueError, "列名"):
media_sources.cost_sql("a.location; DROP TABLE asset", config=self.config)