Files
Curator/tests/test_hardlink_manager.py
T

663 lines
23 KiB
Python

import pytest
import os
from src.core.hardlink_manager import HardlinkManager, create_hardlink_structure
from src.core.file import File
from src.core.tag import Tag
from src.core.tag_manager import TagManager
class TestHardlinkManager:
"""Testy pro HardlinkManager"""
@pytest.fixture
def tag_manager(self):
"""Fixture pro TagManager"""
tm = TagManager()
# Remove default tags for cleaner tests
for cat in list(tm.tags_by_category.keys()):
tm.remove_category(cat)
return tm
@pytest.fixture
def temp_source_dir(self, tmp_path):
"""Fixture pro zdrojovou složku s testovacími soubory"""
source_dir = tmp_path / "source"
source_dir.mkdir()
(source_dir / "file1.txt").write_text("content1")
(source_dir / "file2.txt").write_text("content2")
(source_dir / "file3.txt").write_text("content3")
return source_dir
@pytest.fixture
def temp_output_dir(self, tmp_path):
"""Fixture pro výstupní složku"""
output_dir = tmp_path / "output"
output_dir.mkdir()
return output_dir
@pytest.fixture
def files_with_tags(self, temp_source_dir, tag_manager):
"""Fixture pro soubory s tagy"""
files = []
# File 1 with multiple tags
f1 = File(temp_source_dir / "file1.txt", tag_manager)
f1.tags.clear() # ensure a clean tag set
f1.add_tag(Tag("žánr", "Komedie"))
f1.add_tag(Tag("žánr", "Akční"))
f1.add_tag(Tag("rok", "1988"))
files.append(f1)
# File 2 with one tag
f2 = File(temp_source_dir / "file2.txt", tag_manager)
f2.tags.clear() # ensure a clean tag set
f2.add_tag(Tag("žánr", "Drama"))
files.append(f2)
# File 3 with no tags
f3 = File(temp_source_dir / "file3.txt", tag_manager)
f3.tags.clear() # ensure a clean tag set
files.append(f3)
return files
def test_hardlink_manager_creation(self, temp_output_dir):
"""Test vytvoření HardlinkManager"""
manager = HardlinkManager(temp_output_dir)
assert manager.output_dir == temp_output_dir
assert manager.created_links == []
assert manager.errors == []
def test_create_structure_basic(self, files_with_tags, temp_output_dir):
"""Test základního vytvoření struktury"""
manager = HardlinkManager(temp_output_dir)
success, fail = manager.create_structure_for_files(files_with_tags)
# File1 has 3 tags, File2 has 1 tag, File3 has 0 tags
# Should create 4 hardlinks total
assert success == 4
assert fail == 0
# Check directory structure
assert (temp_output_dir / "žánr" / "Komedie" / "file1.txt").exists()
assert (temp_output_dir / "žánr" / "Akční" / "file1.txt").exists()
assert (temp_output_dir / "rok" / "1988" / "file1.txt").exists()
assert (temp_output_dir / "žánr" / "Drama" / "file2.txt").exists()
def test_hardlinks_are_same_inode(self, files_with_tags, temp_output_dir, temp_source_dir):
"""Test že vytvořené soubory jsou opravdu hardlinky (stejný inode)"""
manager = HardlinkManager(temp_output_dir)
manager.create_structure_for_files(files_with_tags)
original = temp_source_dir / "file1.txt"
hardlink = temp_output_dir / "žánr" / "Komedie" / "file1.txt"
# Same inode = hardlink
assert original.stat().st_ino == hardlink.stat().st_ino
def test_create_structure_with_category_filter(self, files_with_tags, temp_output_dir):
"""Test vytvoření struktury jen pro vybrané kategorie"""
manager = HardlinkManager(temp_output_dir)
success, fail = manager.create_structure_for_files(files_with_tags, categories=["žánr"])
# Only "žánr" tags should be processed (3 links)
assert success == 3
assert fail == 0
assert (temp_output_dir / "žánr" / "Komedie" / "file1.txt").exists()
assert not (temp_output_dir / "rok").exists()
def test_create_structure_with_category_roots(self, files_with_tags, temp_output_dir):
"""category_roots: genres sit at the output root, rok under 'Dle roku'."""
manager = HardlinkManager(temp_output_dir)
roots = {"žánr": "", "rok": "Dle roku"}
manager.create_structure_for_files(files_with_tags, category_roots=roots)
# Genres directly at the output root (no "žánr" wrapper folder)
assert (temp_output_dir / "Komedie" / "file1.txt").exists()
assert (temp_output_dir / "Akční" / "file1.txt").exists()
assert (temp_output_dir / "Drama" / "file2.txt").exists()
assert not (temp_output_dir / "žánr").exists()
# Rok grouped under its own "Dle roku" folder
assert (temp_output_dir / "Dle roku" / "1988" / "file1.txt").exists()
def test_sync_with_roots_leaves_unmanaged_mirror_untouched(
self, files_with_tags, temp_source_dir, temp_output_dir
):
"""Cleanup must not delete links in a copy-as-is mirror (e.g. Seriály)."""
manager = HardlinkManager(temp_output_dir)
roots = {"žánr": "", "rok": "Dle roku"}
manager.create_structure_for_files(files_with_tags, category_roots=roots)
# Simulate a copy-as-is mirror holding a hardlink to a source file
mirror = temp_output_dir / "Seriály"
mirror.mkdir()
mirror_link = mirror / "file1.txt"
os.link(temp_source_dir / "file1.txt", mirror_link)
manager.sync_structure(files_with_tags, category_roots=roots)
# The mirror (not a managed tag folder) is left alone
assert mirror_link.exists()
def test_dry_run(self, files_with_tags, temp_output_dir):
"""Test dry run (bez skutečného vytváření)"""
manager = HardlinkManager(temp_output_dir)
success, fail = manager.create_structure_for_files(files_with_tags, dry_run=True)
assert success == 4
assert fail == 0
# No actual files should be created
assert not (temp_output_dir / "žánr").exists()
def test_get_preview(self, files_with_tags, temp_output_dir):
"""Test náhledu co bude vytvořeno"""
manager = HardlinkManager(temp_output_dir)
preview = manager.get_preview(files_with_tags)
assert len(preview) == 4
# Check that preview contains expected paths
targets = [p[1] for p in preview]
assert temp_output_dir / "žánr" / "Komedie" / "file1.txt" in targets
assert temp_output_dir / "žánr" / "Drama" / "file2.txt" in targets
def test_get_preview_with_category_filter(self, files_with_tags, temp_output_dir):
"""Test náhledu s filtrem kategorií"""
manager = HardlinkManager(temp_output_dir)
preview = manager.get_preview(files_with_tags, categories=["rok"])
assert len(preview) == 1
assert preview[0][1] == temp_output_dir / "rok" / "1988" / "file1.txt"
def test_remove_created_links(self, files_with_tags, temp_output_dir):
"""Test odstranění vytvořených hardlinků"""
manager = HardlinkManager(temp_output_dir)
manager.create_structure_for_files(files_with_tags)
# Verify links exist
assert (temp_output_dir / "žánr" / "Komedie" / "file1.txt").exists()
# Remove links
removed = manager.remove_created_links()
assert removed == 4
# Links should be gone
assert not (temp_output_dir / "žánr" / "Komedie" / "file1.txt").exists()
# Empty directories should also be removed
assert not (temp_output_dir / "žánr" / "Komedie").exists()
def test_empty_files_list(self, temp_output_dir):
"""Test s prázdným seznamem souborů"""
manager = HardlinkManager(temp_output_dir)
success, fail = manager.create_structure_for_files([])
assert success == 0
assert fail == 0
def test_files_without_tags(self, temp_source_dir, temp_output_dir, tag_manager):
"""Test se soubory bez tagů"""
f1 = File(temp_source_dir / "file1.txt", tag_manager)
f1.tags.clear() # Remove default tags
manager = HardlinkManager(temp_output_dir)
success, fail = manager.create_structure_for_files([f1])
assert success == 0
assert fail == 0
def test_duplicate_link_same_file(self, files_with_tags, temp_output_dir):
"""Test že existující hardlink na stejný soubor je přeskočen"""
manager = HardlinkManager(temp_output_dir)
# Create first time
success1, _ = manager.create_structure_for_files(files_with_tags)
# Create second time - should skip existing
manager2 = HardlinkManager(temp_output_dir)
success2, fail2 = manager2.create_structure_for_files(files_with_tags)
# All should be skipped (same inode)
assert success2 == 0
assert fail2 == 0
def test_unique_name_on_conflict(self, temp_source_dir, temp_output_dir, tag_manager):
"""Test že při konfliktu (jiný soubor) se použije unikátní jméno"""
# Create first file
f1 = File(temp_source_dir / "file1.txt", tag_manager)
f1.tags.clear()
f1.add_tag(Tag("test", "tag"))
manager = HardlinkManager(temp_output_dir)
manager.create_structure_for_files([f1])
# Create different file with same name in different location
source2 = temp_source_dir / "subdir"
source2.mkdir()
(source2 / "file1.txt").write_text("different content")
f2 = File(source2 / "file1.txt", tag_manager)
f2.tags.clear()
f2.add_tag(Tag("test", "tag"))
# Should create file1_1.txt
manager2 = HardlinkManager(temp_output_dir)
success, fail = manager2.create_structure_for_files([f2])
assert success == 1
assert (temp_output_dir / "test" / "tag" / "file1_1.txt").exists()
def test_czech_characters_in_tags(self, temp_source_dir, temp_output_dir, tag_manager):
"""Test českých znaků v názvech tagů"""
f1 = File(temp_source_dir / "file1.txt", tag_manager)
f1.tags.clear()
f1.add_tag(Tag("Žánr", "Česká komedie"))
f1.add_tag(Tag("Štítky", "Příběh"))
manager = HardlinkManager(temp_output_dir)
success, fail = manager.create_structure_for_files([f1])
assert success == 2
assert fail == 0
assert (temp_output_dir / "Žánr" / "Česká komedie" / "file1.txt").exists()
assert (temp_output_dir / "Štítky" / "Příběh" / "file1.txt").exists()
class TestConvenienceFunction:
"""Testy pro convenience funkci create_hardlink_structure"""
@pytest.fixture
def tag_manager(self):
tm = TagManager()
for cat in list(tm.tags_by_category.keys()):
tm.remove_category(cat)
return tm
@pytest.fixture
def temp_files(self, tmp_path, tag_manager):
source = tmp_path / "source"
source.mkdir()
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag"))
return [f]
def test_create_hardlink_structure_function(self, temp_files, tmp_path):
"""Test convenience funkce"""
output = tmp_path / "output"
output.mkdir()
success, fail, errors = create_hardlink_structure(temp_files, output)
assert success == 1
assert fail == 0
assert len(errors) == 0
assert (output / "cat" / "tag" / "file.txt").exists()
def test_create_hardlink_structure_with_categories(self, tmp_path, tag_manager):
"""Test convenience funkce s filtrem kategorií"""
source = tmp_path / "source"
source.mkdir()
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("include", "yes"))
f.add_tag(Tag("exclude", "no"))
output = tmp_path / "output"
output.mkdir()
success, fail, errors = create_hardlink_structure([f], output, categories=["include"])
assert success == 1
assert (output / "include" / "yes" / "file.txt").exists()
assert not (output / "exclude").exists()
class TestSyncStructure:
"""Testy pro synchronizaci hardlink struktury"""
@pytest.fixture
def tag_manager(self):
tm = TagManager()
for cat in list(tm.tags_by_category.keys()):
tm.remove_category(cat)
return tm
@pytest.fixture
def setup_dirs(self, tmp_path):
source = tmp_path / "source"
source.mkdir()
output = tmp_path / "output"
output.mkdir()
return source, output
def test_find_obsolete_links_empty_output(self, setup_dirs, tag_manager):
"""Test find_obsolete_links s prázdným výstupem"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag"))
manager = HardlinkManager(output)
obsolete = manager.find_obsolete_links([f])
assert obsolete == []
def test_find_obsolete_links_detects_removed_tag(self, setup_dirs, tag_manager):
"""Test že find_obsolete_links najde hardlink pro odebraný tag"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag1"))
f.add_tag(Tag("cat", "tag2"))
# Create structure with both tags
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
assert (output / "cat" / "tag1" / "file.txt").exists()
assert (output / "cat" / "tag2" / "file.txt").exists()
# Remove one tag from file
f.tags.clear()
f.add_tag(Tag("cat", "tag1")) # Only tag1 remains
# Find obsolete
obsolete = manager.find_obsolete_links([f])
assert len(obsolete) == 1
assert obsolete[0][0] == output / "cat" / "tag2" / "file.txt"
def test_remove_obsolete_links(self, setup_dirs, tag_manager):
"""Test odstranění zastaralých hardlinků"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag1"))
f.add_tag(Tag("cat", "tag2"))
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
# Remove tag2
f.tags.clear()
f.add_tag(Tag("cat", "tag1"))
# Remove obsolete links
removed, paths = manager.remove_obsolete_links([f])
assert removed == 1
assert not (output / "cat" / "tag2" / "file.txt").exists()
assert (output / "cat" / "tag1" / "file.txt").exists()
def test_remove_obsolete_links_dry_run(self, setup_dirs, tag_manager):
"""Test dry run pro remove_obsolete_links"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag1"))
f.add_tag(Tag("cat", "tag2"))
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
f.tags.clear()
f.add_tag(Tag("cat", "tag1"))
removed, paths = manager.remove_obsolete_links([f], dry_run=True)
assert removed == 1
# File should still exist (dry run)
assert (output / "cat" / "tag2" / "file.txt").exists()
def test_sync_structure_creates_and_removes(self, setup_dirs, tag_manager):
"""Test sync_structure vytvoří nové a odstraní staré hardlinky"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "old_tag"))
# Create initial structure
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
assert (output / "cat" / "old_tag" / "file.txt").exists()
# Change tags
f.tags.clear()
f.add_tag(Tag("cat", "new_tag"))
# Sync
created, c_fail, removed, r_fail = manager.sync_structure([f])
assert created == 1
assert removed == 1
assert c_fail == 0
assert r_fail == 0
assert not (output / "cat" / "old_tag").exists()
assert (output / "cat" / "new_tag" / "file.txt").exists()
def test_sync_structure_no_changes_needed(self, setup_dirs, tag_manager):
"""Test sync_structure když není potřeba žádná změna"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag"))
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
# Sync again without changes
created, c_fail, removed, r_fail = manager.sync_structure([f])
# Nothing should change (existing links are skipped)
assert removed == 0
assert (output / "cat" / "tag" / "file.txt").exists()
def test_find_obsolete_with_category_filter(self, setup_dirs, tag_manager):
"""Test find_obsolete_links s filtrem kategorií"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat1", "tag"))
f.add_tag(Tag("cat2", "tag"))
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
# Remove both tags
f.tags.clear()
# Find obsolete only in cat1
obsolete = manager.find_obsolete_links([f], categories=["cat1"])
assert len(obsolete) == 1
assert obsolete[0][0] == output / "cat1" / "tag" / "file.txt"
def test_removes_empty_directories(self, setup_dirs, tag_manager):
"""Test že prázdné adresáře jsou odstraněny po sync"""
source, output = setup_dirs
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("category", "tag"))
manager = HardlinkManager(output)
manager.create_structure_for_files([f])
# Remove all tags
f.tags.clear()
manager.remove_obsolete_links([f])
# Directory should be gone
assert not (output / "category" / "tag").exists()
assert not (output / "category").exists()
class TestEdgeCases:
"""Testy pro okrajové případy"""
@pytest.fixture
def tag_manager(self):
tm = TagManager()
for cat in list(tm.tags_by_category.keys()):
tm.remove_category(cat)
return tm
def test_nonexistent_output_dir_created(self, tmp_path, tag_manager):
"""Test že výstupní složka je vytvořena pokud neexistuje"""
source = tmp_path / "source"
source.mkdir()
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag"))
output = tmp_path / "output" / "nested" / "deep"
# output doesn't exist
manager = HardlinkManager(output)
success, fail = manager.create_structure_for_files([f])
assert success == 1
assert (output / "cat" / "tag" / "file.txt").exists()
def test_special_characters_in_filename(self, tmp_path, tag_manager):
"""Test souboru se speciálními znaky v názvu"""
source = tmp_path / "source"
source.mkdir()
(source / "file with spaces (2024).txt").write_text("content")
f = File(source / "file with spaces (2024).txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("test", "tag"))
output = tmp_path / "output"
output.mkdir()
manager = HardlinkManager(output)
success, fail = manager.create_structure_for_files([f])
assert success == 1
assert (output / "test" / "tag" / "file with spaces (2024).txt").exists()
def test_empty_category_filter(self, tmp_path, tag_manager):
"""Test s prázdným seznamem kategorií"""
source = tmp_path / "source"
source.mkdir()
(source / "file.txt").write_text("content")
f = File(source / "file.txt", tag_manager)
f.tags.clear()
f.add_tag(Tag("cat", "tag"))
output = tmp_path / "output"
output.mkdir()
manager = HardlinkManager(output)
# Empty list = no categories = no links
success, fail = manager.create_structure_for_files([f], categories=[])
assert success == 0
def test_is_same_file_method(self, tmp_path):
"""Test metody _is_same_file"""
file1 = tmp_path / "file1.txt"
file1.write_text("content")
link = tmp_path / "link.txt"
os.link(file1, link)
file2 = tmp_path / "file2.txt"
file2.write_text("different")
manager = HardlinkManager(tmp_path)
# Same inode
assert manager._is_same_file(file1, link) is True
# Different inode
assert manager._is_same_file(file1, file2) is False
# Non-existent file
assert manager._is_same_file(file1, tmp_path / "nonexistent") is False
def test_get_unique_name_method(self, tmp_path):
"""Test metody _get_unique_name"""
(tmp_path / "file.txt").write_text("1")
(tmp_path / "file_1.txt").write_text("2")
(tmp_path / "file_2.txt").write_text("3")
manager = HardlinkManager(tmp_path)
unique = manager._get_unique_name(tmp_path / "file.txt")
assert unique == tmp_path / "file_3.txt"
class TestMirrorAsIs:
"""Testy pro copy-as-is zrcadlení (Seriály)"""
def test_mirror_clones_hierarchy_with_hardlinks(self, tmp_path):
"""Adresářová struktura se zrcadlí 1:1 a soubory jsou hardlinky"""
source = tmp_path / "Seriály"
(source / "Show" / "S01").mkdir(parents=True)
ep1 = source / "Show" / "S01" / "ep1.mkv"
ep2 = source / "Show" / "S01" / "ep2.mkv"
ep1.write_text("a")
ep2.write_text("b")
output = tmp_path / "out"
manager = HardlinkManager(output)
created, failed = manager.mirror_as_is(source, "Seriály")
assert failed == 0
assert created == 2
linked = output / "Seriály" / "Show" / "S01" / "ep1.mkv"
assert linked.exists()
assert linked.stat().st_ino == ep1.stat().st_ino
def test_mirror_skips_curator_metadata(self, tmp_path):
"""Metadata soubory (.!tag, .!index) se nezrcadlí"""
source = tmp_path / "Seriály"
source.mkdir()
(source / "ep1.mkv").write_text("a")
(source / ".ep1.mkv.!tag").write_text("{}")
(source / ".Curator.!index").write_text("{}")
output = tmp_path / "out"
manager = HardlinkManager(output)
created, failed = manager.mirror_as_is(source, "Seriály")
assert created == 1
assert failed == 0
assert not (output / "Seriály" / ".ep1.mkv.!tag").exists()
def test_mirror_nonexistent_source_is_noop(self, tmp_path):
"""Neexistující zdroj nic neudělá"""
manager = HardlinkManager(tmp_path / "out")
assert manager.mirror_as_is(tmp_path / "missing", "Seriály") == (0, 0)