Files
Curator/tests/test_file_manager.py
T

518 lines
22 KiB
Python

import pytest
from src.core.file import File
from src.core.file_manager import FileManager
from src.core.tag_manager import TagManager
from src.core.tag import Tag
@pytest.fixture
def temp_global_config(tmp_path, monkeypatch):
"""Point the global config file at a throwaway path for each test."""
config_path = tmp_path / "test_config.json"
import src.core.config as config_module
monkeypatch.setattr(config_module, "GLOBAL_CONFIG_FILE", config_path)
return config_path
@pytest.fixture
def tag_manager():
return TagManager()
@pytest.fixture
def file_manager(tag_manager, temp_global_config):
return FileManager(tag_manager)
def _in_memory_files(tmp_path, tag_manager, *names):
"""Create real files and wrap them as index-less (in-memory) File objects."""
files = []
for name in names:
path = tmp_path / name
path.write_text("content")
files.append(File(path, tag_manager))
return files
class TestFileManager:
"""Basic FileManager state."""
def test_file_manager_creation(self, file_manager, tag_manager):
assert file_manager.filelist == []
assert file_manager.tagmanager == tag_manager
assert file_manager.global_config is not None
assert file_manager.index is None
class TestFileManagerTagOperations:
"""assign/remove tag operations over the loaded file list."""
def test_assign_tag_to_file_objects_tag_object(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt", "f2.txt")
file_manager.filelist = files
tag = Tag("Video", "HD")
file_manager.assign_tag_to_file_objects(files, tag)
for f in files:
assert tag in f.tags
def test_assign_tag_string_with_category(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt")
file_manager.filelist = files
file_manager.assign_tag_to_file_objects(files, "Video/4K")
tag_paths = {tag.full_path for tag in files[0].tags}
assert "Video/4K" in tag_paths
def test_assign_tag_string_without_category(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt")
file_manager.filelist = files
file_manager.assign_tag_to_file_objects(files, "SimpleTag")
tag_paths = {tag.full_path for tag in files[0].tags}
assert "default/SimpleTag" in tag_paths
def test_assign_tag_no_duplicate(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt")
file_manager.filelist = files
tag = Tag("Video", "HD")
file_manager.assign_tag_to_file_objects(files, tag)
file_manager.assign_tag_to_file_objects(files, tag)
count = sum(1 for t in files[0].tags if t == tag)
assert count == 1
def test_remove_tag_from_file_objects(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt", "f2.txt")
file_manager.filelist = files
tag = Tag("Video", "HD")
file_manager.assign_tag_to_file_objects(files, tag)
file_manager.remove_tag_from_file_objects(files, tag)
for f in files:
assert tag not in f.tags
def test_remove_tag_string(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt")
file_manager.filelist = files
file_manager.assign_tag_to_file_objects(files, "Video/HD")
file_manager.remove_tag_from_file_objects(files, "Video/HD")
tag_paths = {tag.full_path for tag in files[0].tags}
assert "Video/HD" not in tag_paths
def test_callback_on_tag_change(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt")
file_manager.filelist = files
callback_calls = []
file_manager.on_files_changed = lambda filelist: callback_calls.append(len(filelist))
file_manager.assign_tag_to_file_objects([files[0]], Tag("Test", "Tag"))
assert len(callback_calls) == 1
class TestFileManagerFiltering:
"""filter_files_by_tags over the loaded file list."""
@pytest.fixture
def loaded(self, file_manager, tmp_path, tag_manager):
files = _in_memory_files(tmp_path, tag_manager, "f1.txt", "f2.txt", "f3.txt")
file_manager.filelist = files
return file_manager
def test_filter_empty_tags_returns_all(self, loaded):
assert len(loaded.filter_files_by_tags([])) == len(loaded.filelist)
def test_filter_none_returns_all(self, loaded):
assert len(loaded.filter_files_by_tags(None)) == len(loaded.filelist)
def test_filter_by_single_tag(self, loaded):
tag = Tag("Video", "HD")
loaded.assign_tag_to_file_objects(loaded.filelist[:2], tag)
filtered = loaded.filter_files_by_tags([tag])
assert len(filtered) == 2
for f in filtered:
assert tag in f.tags
def test_filter_by_multiple_tags_and_logic(self, loaded):
tag1 = Tag("Video", "HD")
tag2 = Tag("Audio", "Stereo")
loaded.assign_tag_to_file_objects([loaded.filelist[0]], tag1)
loaded.assign_tag_to_file_objects([loaded.filelist[0]], tag2)
loaded.assign_tag_to_file_objects([loaded.filelist[1]], tag1)
filtered = loaded.filter_files_by_tags([tag1, tag2])
assert len(filtered) == 1
assert filtered[0] == loaded.filelist[0]
def test_filter_by_tag_strings(self, loaded):
loaded.assign_tag_to_file_objects([loaded.filelist[0]], "Video/HD")
assert len(loaded.filter_files_by_tags(["Video/HD"])) == 1
def test_filter_no_match(self, loaded):
assert len(loaded.filter_files_by_tags([Tag("NonExistent", "Tag")])) == 0
class TestFileManagerEdgeCases:
def test_empty_filelist_operations(self, file_manager):
assert file_manager.filter_files_by_tags([Tag("Video", "HD")]) == []
file_manager.assign_tag_to_file_objects([], Tag("Video", "HD"))
assert len(file_manager.filelist) == 0
def test_assign_tag_to_empty_list(self, file_manager):
file_manager.assign_tag_to_file_objects([], Tag("Test", "Tag")) # no exception
def test_remove_nonexistent_tag(self, file_manager, tmp_path, tag_manager):
file_manager.filelist = _in_memory_files(tmp_path, tag_manager, "file.txt")
file_manager.remove_tag_from_file_objects(
file_manager.filelist, Tag("NonExistent", "Tag")) # no exception
class TestPoolManagement:
"""Testy pro pool a copy-as-is složky"""
@pytest.fixture
def temp_global_config(self, tmp_path, monkeypatch):
config_path = tmp_path / "test_config.json"
import src.core.config as config_module
monkeypatch.setattr(config_module, 'GLOBAL_CONFIG_FILE', config_path)
return config_path
@pytest.fixture
def file_manager(self, temp_global_config):
return FileManager(TagManager())
def test_set_pool_creates_top_level_folders(self, file_manager, tmp_path):
pool = tmp_path / "pool"
file_manager.set_pool_dir(pool)
assert (pool / "Filmy").is_dir()
assert (pool / "Seriály").is_dir()
assert file_manager.pool_dir == pool
def test_import_movie_copies_and_indexes(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mkv"
source.write_bytes(b"x" * 10)
movie = file_manager.import_movie(source, "Matrix", "https://csfd.cz/film/1")
assert movie.file_path == tmp_path / "pool" / "Filmy" / "Matrix.mkv"
assert source.exists() # non-destructive copy
assert movie.title == "Matrix"
assert movie.csfd_link == "https://csfd.cz/film/1"
assert file_manager.index.get(movie.file_path) is not None
def test_statistics_aggregates_pool(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"x" * 100)
(tmp_path / "b.mkv").write_bytes(b"y" * 50)
a = file_manager.import_movie(tmp_path / "a.mkv", "A", "https://csfd/1")
a.add_tag("Žánr/Akční")
a.add_tag("Žánr/Sci-Fi")
a.csfd_cache = {"rating": 90}
a.save_metadata()
b = file_manager.import_movie(tmp_path / "b.mkv", "B")
b.add_tag("Žánr/Akční")
s = file_manager.statistics()
assert s["count"] == 2
assert s["total_size"] == 150
assert s["with_csfd"] == 1 and s["without_csfd"] == 1
assert s["untagged"] == 0
assert s["avg_rating"] == 90
# Akční on both movies → count 2, sorted first
assert dict(s["categories"]["Žánr"])["Akční"] == 2
assert s["categories"]["Žánr"][0] == ("Akční", 2)
def test_check_data_consistency_clean(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
result = file_manager.check_data_consistency()
assert result["ok"]
assert result["missing"] == []
assert result["untracked"] == []
assert result["index_count"] == 1 and result["disk_count"] == 1
def test_check_data_consistency_detects_missing_file(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
movie = file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
movie.file_path.unlink() # someone deleted the file directly
result = file_manager.check_data_consistency()
assert result["missing"] == ["Filmy/Matrix.mkv"]
assert result["untracked"] == []
def test_check_data_consistency_detects_untracked_file(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
# a file dropped into the pool without importing
(file_manager.movies_dir / "Sneaked.mkv").write_bytes(b"x")
result = file_manager.check_data_consistency()
assert result["missing"] == []
assert result["untracked"] == ["Filmy/Sneaked.mkv"]
def test_import_movie_records_added_timestamp(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mkv"
source.write_bytes(b"x")
movie = file_manager.import_movie(source, "Matrix")
assert movie.added is not None
# parseable ISO timestamp, persisted in the index
from datetime import datetime
datetime.fromisoformat(movie.added)
assert file_manager.index.get(movie.file_path)["added"] == movie.added
def test_import_movie_move_removes_source(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mkv"
source.write_bytes(b"x" * 10)
movie = file_manager.import_movie(source, "Matrix", move=True)
assert movie.file_path == tmp_path / "pool" / "Filmy" / "Matrix.mkv"
assert movie.file_path.exists()
assert not source.exists() # moved, not copied
def test_filmoteka_category_roots_from_schema(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
file_manager.set_tag_schema([
{"category": "Žánr", "csfd_field": "genres", "transform": None, "filmoteka_root": ""},
{"category": "Rok", "csfd_field": "year", "transform": None, "filmoteka_root": "Dle roku"},
{"category": "Herec", "csfd_field": "actors", "transform": None, "filmoteka_root": None},
])
roots = file_manager.filmoteka_category_roots()
assert roots == {"Žánr": "", "Rok": "Dle roku"} # None-root excluded
assert "Herec" not in roots
def test_import_movie_suffix_keeps_both(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
(tmp_path / "b.mkv").write_bytes(b"b")
first = file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
second = file_manager.import_movie(tmp_path / "b.mkv", "Matrix") # default suffix
assert first.file_path.name == "Matrix.mkv"
assert second.file_path.name == "Matrix_1.mkv"
assert len(file_manager.filelist) == 2
def test_import_movie_replace_evicts_existing(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
(tmp_path / "b.mkv").write_bytes(b"bb")
first = file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
first.add_tag("Žánr/Akční")
old_path = first.file_path
second = file_manager.import_movie(
tmp_path / "b.mkv", "Matrix", csfd_link="x", on_conflict="replace")
assert second.file_path.name == "Matrix.mkv"
assert second.file_path == old_path # same name reused
assert second.file_path.read_bytes() == b"bb" # new content in place
assert [f.file_path.name for f in file_manager.filelist] == ["Matrix.mkv"]
# index record reflects the new import (fresh metadata, old tags dropped)
record = file_manager.index.get(second.file_path)
assert record is not None
assert record["csfd_link"] == "x"
assert record["tags"] == []
assert second.tags == []
def test_import_movie_replace_across_extensions(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
(tmp_path / "b.mp4").write_bytes(b"b")
file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
file_manager.import_movie(tmp_path / "b.mp4", "Matrix", on_conflict="replace")
names = [f.file_path.name for f in file_manager.filelist]
assert names == ["Matrix.mp4"]
assert not (tmp_path / "pool" / "Filmy" / "Matrix.mkv").exists()
def test_import_movie_skip_returns_none(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
(tmp_path / "b.mkv").write_bytes(b"b")
file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
result = file_manager.import_movie(tmp_path / "b.mkv", "Matrix", on_conflict="skip")
assert result is None
assert len(file_manager.filelist) == 1
def test_rename_movie_renames_file_and_reindexes(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mkv"
source.write_bytes(b"x" * 10)
movie = file_manager.import_movie(source, "Matrix")
movie.add_tag("Žánr/Sci-Fi")
old_path = movie.file_path
file_manager.rename_movie(movie, "Matrix Reloaded")
new_path = tmp_path / "pool" / "Filmy" / "Matrix Reloaded.mkv"
assert movie.file_path == new_path
assert new_path.exists()
assert not old_path.exists()
assert movie.title == "Matrix Reloaded"
# metadata moved to the new key, old key gone, tags preserved
assert file_manager.index.get(new_path) is not None
assert file_manager.index.get(old_path) is None
# a fresh manager reading the index sees the renamed file with its tags
reloaded = FileManager(TagManager())
reloaded.set_pool_dir(tmp_path / "pool")
reloaded.load_pool_movies()
assert [f.filename for f in reloaded.filelist] == ["Matrix Reloaded.mkv"]
assert "Žánr/Sci-Fi" in {t.full_path for t in reloaded.filelist[0].tags}
def test_rename_movie_preserves_extension(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mp4"
source.write_bytes(b"x")
movie = file_manager.import_movie(source, "Film")
file_manager.rename_movie(movie, "Jiný název")
assert movie.file_path.name == "Jiný název.mp4"
def test_rename_movie_rejects_existing_name(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
(tmp_path / "b.mkv").write_bytes(b"b")
first = file_manager.import_movie(tmp_path / "a.mkv", "Already")
second = file_manager.import_movie(tmp_path / "b.mkv", "Other")
with pytest.raises(FileExistsError):
file_manager.rename_movie(second, "Already")
# second movie is left untouched
assert second.file_path.name == "Other.mkv"
assert first.file_path.exists()
def test_rename_movie_rejects_empty_name(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
movie = file_manager.import_movie(tmp_path / "a.mkv", "Name")
with pytest.raises(ValueError):
file_manager.rename_movie(movie, " ")
def test_import_movie_with_year_uses_convention(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mkv"
source.write_bytes(b"x" * 10)
movie = file_manager.import_movie(source, "Matrix", year=1999)
assert movie.file_path.name == "Matrix (1999).mkv"
assert movie.title == "Matrix" # clean title, no year
def test_import_movie_rejects_invalid_year(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
with pytest.raises(ValueError):
file_manager.import_movie(tmp_path / "a.mkv", "Matrix", year="99")
def test_import_same_title_different_year_coexist(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "a.mkv").write_bytes(b"a")
(tmp_path / "b.mkv").write_bytes(b"b")
first = file_manager.import_movie(tmp_path / "a.mkv", "Solaris", year=1972)
second = file_manager.import_movie(tmp_path / "b.mkv", "Solaris", year=2002)
assert first.file_path.name == "Solaris (1972).mkv"
assert second.file_path.name == "Solaris (2002).mkv" # no collision/suffix
def test_rename_to_canonical_from_csfd_year(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "raw.mkv").write_bytes(b"x")
movie = file_manager.import_movie(tmp_path / "raw.mkv", "Matrix") # plain
assert movie.file_path.name == "Matrix.mkv"
movie.csfd_cache = {"year": 1999}
renamed = file_manager.rename_to_canonical(movie)
assert renamed is not None
assert movie.file_path.name == "Matrix (1999).mkv"
assert movie.title == "Matrix"
def test_rename_to_canonical_skips_without_year(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
(tmp_path / "raw.mkv").write_bytes(b"x")
movie = file_manager.import_movie(tmp_path / "raw.mkv", "Unknown")
assert file_manager.rename_to_canonical(movie) is None
assert movie.file_path.name == "Unknown.mkv" # untouched
def test_rename_all_to_canonical_reports(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
for n in ("a.mkv", "b.mkv", "c.mkv"):
(tmp_path / n).write_bytes(b"x")
with_year = file_manager.import_movie(tmp_path / "a.mkv", "Matrix")
with_year.csfd_cache = {"year": 1999}
already = file_manager.import_movie(tmp_path / "b.mkv", "Solaris", year=1972)
_no_year = file_manager.import_movie(tmp_path / "c.mkv", "Mystery")
result = file_manager.rename_all_to_canonical(file_manager.filelist)
assert result["renamed"] == [("Matrix.mkv", "Matrix (1999).mkv")]
assert result["unchanged"] == 1 # Solaris (1972) already canonical
assert result["skipped_no_year"] == ["Mystery.mkv"]
assert already.file_path.name == "Solaris (1972).mkv"
def test_load_pool_movies_reads_from_index(self, file_manager, tmp_path):
file_manager.set_pool_dir(tmp_path / "pool")
source = tmp_path / "raw.mkv"
source.write_bytes(b"x" * 10)
file_manager.import_movie(source, "Matrix", "https://csfd.cz/film/1")
reloaded = FileManager(TagManager())
reloaded.set_pool_dir(tmp_path / "pool")
reloaded.load_pool_movies()
assert len(reloaded.filelist) == 1
assert reloaded.filelist[0].title == "Matrix"
def test_copyasis_folders_default_and_set(self, file_manager, tmp_path):
assert file_manager.copyasis_folders == ["Seriály"]
file_manager.set_pool_dir(tmp_path / "pool")
file_manager.set_copyasis_folders(["Seriály", " Dokumenty ", ""])
assert file_manager.copyasis_folders == ["Seriály", "Dokumenty"]
def test_library_settings_stored_in_index_not_global(self, file_manager, tmp_path):
"""tag_schema / copyasis_folders persist in the pool index, not .!gtag."""
file_manager.set_pool_dir(tmp_path / "pool")
file_manager.set_copyasis_folders(["Seriály", "Dokumenty"])
assert "copyasis_folders" not in file_manager.global_config
assert file_manager.index.get_setting("copyasis_folders") == ["Seriály", "Dokumenty"]
# a fresh manager over the same pool reads the stored settings back
reloaded = FileManager(TagManager())
reloaded.set_pool_dir(tmp_path / "pool")
assert reloaded.copyasis_folders == ["Seriály", "Dokumenty"]
def test_migrates_legacy_global_settings_into_index(self, file_manager, tmp_path):
"""Old settings left in the global config move into the index on open."""
file_manager.set_pool_dir(tmp_path / "pool")
# simulate a pre-migration global config carrying library settings
file_manager.global_config["copyasis_folders"] = ["Seriály", "Filmy jinak"]
file_manager.index = None # force a fresh open + migration
assert file_manager.copyasis_folders == ["Seriály", "Filmy jinak"]
assert "copyasis_folders" not in file_manager.global_config
assert file_manager.index.get_setting("copyasis_folders") == ["Seriály", "Filmy jinak"]