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_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"]