Files
Curator/tests/test_pool_index.py

250 lines
8.9 KiB
Python

import json
from src.core.file import File
from src.core.pool_index import INDEX_FILENAME, INDEX_VERSION, PoolIndex
from src.core.tag_manager import TagManager
class TestPoolIndex:
"""Testy pro unifikovaný metadata index poolu (v2 — klíč = identita filmu)"""
def test_empty_index_when_no_file(self, tmp_path):
index = PoolIndex(tmp_path)
assert index.records == {}
assert index.get_movie("tmdb-603") is None
def test_set_and_get_movie_by_identity(self, tmp_path):
index = PoolIndex(tmp_path)
index.set_movie("tmdb-603", {"title": "Matrix", "media": []})
assert index.get_movie("tmdb-603") == {"title": "Matrix", "media": []}
assert "tmdb-603" in index.records
def test_set_persists_to_disk_with_version(self, tmp_path):
index = PoolIndex(tmp_path)
index.set_movie("tmdb-603", {"title": "Matrix", "media": []})
on_disk = json.loads((tmp_path / INDEX_FILENAME).read_text(encoding="utf-8"))
assert on_disk["version"] == INDEX_VERSION
assert on_disk["movies"]["tmdb-603"]["title"] == "Matrix"
def test_reload_reads_existing_index(self, tmp_path):
PoolIndex(tmp_path).set_movie("tmdb-603", {"title": "Matrix", "media": []})
reloaded = PoolIndex(tmp_path)
assert reloaded.get_movie("tmdb-603")["title"] == "Matrix"
def test_delete_movie_removes_record(self, tmp_path):
index = PoolIndex(tmp_path)
index.set_movie("tmdb-603", {"title": "Matrix", "media": []})
index.delete_movie("tmdb-603")
assert index.get_movie("tmdb-603") is None
assert PoolIndex(tmp_path).get_movie("tmdb-603") is None
def test_rekey_movie(self, tmp_path):
index = PoolIndex(tmp_path)
index.set_movie("name-matrix-1999", {"title": "Matrix", "media": []})
index.rekey_movie("name-matrix-1999", "tmdb-603")
assert index.get_movie("name-matrix-1999") is None
assert index.get_movie("tmdb-603")["title"] == "Matrix"
def test_find_movie_by_media(self, tmp_path):
index = PoolIndex(tmp_path)
index.set_movie(
"tmdb-603",
{
"title": "Matrix",
"media": [{"role": "video", "label": "", "file": "Filmy/Matrix.mkv"}],
},
)
found = index.find_movie_by_media(tmp_path / "Filmy" / "Matrix.mkv")
assert found is not None
key, record = found
assert key == "tmdb-603"
assert record["title"] == "Matrix"
assert index.find_movie_by_media(tmp_path / "Filmy" / "Jiny.mkv") is None
def test_corrupt_index_loads_empty(self, tmp_path):
(tmp_path / INDEX_FILENAME).write_text("{ not json", encoding="utf-8")
index = PoolIndex(tmp_path)
assert index.records == {}
class TestV1Migration:
"""Starý path-klíčovaný index se převede na identity-klíčovaný (bez hýbání soubory)"""
def _write_v1(self, tmp_path, movies: dict) -> None:
(tmp_path / INDEX_FILENAME).write_text(
json.dumps({"settings": {"tag_schema": []}, "movies": movies}),
encoding="utf-8",
)
def test_flat_movie_becomes_identity_keyed_record(self, tmp_path):
self._write_v1(
tmp_path,
{
"Filmy/Matrix (1999) [tmdbid-603].mkv": {
"title": "Matrix",
"tags": ["Žánr/Akční"],
"csfd_url": "https://www.csfd.cz/film/9499/",
"date": "1999-03-31",
"added": "2026-01-01T00:00:00",
"attributes": {"collection_sort": "01"},
}
},
)
index = PoolIndex(tmp_path)
record = index.get_movie("tmdb-603")
assert record is not None
assert record["title"] == "Matrix"
assert record["year"] == "1999"
assert record["tmdb_id"] == 603
assert record["library"] == "Filmy"
assert record["media"] == [
{
"role": "video",
"label": "",
"file": "Filmy/Matrix (1999) [tmdbid-603].mkv",
}
]
assert record["tags"] == ["Žánr/Akční"]
assert record["csfd_url"] == "https://www.csfd.cz/film/9499/"
assert record["attributes"] == {"collection_sort": "01"}
def test_grouped_versions_merge_into_one_record(self, tmp_path):
base = "Filmy/Avatar (2009) [tmdbid-19995]"
self._write_v1(
tmp_path,
{
f"{base}/Avatar (2009) [tmdbid-19995] - Výchozí.mkv": {
"title": "Avatar"
},
f"{base}/Avatar (2009) [tmdbid-19995] - 3D HSBS.mkv": {
"title": "Avatar"
},
},
)
index = PoolIndex(tmp_path)
assert len(index.all_movies()) == 1
record = index.get_movie("tmdb-19995")
assert {m["label"] for m in record["media"]} == {"Výchozí", "3D HSBS"}
assert record["library"] == "Filmy"
def test_movie_without_tmdbid_gets_slug_key(self, tmp_path):
self._write_v1(tmp_path, {"Filmy/Pelíšky (1999).mkv": {"title": "Pelíšky"}})
index = PoolIndex(tmp_path)
assert index.get_movie("name-pelisky-1999") is not None
def test_migration_backs_up_and_bumps_version(self, tmp_path):
self._write_v1(tmp_path, {"Filmy/A (2000).mkv": {"title": "A"}})
PoolIndex(tmp_path)
backup = tmp_path / (INDEX_FILENAME + ".v1.bak")
assert backup.exists()
assert (
"Filmy/A (2000).mkv"
in json.loads(backup.read_text(encoding="utf-8"))["movies"]
)
on_disk = json.loads((tmp_path / INDEX_FILENAME).read_text(encoding="utf-8"))
assert on_disk["version"] == INDEX_VERSION
def test_migration_keeps_settings(self, tmp_path):
self._write_v1(tmp_path, {"Filmy/A (2000).mkv": {"title": "A"}})
index = PoolIndex(tmp_path)
assert index.get_setting("tag_schema") == []
def test_v2_index_is_not_migrated_again(self, tmp_path):
(tmp_path / INDEX_FILENAME).write_text(
json.dumps(
{
"version": INDEX_VERSION,
"settings": {},
"movies": {"tmdb-1": {"title": "X", "media": []}},
}
),
encoding="utf-8",
)
index = PoolIndex(tmp_path)
assert index.get_movie("tmdb-1")["title"] == "X"
assert not (tmp_path / (INDEX_FILENAME + ".v1.bak")).exists()
class TestFileWithIndex:
"""File je pohled na jeden media záznam filmu v indexu"""
def test_index_backed_file_writes_no_sidecar(self, tmp_path):
index = PoolIndex(tmp_path)
movie = tmp_path / "Filmy" / "Matrix.mkv"
movie.parent.mkdir(parents=True)
movie.write_bytes(b"x")
f = File(movie, TagManager(), index=index)
assert not (movie.parent / ".Matrix.mkv.!tag").exists() # no sidecar
assert index.find_movie_by_media(movie) is not None # record created
assert f.tags == [] # no automatic tags
def test_index_backed_metadata_persists_across_reload(self, tmp_path):
index = PoolIndex(tmp_path)
movie = tmp_path / "Filmy" / "Matrix.mkv"
movie.parent.mkdir(parents=True)
movie.write_bytes(b"x")
tm = TagManager()
f = File(movie, tm, index=index)
f.title = "Matrix"
f.meta = {"year": "1999", "rating": 87}
f.add_tag("Žánr/Akční")
f.set_date("1999-03-31")
# Fresh index + File read from disk
index2 = PoolIndex(tmp_path)
f2 = File(movie, TagManager(), index=index2)
assert f2.title == "Matrix"
assert f2.meta == {"year": "1999", "rating": 87}
assert f2.date == "1999-03-31"
assert "Žánr/Akční" in {t.full_path for t in f2.tags}
def test_delete_metadata_removes_index_record(self, tmp_path):
index = PoolIndex(tmp_path)
movie = tmp_path / "Filmy" / "Matrix.mkv"
movie.parent.mkdir(parents=True)
movie.write_bytes(b"x")
f = File(movie, TagManager(), index=index)
f.delete_metadata()
assert index.find_movie_by_media(movie) is None
def test_two_versions_share_one_movie_record(self, tmp_path):
"""Obě verze čtou stejná metadata — to je smysl identity klíče."""
index = PoolIndex(tmp_path)
folder = tmp_path / "Filmy" / "Avatar (2009) [tmdbid-19995]"
folder.mkdir(parents=True)
a = folder / "Avatar (2009) [tmdbid-19995] - Výchozí.mkv"
b = folder / "Avatar (2009) [tmdbid-19995] - 3D HSBS.mkv"
a.write_bytes(b"x")
b.write_bytes(b"y")
tm = TagManager()
fa = File(a, tm, index=index)
fa.title = "Avatar"
fa.add_tag("Žánr/Sci-Fi")
fb = File(b, tm, index=index)
assert len(index.all_movies()) == 1
assert fb.title == "Avatar"
assert "Žánr/Sci-Fi" in {t.full_path for t in fb.tags}