mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-25 18:40:06 -04:00
04779b8557
* Preview markdown files formatted, with a raw toggle Markdown files are typed text/markdown (libmagic reports them as text/plain, or as a programming language when they hold code) and keep the text indexer, so they stay deep-searchable. Existing files switch type on their next refresh. The file preview renders markdown with markdown-it into the page, so it follows the theme tokens and the night media filter. A Formatted/Raw toggle shows the source instead, and the choice is remembered. Relative links and images resolve against the file's directory; URLs that climb out of /media/ are dropped. Raw HTML in a file is escaped rather than parsed, and the output is also sanitized with DOMPurify. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Do not load images from other sites in the markdown preview An image from another site (http:, https:, or protocol-relative //) loaded as soon as the preview opened, telling that site who was reading the file and when. Such an image now becomes a link labeled with its alt text, so the reader chooses whether to follow it. Inside a link (a README badge) it becomes the outer link's text, since links cannot nest. Media and embedded data: images still load. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1898 lines
74 KiB
Python
1898 lines
74 KiB
Python
import json
|
|
import pathlib
|
|
import shutil
|
|
import zipfile
|
|
from http import HTTPStatus
|
|
from pathlib import Path
|
|
from typing import List
|
|
from uuid import uuid4
|
|
|
|
import mock
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
import wrolpi.common
|
|
from modules.videos import Video
|
|
from wrolpi.common import timer, get_wrolpi_config
|
|
from wrolpi.conftest import await_switches
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from wrolpi.errors import InvalidFile, UnknownDirectory, FileGroupIsTagged, NoPrimaryFile
|
|
from wrolpi.files import lib, indexers
|
|
from wrolpi.files.models import FileGroup, Directory
|
|
from wrolpi.tags import TagFile
|
|
from wrolpi.test.common import only_macos, skip_circleci
|
|
from wrolpi.vars import PROJECT_DIR, IS_MACOS
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file(async_client, test_session, make_files_structure, test_directory):
|
|
"""
|
|
File in the media directory can be deleted.
|
|
"""
|
|
make_files_structure([
|
|
'archives/foo.txt',
|
|
'bar.txt',
|
|
'baz/',
|
|
])
|
|
|
|
await lib.delete('bar.txt')
|
|
assert (test_directory / 'archives/foo.txt').is_file()
|
|
assert not (test_directory / 'bar.txt').exists()
|
|
assert (test_directory / 'baz').is_dir()
|
|
|
|
await lib.delete('archives/foo.txt')
|
|
assert not (test_directory / 'archives/foo.txt').exists()
|
|
assert not (test_directory / 'bar.txt').exists()
|
|
|
|
# Can also delete directories.
|
|
await lib.delete('baz')
|
|
assert not (test_directory / 'baz').exists()
|
|
|
|
# Trying to delete a file that does not exist only logs a warning.
|
|
await lib.delete('does not exist')
|
|
|
|
# Cannot delete the media directory.
|
|
with pytest.raises(InvalidFile):
|
|
await lib.delete('.')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_multiple(async_client, test_session, make_files_structure, test_directory):
|
|
"""Multiple files can be deleted at once."""
|
|
foo, bar, baz = make_files_structure([
|
|
'archives/foo.txt',
|
|
'archives/bar.txt',
|
|
'archives/baz.txt',
|
|
])
|
|
assert foo.is_file()
|
|
assert bar.is_file()
|
|
assert baz.is_file()
|
|
|
|
await lib.delete('archives/foo.txt', 'archives/bar.txt', 'archives/baz.txt')
|
|
assert not foo.is_file()
|
|
assert not bar.is_file()
|
|
assert not baz.is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_missing(async_client, test_session, make_files_structure, test_directory):
|
|
"""Deleting a mix of existing and nonexistent files should succeed, deleting only the existing files."""
|
|
foo, bar = make_files_structure([
|
|
'config/foo.txt',
|
|
'config/bar.txt',
|
|
])
|
|
|
|
await lib.delete('config/foo.txt', 'config/._bar.txt', 'config/._baz.txt')
|
|
assert not foo.is_file()
|
|
assert bar.is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_names(async_client, test_session, make_files_structure, test_directory, tag_factory):
|
|
"""Will not refuse to delete a file that shares the name of a nearby file when they are in different FileGroups."""
|
|
foo, foo1 = make_files_structure({
|
|
'archives/foo': 'text',
|
|
'archives/foo1': 'text',
|
|
})
|
|
foo_fg = FileGroup.from_paths(test_session, foo)
|
|
foo1_fg = FileGroup.from_paths(test_session, foo1)
|
|
tag = await tag_factory()
|
|
foo1_fg.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
assert foo.is_file()
|
|
assert foo1.is_file()
|
|
|
|
await lib.delete('archives/foo')
|
|
assert not foo.is_file()
|
|
assert foo1.is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_link(async_client, test_session, test_directory):
|
|
"""Links can be deleted."""
|
|
foo, bar = test_directory / 'foo', test_directory / 'bar'
|
|
foo.touch()
|
|
bar.symlink_to(foo)
|
|
|
|
await lib.delete(bar)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_tagged(await_switches, test_session, test_directory, make_files_structure, tag_factory,
|
|
video_bytes, refresh_files):
|
|
"""Cannot delete a file that has been tagged."""
|
|
tag = await tag_factory()
|
|
make_files_structure({'foo/bar.txt': 'asdf', 'foo/bar.mp4': video_bytes})
|
|
await refresh_files()
|
|
# Both files end up in a group.
|
|
bar = test_session.query(FileGroup).one()
|
|
bar.add_tag(test_session, tag.id)
|
|
await await_switches()
|
|
test_session.commit()
|
|
|
|
# Neither file can be deleted without force; tagged file groups are returned.
|
|
result = await lib.delete('foo/bar.txt')
|
|
assert result and len(result) == 1
|
|
result = await lib.delete('foo/bar.mp4')
|
|
assert result and len(result) == 1
|
|
result = await lib.delete('foo')
|
|
assert result and len(result) == 1
|
|
|
|
# Force delete succeeds and actually removes the files.
|
|
await lib.delete('foo', force=True)
|
|
assert not (test_directory / 'foo').exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tagged_file_groups_by_ids(test_session, test_directory, make_files_structure, tag_factory,
|
|
refresh_files, await_switches):
|
|
"""`get_tagged_file_groups_by_ids` returns only the FileGroups with at least one TagFile."""
|
|
tag = await tag_factory()
|
|
make_files_structure({'a.txt': 'a', 'b.txt': 'b', 'c.txt': 'c'})
|
|
await refresh_files()
|
|
|
|
fgs = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
|
|
assert len(fgs) == 3
|
|
a_fg, b_fg, c_fg = fgs
|
|
|
|
# Tag two of the three FileGroups.
|
|
a_fg.add_tag(test_session, tag.id)
|
|
c_fg.add_tag(test_session, tag.id)
|
|
await await_switches()
|
|
test_session.commit()
|
|
|
|
# Empty input → empty list.
|
|
assert lib.get_tagged_file_groups_by_ids(test_session, []) == []
|
|
|
|
# Untagged FileGroup is excluded.
|
|
assert lib.get_tagged_file_groups_by_ids(test_session, [b_fg.id]) == []
|
|
|
|
# All-untagged batch is empty.
|
|
result = lib.get_tagged_file_groups_by_ids(test_session, [a_fg.id, b_fg.id, c_fg.id])
|
|
returned_ids = {fg['id'] for fg in result}
|
|
assert returned_ids == {a_fg.id, c_fg.id}
|
|
# Each returned FileGroup is serialized and includes its tag names.
|
|
for fg in result:
|
|
assert tag.name in fg.get('tags', [])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_nested(test_session, make_files_structure):
|
|
"""Refuse to delete nested files in case user mis-clicks."""
|
|
make_files_structure(['foo/bar'])
|
|
|
|
with pytest.raises(InvalidFile):
|
|
await lib.delete('foo', 'foo/bar')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_similar_names(async_client, test_session, make_files_structure):
|
|
"""Deleting sibling files where one filename is a prefix of another should not be considered nested."""
|
|
make_files_structure([
|
|
'config/._wrolpi.yaml',
|
|
'config/._wrolpi.yaml.sb-98fa34a1-MG7phT',
|
|
'config/._wrolpi.yaml.sb-98fa34a1-N4k8NB',
|
|
])
|
|
|
|
await lib.delete('config/._wrolpi.yaml', 'config/._wrolpi.yaml.sb-98fa34a1-MG7phT',
|
|
'config/._wrolpi.yaml.sb-98fa34a1-N4k8NB')
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'path,full,expected',
|
|
[
|
|
('foo', False, ('foo', '')),
|
|
('foo.mp4', False, ('foo', '.mp4')),
|
|
('foo.info.json', False, ('foo', '.info.json')),
|
|
('foo.ffprobe.json', False, ('foo', '.ffprobe.json')),
|
|
('foo.something.info.json', False, ('foo.something', '.info.json')),
|
|
('foo-something.info.json', False, ('foo-something', '.info.json')),
|
|
('/absolute/foo-something.info.json', False, ('foo-something', '.info.json')),
|
|
('/absolute/foo', False, ('foo', '')),
|
|
('/absolute/foo.bar', False, ('foo', '.bar')),
|
|
('foo.en.srt', False, ('foo', '.en.srt')),
|
|
('foo.pl.srt', False, ('foo', '.pl.srt')),
|
|
('foo.en-US.srt', False, ('foo', '.en-US.srt')),
|
|
('foo.en-US.vtt', False, ('foo', '.en-US.vtt')),
|
|
('foo.en-us.vtt', False, ('foo', '.en-us.vtt')),
|
|
('foo.en-AUTO.srt', False, ('foo', '.en-AUTO.srt')),
|
|
('foo.en-auto.vtt', False, ('foo', '.en-auto.vtt')),
|
|
# Absolute path can be returned.
|
|
('/absolute//foo.bar', True, ('/absolute/foo', '.bar')),
|
|
# Case is preserved.
|
|
('foo.EN.SRT', False, ('foo', '.EN.SRT')),
|
|
('foo.INFO.JSON', False, ('foo', '.INFO.JSON')),
|
|
('foo.FFPROBE.JSON', False, ('foo', '.FFPROBE.JSON')),
|
|
# Part files from yt-dlp.
|
|
('foo.webm.part', False, ('foo', '.webm.part')),
|
|
('foo.f248.webm.part', False, ('foo', '.f248.webm.part')),
|
|
('foo.info.json.part', False, ('foo', '.info.json.part')),
|
|
('/absolute/foo.webm.part', True, ('/absolute/foo', '.webm.part')),
|
|
])
|
|
def test_split_path_stem_and_suffix(path, full, expected):
|
|
assert lib.split_path_stem_and_suffix(Path(path), full) == expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_files(async_client, test_session, make_files_structure, assert_file_groups, refresh_files):
|
|
"""All files in the media directory should be found when calling `refresh_files`"""
|
|
files = make_files_structure([
|
|
'foo.txt',
|
|
'bar.txt',
|
|
'baz.txt',
|
|
])
|
|
foo, bar, baz = files
|
|
|
|
await refresh_files()
|
|
assert_file_groups([
|
|
{'primary_path': 'foo.txt', 'indexed': True},
|
|
{'primary_path': 'bar.txt', 'indexed': True},
|
|
{'primary_path': 'baz.txt', 'indexed': True}])
|
|
|
|
baz.unlink()
|
|
|
|
await refresh_files()
|
|
assert_file_groups([{'primary_path': 'foo.txt', 'indexed': True}, {'primary_path': 'bar.txt', 'indexed': True}])
|
|
|
|
foo.unlink()
|
|
|
|
await refresh_files()
|
|
assert_file_groups([{'primary_path': 'bar.txt', 'indexed': True}])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_populates_suffix(async_client, test_session, make_files_structure, refresh_files):
|
|
"""A refresh populates `file_group.suffix` for every file it discovers.
|
|
|
|
Regression: suffix is path-derived metadata (like mimetype) and must be set when the FileGroup is
|
|
upserted during discovery, not deferred to content-indexing. Deferring it meant a file that got marked
|
|
indexed through a path that skipped that step ended up indexed with a NULL suffix and was invisible to
|
|
suffix search (the firmware flasher's media picker)."""
|
|
make_files_structure([
|
|
'software/firmware.bin',
|
|
'software/bffb/esp32_marauder.ino.bootloader.bin',
|
|
'notes.txt',
|
|
])
|
|
|
|
await refresh_files()
|
|
|
|
file_groups = test_session.query(FileGroup).all()
|
|
# Suffix is set for every file the moment it is discovered, and every row is indexed.
|
|
assert all(fg.indexed for fg in file_groups)
|
|
suffixes = {fg.primary_path.name: fg.suffix for fg in file_groups}
|
|
assert suffixes == {
|
|
'firmware.bin': '.bin',
|
|
'esp32_marauder.ino.bootloader.bin': '.bin',
|
|
'notes.txt': '.txt',
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_group_location(async_client, test_session, make_files_structure, refresh_files):
|
|
"""FileGroup can return the URL necessary to preview the FileGroup."""
|
|
make_files_structure({
|
|
'foo/one.txt': 'one',
|
|
'two.txt': 'two',
|
|
})
|
|
await refresh_files()
|
|
|
|
one, two = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
|
|
assert one.location == '/files?folders=foo&preview=foo%2Fone.txt'
|
|
assert two.location == '/files?preview=two.txt'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_bogus_files(async_client, test_session, make_files_structure, test_directory,
|
|
assert_file_groups, insert_file_group, refresh_files):
|
|
"""Bogus files are removed during a refresh."""
|
|
make_files_structure(['does exist.txt'])
|
|
await refresh_files()
|
|
insert_file_group([test_directory / 'does not exist.txt'])
|
|
test_session.commit()
|
|
|
|
# Bogus file was inserted.
|
|
assert_file_groups([
|
|
{'primary_path': 'does exist.txt', 'indexed': True},
|
|
{'primary_path': 'does not exist.txt', 'indexed': True},
|
|
])
|
|
|
|
await refresh_files()
|
|
|
|
assert test_session.query(FileGroup).count() == 1, 'Bogus file was not removed'
|
|
assert_file_groups([{'primary_path': 'does exist.txt', 'indexed': True}])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_empty_media_directory(async_client, test_session, test_directory, refresh_files):
|
|
"""refresh_paths will refuse to refresh with an empty media directory."""
|
|
with pytest.raises(UnknownDirectory):
|
|
await refresh_files()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test__upsert_files(test_session, make_files_structure, test_directory, assert_file_groups, video_file,
|
|
srt_file3):
|
|
baz, bar = make_files_structure({
|
|
'dir1/bar.txt': None,
|
|
'baz.txt': 'baz file',
|
|
})
|
|
video_file = video_file.rename(test_directory / 'video.mp4')
|
|
srt_file3 = srt_file3.rename(test_directory / 'video.en.srt')
|
|
|
|
# All files are found because they are in this refresh request, or in the `dir1` directory.
|
|
lib._upsert_files([video_file, srt_file3, bar, baz])
|
|
# Note: files now store relative filenames only (not absolute paths)
|
|
assert_file_groups([
|
|
{'primary_path': video_file, 'indexed': False, 'stem': 'video',
|
|
'files': [
|
|
{'path': srt_file3.name, 'size': 951, 'suffix': '.en.srt', 'mimetype': 'text/srt'},
|
|
{'path': video_file.name, 'size': 1056318, 'suffix': '.mp4', 'mimetype': 'video/mp4'},
|
|
]},
|
|
{'primary_path': bar, 'indexed': False, 'stem': 'bar',
|
|
'files': [{'path': bar.name, 'size': 0, 'suffix': '.txt', 'mimetype': 'inode/x-empty'}]
|
|
},
|
|
{'primary_path': baz, 'indexed': False, 'stem': 'baz',
|
|
'files': [{'path': baz.name, 'size': 8, 'suffix': '.txt', 'mimetype': 'text/plain'}]
|
|
},
|
|
])
|
|
|
|
# Modified files should be re-indexed.
|
|
bar.touch()
|
|
baz.write_text('new baz')
|
|
# Simulate that the video was indexed, it should not be re-indexed because it was not modified.
|
|
test_session.query(FileGroup).filter_by(primary_path=video_file).one().indexed = True
|
|
test_session.commit()
|
|
assert_file_groups(
|
|
[{'primary_path': str(video_file), 'indexed': True}],
|
|
assert_count=False)
|
|
|
|
# Only modified files need to be re-indexed.
|
|
lib._upsert_files([video_file, srt_file3, bar, baz])
|
|
# Note: files now store relative filenames only (not absolute paths)
|
|
assert_file_groups([
|
|
{'primary_path': video_file, 'indexed': True, 'stem': 'video',
|
|
'files': [
|
|
{'path': srt_file3.name, 'size': 951, 'suffix': '.en.srt', 'mimetype': 'text/srt'},
|
|
{'path': video_file.name, 'size': 1056318, 'suffix': '.mp4', 'mimetype': 'video/mp4'},
|
|
]},
|
|
{'primary_path': bar, 'indexed': False,
|
|
'files': [{'path': bar.name, 'size': 0, 'suffix': '.txt', 'mimetype': 'inode/x-empty'}],
|
|
},
|
|
{'primary_path': baz, 'indexed': False,
|
|
'files': [{'path': baz.name, 'size': 7, 'suffix': '.txt', 'mimetype': 'text/plain'}],
|
|
},
|
|
])
|
|
|
|
# Deleting SRT removes it from the video.
|
|
srt_file3.unlink()
|
|
lib._upsert_files([video_file, bar, baz])
|
|
video_file_group: FileGroup = test_session.query(FileGroup).filter_by(primary_path=str(video_file)).one()
|
|
assert len(video_file_group.files) == 1, 'SRT file was not removed from files'
|
|
# Note: files now store relative filenames only (not absolute paths)
|
|
assert_file_groups([
|
|
# Video is no longer indexed because SRT was removed.
|
|
{'primary_path': video_file, 'indexed': False,
|
|
'files': [{'path': video_file.name, 'size': 1056318, 'suffix': '.mp4', 'mimetype': 'video/mp4'}]},
|
|
{'primary_path': bar, 'indexed': False,
|
|
'files': [{'path': bar.name, 'size': 0, 'suffix': '.txt', 'mimetype': 'inode/x-empty'}],
|
|
},
|
|
{'primary_path': baz, 'indexed': False,
|
|
'files': [{'path': baz.name, 'size': 7, 'suffix': '.txt', 'mimetype': 'text/plain'}],
|
|
},
|
|
])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_paths(test_session, make_files_structure, test_directory, assert_files,
|
|
refresh_files):
|
|
"""Refreshing a file, a directory, or missing files updates FileGroups via FileWorker."""
|
|
foo, bar, baz = make_files_structure(['dir1/foo.txt', 'dir1/bar.txt', 'baz.txt'])
|
|
dir1 = foo.parent
|
|
|
|
# Refreshing one file only discovers that file.
|
|
await refresh_files([foo])
|
|
assert_files([
|
|
{'path': 'dir1/foo.txt'},
|
|
])
|
|
|
|
# Refreshing a directory discovers recursively.
|
|
await refresh_files([foo.parent])
|
|
assert_files([
|
|
{'path': 'dir1/foo.txt'},
|
|
{'path': 'dir1/bar.txt'},
|
|
])
|
|
|
|
# Refreshing the media directory discovers the file at the top.
|
|
await refresh_files([test_directory])
|
|
assert_files([
|
|
{'path': 'dir1/foo.txt'},
|
|
{'path': 'dir1/bar.txt'},
|
|
{'path': 'baz.txt'},
|
|
])
|
|
|
|
# Records for deleted files are deleted. Request a refresh of `dir1` so we indirectly refresh `foo`.
|
|
foo.unlink()
|
|
await refresh_files([dir1])
|
|
assert_files([
|
|
{'path': 'dir1/bar.txt'},
|
|
{'path': 'baz.txt'},
|
|
])
|
|
|
|
# Records for all children of a directory are deleted.
|
|
bar.unlink()
|
|
dir1.rmdir()
|
|
await refresh_files([dir1])
|
|
assert_files([
|
|
{'path': 'baz.txt'},
|
|
])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_groups(test_session, make_files_structure, test_directory, video_bytes, refresh_files):
|
|
make_files_structure({'dir1/foo.mp4': video_bytes, 'dir1/foo.info.json': 'hello', 'baz.txt': 'hello'})
|
|
await refresh_files([test_directory])
|
|
|
|
# Two "foo" files, one "baz" file.
|
|
assert test_session.query(FileGroup).count() == 2
|
|
|
|
baz, foo = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
|
|
# "foo" files are related to the "foo" group. Modeling may add sidecars (.ffprobe.json).
|
|
foo_names = {i['path'].name for i in foo.my_files()}
|
|
assert {'foo.info.json', 'foo.mp4'} <= foo_names
|
|
assert foo.stem == 'foo'
|
|
# "baz" file is the only file related to the "baz" group.
|
|
assert len(baz.my_files()) == 1
|
|
assert str(baz.my_files()[0]['path'].relative_to(test_directory)) == 'baz.txt'
|
|
assert baz.indexed is True # leftover indexer claims plain text
|
|
assert baz.stem == 'baz'
|
|
|
|
|
|
def test_find_indexer_longest_prefix():
|
|
"""Longest registered mimetype prefix wins, even if a shorter key was inserted later."""
|
|
# Simulate the old fallback, which did indexer_map[mimetype.split('/')[0]] and inserted
|
|
# DefaultIndexer for 'text' into the defaultdict.
|
|
indexers.indexer_map['text']
|
|
try:
|
|
assert indexers.find_indexer('text/html') is indexers.HTMLIndexer
|
|
assert indexers.find_indexer('text/plain') is indexers.TextIndexer
|
|
assert indexers.find_indexer('text/markdown') is indexers.TextIndexer
|
|
assert indexers.find_indexer('text/csv') is indexers.DefaultIndexer
|
|
assert indexers.find_indexer('application/zip') is indexers.ZipIndexer
|
|
finally:
|
|
indexers.indexer_map.pop('text', None)
|
|
|
|
|
|
def test_ensure_file_group_stems_recomputes_on_version_change(test_session, test_directory, make_files_structure):
|
|
"""A stored stem is a cache of the grouping algorithm; a version bump must backfill."""
|
|
foo, = make_files_structure(['foo.en.srt'])
|
|
fg = FileGroup.from_paths(test_session, foo)
|
|
test_session.commit()
|
|
assert fg.stem == 'foo'
|
|
fg.stem = 'stale-stem'
|
|
test_session.commit()
|
|
|
|
lib.set_stem_algorithm_version(0)
|
|
lib.ensure_file_group_stems()
|
|
test_session.expire(fg)
|
|
assert fg.stem == 'foo'
|
|
assert lib.get_stem_algorithm_version() == lib.STEM_ALGORITHM_VERSION
|
|
|
|
|
|
def test_file_group_directory_stem_is_unique(test_session, test_directory, make_files_structure):
|
|
"""Two FileGroups in the same directory cannot share a stem."""
|
|
foo_mp4, foo_png = make_files_structure(['foo.mp4', 'foo.png'])
|
|
FileGroup.from_paths(test_session, foo_mp4)
|
|
test_session.commit()
|
|
|
|
other = FileGroup(
|
|
directory=foo_png.parent,
|
|
primary_path=foo_png,
|
|
stem='foo',
|
|
files=[{'path': foo_png.name, 'mimetype': 'image/png'}],
|
|
)
|
|
test_session.add(other)
|
|
with pytest.raises(IntegrityError):
|
|
test_session.commit()
|
|
|
|
|
|
def test_choose_primary_file_is_sorted_and_set_stable(test_directory, make_files_structure):
|
|
"""Unmodeled groups must pick the same primary regardless of set iteration order."""
|
|
csv, txt = make_files_structure(['n.csv', 'n.txt'])
|
|
assert lib.choose_primary_file([txt, csv]) == csv
|
|
assert lib.choose_primary_file({txt, csv}) == csv
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_keeps_tag_when_sidecar_added(async_client, test_session, test_directory,
|
|
make_files_structure, tag_factory, refresh_files):
|
|
"""Adding a sidecar to a tagged unmodeled FileGroup must not drop the tag on refresh."""
|
|
txt, = make_files_structure(['note.txt'])
|
|
fg = FileGroup.from_paths(test_session, txt)
|
|
test_session.commit()
|
|
tag = await tag_factory('keep')
|
|
fg.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
fg_id = fg.id
|
|
|
|
(test_directory / 'note.csv').write_text('a,b\n')
|
|
await refresh_files()
|
|
test_session.expire_all()
|
|
|
|
remaining = test_session.query(FileGroup).filter_by(id=fg_id).one()
|
|
assert remaining.tag_names == ['keep']
|
|
names = {pathlib.Path(f['path']).name for f in remaining.files}
|
|
assert names == {'note.txt', 'note.csv'}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_into_existing_stem_merges(async_client, test_session, test_directory, make_files_structure):
|
|
"""Renaming onto another FileGroup's stem merges before moving files, so disk and DB agree."""
|
|
make_files_structure({'foo.jpg': 'x', 'bar.mp4': 'video'})
|
|
jpg = test_directory / 'foo.jpg'
|
|
mp4 = test_directory / 'bar.mp4'
|
|
FileGroup.from_paths(test_session, jpg)
|
|
FileGroup.from_paths(test_session, mp4)
|
|
test_session.commit()
|
|
assert test_session.query(FileGroup).count() == 2
|
|
assert {fg.primary_path.name for fg in test_session.query(FileGroup)} == {'foo.jpg', 'bar.mp4'}
|
|
|
|
await lib.rename_file(mp4, 'foo.mp4')
|
|
test_session.expire_all()
|
|
|
|
remaining = test_session.query(FileGroup).one()
|
|
names = {pathlib.Path(f['path']).name for f in remaining.files}
|
|
assert names == {'foo.jpg', 'foo.mp4'}
|
|
assert (test_directory / 'foo.mp4').is_file()
|
|
assert (test_directory / 'foo.jpg').is_file()
|
|
assert not (test_directory / 'bar.mp4').exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dedupe_file_groups_merges_tags_onto_winner(async_client, test_session, test_directory,
|
|
make_files_structure, tag_factory):
|
|
"""Duplicates sharing (directory, stem) collapse onto one row; tags on the loser move."""
|
|
from sqlalchemy import text
|
|
|
|
foo, sidecar = make_files_structure(['foo.mp4', 'foo.info.json'])
|
|
test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx'))
|
|
|
|
keeper = FileGroup.from_paths(test_session, foo)
|
|
keeper.model = 'video'
|
|
loser = FileGroup(
|
|
directory=foo.parent,
|
|
primary_path=sidecar,
|
|
stem='foo',
|
|
files=[{'path': sidecar.name, 'mimetype': 'application/json'}],
|
|
)
|
|
test_session.add(loser)
|
|
test_session.commit()
|
|
|
|
tag = await tag_factory('keepme')
|
|
loser.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
keeper_id, loser_id = keeper.id, loser.id
|
|
assert test_session.query(FileGroup).count() == 2
|
|
|
|
deleted = lib.dedupe_file_groups_by_directory_stem()
|
|
test_session.expire_all()
|
|
|
|
assert deleted == 1
|
|
remaining = test_session.query(FileGroup).one()
|
|
assert remaining.id == keeper_id
|
|
assert remaining.tag_names == ['keepme']
|
|
assert test_session.query(FileGroup).filter_by(id=loser_id).one_or_none() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dedupe_file_groups_prefers_tagged_row(async_client, test_session, test_directory,
|
|
make_files_structure, tag_factory):
|
|
"""The tagged FileGroup wins even when it has the lower id and no model."""
|
|
from sqlalchemy import text
|
|
|
|
a, b = make_files_structure(['bar.mp4', 'bar.png'])
|
|
test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx'))
|
|
first = FileGroup.from_paths(test_session, a)
|
|
second = FileGroup(
|
|
directory=b.parent,
|
|
primary_path=b,
|
|
stem='bar',
|
|
files=[{'path': b.name, 'mimetype': 'image/png'}],
|
|
model='video',
|
|
)
|
|
test_session.add(second)
|
|
test_session.commit()
|
|
assert first.id < second.id
|
|
tag = await tag_factory('tagged-first')
|
|
first.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
|
|
lib.dedupe_file_groups_by_directory_stem()
|
|
test_session.expire_all()
|
|
|
|
remaining = test_session.query(FileGroup).one()
|
|
# has_model ranks above tags, so the modeled later row wins; tags move onto it.
|
|
assert remaining.id == second.id
|
|
assert remaining.tag_names == ['tagged-first']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_group_tag(test_session, make_files_structure, test_directory, tag_factory, await_switches,
|
|
refresh_files):
|
|
"""A FileGroup can be tagged."""
|
|
make_files_structure(['foo.mp4'])
|
|
await refresh_files()
|
|
one = await tag_factory()
|
|
|
|
foo: FileGroup = test_session.query(FileGroup).one()
|
|
foo.add_tag(test_session, one.id)
|
|
test_session.commit()
|
|
|
|
tag_file2: TagFile = test_session.query(TagFile).one()
|
|
assert tag_file2 and tag_file2.file_group == foo
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_a_text_no_indexer(async_client, test_session, make_files_structure, refresh_files):
|
|
"""File.a_text is filled even if the file does not match an Indexer."""
|
|
make_files_structure(['foo', 'bar-bar'])
|
|
|
|
await refresh_files()
|
|
|
|
files = {i.a_text for i in test_session.query(FileGroup)}
|
|
assert files == {'bar bar bar-bar', 'foo'}
|
|
|
|
|
|
@skip_circleci
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_many_files(async_client, test_session, make_files_structure, refresh_files):
|
|
"""Used to profile file refreshing"""
|
|
file_count = 10_000
|
|
make_files_structure([f'{uuid4()}.txt' for _ in range(file_count)])
|
|
with timer('first refresh'):
|
|
await refresh_files()
|
|
assert test_session.query(FileGroup).count() == file_count
|
|
|
|
with timer('second refresh'):
|
|
await refresh_files()
|
|
assert test_session.query(FileGroup).count() == file_count
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mime_type(async_client, test_session, make_files_structure, test_directory, assert_files, refresh_files):
|
|
"""Files module uses the `file` command to get the mimetype of each file."""
|
|
from PIL import Image
|
|
|
|
foo, bar, baz, empty = make_files_structure([
|
|
'dir/foo text.txt',
|
|
'dir/bar.jpeg',
|
|
'dir/baz.mp4',
|
|
'dir/empty',
|
|
])
|
|
foo.write_text('some text')
|
|
Image.new('RGB', (25, 25), color='grey').save(bar)
|
|
shutil.copy(PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4', baz)
|
|
|
|
await refresh_files()
|
|
assert_files([
|
|
{'path': 'dir/foo text.txt'},
|
|
{'path': 'dir/bar.jpeg'},
|
|
{'path': 'dir/baz.mp4'},
|
|
{'path': 'dir/empty'},
|
|
])
|
|
|
|
foo = test_session.query(FileGroup).filter_by(primary_path=f'{test_directory}/dir/foo text.txt').one()
|
|
bar = test_session.query(FileGroup).filter_by(primary_path=f'{test_directory}/dir/bar.jpeg').one()
|
|
baz = test_session.query(FileGroup).filter_by(primary_path=f'{test_directory}/dir/baz.mp4').one()
|
|
empty = test_session.query(FileGroup).filter_by(primary_path=f'{test_directory}/dir/empty').one()
|
|
|
|
assert foo.mimetype == 'text/plain'
|
|
assert bar.mimetype == 'image/jpeg'
|
|
assert baz.mimetype == 'video/mp4'
|
|
assert empty.mimetype == 'inode/x-empty'
|
|
|
|
|
|
@skip_circleci
|
|
@pytest.mark.asyncio
|
|
async def test_files_indexer(async_client, test_session, make_files_structure, test_directory, refresh_files,
|
|
await_background_tasks):
|
|
"""An Indexer is provided for each file based on it's mimetype or contents."""
|
|
source_files: List[str] = [
|
|
'a text file.txt',
|
|
'a zip file.zip',
|
|
'images/an image file.jpeg',
|
|
'unknown file',
|
|
'videos/a video file.info.json', # This is "associated" and will be hidden.
|
|
'videos/a video file.mp4',
|
|
]
|
|
text_path, zip_path, image_path, unknown_path, info_json_path, video_path \
|
|
= make_files_structure(source_files)
|
|
text_path.write_text('text file contents')
|
|
Image.new('RGB', (25, 25), color='grey').save(image_path)
|
|
shutil.copy(PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4', video_path)
|
|
with zipfile.ZipFile(zip_path, 'w') as zip_file:
|
|
zip_file.write(PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4')
|
|
info_json_path.write_text(json.dumps({'description': 'the video description'}))
|
|
|
|
# Enable slow feature for testing.
|
|
# TODO can this be sped up to always be included?
|
|
with mock.patch('modules.videos.lib.EXTRACT_SUBTITLES', True):
|
|
await refresh_files()
|
|
|
|
text_file, zip_file, image_file, unknown_file, video_file \
|
|
= test_session.query(FileGroup).order_by(FileGroup.primary_path)
|
|
|
|
# Indexers are detected correctly.
|
|
assert text_file.mimetype == 'text/plain' and text_file.indexer == indexers.TextIndexer
|
|
assert zip_file.mimetype == 'application/zip' and zip_file.indexer == indexers.ZipIndexer
|
|
assert image_file.mimetype == 'image/jpeg' and image_file.indexer == indexers.DefaultIndexer
|
|
assert unknown_file.mimetype == 'inode/x-empty' and unknown_file.indexer == indexers.DefaultIndexer
|
|
# Video are indexed by the modeler, not by an indexer.
|
|
assert video_file.mimetype == 'video/mp4' and video_file.indexer == indexers.DefaultIndexer
|
|
|
|
# File are indexed by their titles and contents. Contents (d_text) require a deep search.
|
|
files, total = lib.search_files(test_session, 'file', 10, 0)
|
|
assert total == 5, 'All files contain "file" in their file name. The associated video file is hidden.'
|
|
files, total = lib.search_files(test_session, 'image', 10, 0)
|
|
assert total == 1 and files[0]['title'] == 'an image file.jpeg', 'The image file title contains "image".'
|
|
files, total = lib.search_files(test_session, 'contents', 10, 0, deep=True)
|
|
assert total == 1 and files[0]['title'] == 'a text file.txt', 'The text file contains "contents".'
|
|
files, total = lib.search_files(test_session, 'contents', 10, 0)
|
|
assert total == 0, 'File contents are not searched unless deep=True.'
|
|
files, total = lib.search_files(test_session, 'video', 10, 0)
|
|
assert total == 1 and {i['title'] for i in files} == {'a video file'}, 'The video file contains "video".'
|
|
files, total = lib.search_files(test_session, 'yawn', 10, 0, deep=True)
|
|
assert total == 1 and files[0]['title'] == 'a video file', 'The video file captions contain "yawn".'
|
|
files, total = lib.search_files(test_session, 'bunny', 10, 0, deep=True)
|
|
assert total == 1 and {i['title'] for i in files} == {'a zip file.zip'}, \
|
|
'The zip file contains a file with "bunny" in the title.'
|
|
|
|
with mock.patch('modules.videos.models.Video.validate') as mock_validate:
|
|
mock_validate.side_effect = Exception('This should not be called twice')
|
|
await refresh_files()
|
|
|
|
# Change the contents, the file should be re-indexed.
|
|
text_path.write_text('new text contents')
|
|
await refresh_files()
|
|
await await_background_tasks()
|
|
files, total = lib.search_files(test_session, 'new', 10, 0, deep=True)
|
|
assert total == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_markdown_indexed(test_session, make_files_structure, refresh_files):
|
|
"""Markdown files are text/markdown FileGroups whose contents are deep-searchable."""
|
|
make_files_structure({'notes.md': '# Water storage\n\nRotate the barrels every sixmonths.\n'})
|
|
await refresh_files()
|
|
|
|
markdown, = test_session.query(FileGroup).all()
|
|
assert markdown.mimetype == 'text/markdown'
|
|
assert markdown.indexer == indexers.TextIndexer
|
|
files, total = lib.search_files(test_session, 'sixmonths', 10, 0, deep=True)
|
|
assert total == 1 and files[0]['id'] == markdown.id
|
|
|
|
|
|
@pytest.mark.parametrize('name,expected', [
|
|
('this.txt', 'this txt'),
|
|
('name', 'name'),
|
|
('name two', 'name two'),
|
|
('this self-reliance_split.txt', 'this self reliance self-reliance split txt'),
|
|
('-be_split!.txt', '-be split! txt'),
|
|
('WROLPi-v0.10-aarch64-desktop.img.xz', 'WROLPi v0.10 aarch64 desktop.img WROLPi-v0.10-aarch64-desktop.img xz'),
|
|
('some words [but words in braces].txt', 'some words but words in braces txt'),
|
|
('some words (but words in parentheses).txt', 'some words but words in parentheses txt'),
|
|
])
|
|
def test_split_file_name_words(name, expected):
|
|
assert lib.split_file_name_words(name) == expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_large_text_indexer(async_client, test_session, make_files_structure, refresh_files):
|
|
"""
|
|
Large files have their indexes truncated.
|
|
"""
|
|
large, = make_files_structure({
|
|
'large_file.txt': 'foo ' * 1_000_000,
|
|
})
|
|
await refresh_files()
|
|
assert test_session.query(FileGroup).count() == 1
|
|
|
|
assert large.is_file() and large.stat().st_size == 4_000_000
|
|
|
|
large_file: FileGroup = test_session.query(FileGroup).one()
|
|
assert len(large_file.d_text) < large.stat().st_size
|
|
assert len(large_file.d_text) == 90_072
|
|
|
|
|
|
def test_glob_shared_stem(make_files_structure):
|
|
mp4, png, j, name, video, something, vid2, vid2j = make_files_structure([
|
|
'video.mp4',
|
|
'video.png',
|
|
'video.info.json',
|
|
'video-name.txt',
|
|
'video/',
|
|
'something',
|
|
'videos/video2 [name].mp4',
|
|
'videos/video2 [name].info.json',
|
|
])
|
|
|
|
def check(path, expected):
|
|
assert sorted([i.name for i in lib.glob_shared_stem(path)]) == sorted(expected)
|
|
|
|
check(mp4, ['video.mp4', 'video.png', 'video.info.json'])
|
|
check(png, ['video.mp4', 'video.png', 'video.info.json'])
|
|
check(j, ['video.mp4', 'video.png', 'video.info.json'])
|
|
check(video, ['video.mp4', 'video.png', 'video.info.json', 'video'])
|
|
|
|
check(something, ['something'])
|
|
|
|
check(vid2, ['video2 [name].mp4', 'video2 [name].info.json'])
|
|
check(vid2j, ['video2 [name].mp4', 'video2 [name].info.json'])
|
|
|
|
|
|
def test_matching_directories(make_files_structure, test_directory):
|
|
make_files_structure([
|
|
'foo/qux/',
|
|
'Bar/',
|
|
'baz/baz'
|
|
'barr',
|
|
'bazz',
|
|
])
|
|
|
|
# No directories have c
|
|
matches = lib.get_matching_directories(test_directory / 'c')
|
|
assert matches == []
|
|
|
|
# Get all directories starting with f
|
|
matches = lib.get_matching_directories(test_directory / 'f')
|
|
assert matches == [test_directory / 'foo']
|
|
|
|
# Get all directories starting with b, ignore case
|
|
matches = lib.get_matching_directories(test_directory / 'b')
|
|
assert matches == [test_directory / 'Bar', test_directory / 'baz']
|
|
|
|
# baz matches, but it has no subdirectories
|
|
matches = lib.get_matching_directories(test_directory / 'baz')
|
|
assert matches == [test_directory / 'baz']
|
|
|
|
# foo is an exact match, return subdirectories
|
|
matches = lib.get_matching_directories(test_directory / 'foo')
|
|
assert matches == [test_directory / 'foo/qux']
|
|
|
|
|
|
def test_get_mimetype(example_epub, example_mobi, example_pdf, image_file, video_file):
|
|
assert lib.get_mimetype(example_epub) == 'application/epub+zip'
|
|
assert lib.get_mimetype(example_mobi) == 'application/x-mobipocket-ebook'
|
|
assert lib.get_mimetype(example_pdf) == 'application/pdf'
|
|
assert lib.get_mimetype(image_file) == 'image/jpeg'
|
|
assert lib.get_mimetype(video_file) == 'video/mp4'
|
|
|
|
|
|
def test_get_mimetype_stl(test_directory):
|
|
""".stl files should resolve to model/stl regardless of magic's initial guess."""
|
|
stl_path = test_directory / 'example.stl'
|
|
# Minimal ASCII STL body; python-magic reports this as text/plain.
|
|
stl_path.write_text(
|
|
'solid cube\n'
|
|
'facet normal 0 0 1\n'
|
|
' outer loop\n'
|
|
' vertex 0 0 0\n'
|
|
' vertex 1 0 0\n'
|
|
' vertex 0 1 0\n'
|
|
' endloop\n'
|
|
'endfacet\n'
|
|
'endsolid cube\n'
|
|
)
|
|
lib.get_mimetype.cache_clear()
|
|
assert lib.get_mimetype(stl_path) == 'model/stl'
|
|
|
|
|
|
@pytest.mark.parametrize('name,contents', [
|
|
('notes.md', '# Title\n\nSome *text* and a [link](https://example.com).\n\n- a\n- b\n'),
|
|
('plain.md', 'Only a paragraph.\nNo markdown syntax at all.\n'),
|
|
# libmagic sniffs fenced code and may report a programming language instead of text/plain.
|
|
('code.markdown', '#include <stdio.h>\n\nint main(void) {\n return 0;\n}\n'),
|
|
('UPPER.MD', '# Upper case suffix\n'),
|
|
])
|
|
def test_get_mimetype_markdown(test_directory, name, contents):
|
|
"""Markdown files resolve to text/markdown however libmagic sniffs their text."""
|
|
path = test_directory / name
|
|
path.write_text(contents)
|
|
lib.get_mimetype.cache_clear()
|
|
assert lib.get_mimetype(path) == 'text/markdown'
|
|
|
|
|
|
def test_get_mimetype_3mf(test_directory):
|
|
""".3mf files are zipped XML; ensure they resolve to model/3mf."""
|
|
import zipfile
|
|
threemf_path = test_directory / 'example.3mf'
|
|
with zipfile.ZipFile(threemf_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr('[Content_Types].xml',
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
|
|
'<Default Extension="model" ContentType="application/vnd.ms-package.3dmanufacturing-3dmodel+xml"/>'
|
|
'</Types>')
|
|
zf.writestr('3D/3dmodel.model',
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
'<model unit="millimeter" xml:lang="en-US"'
|
|
' xmlns="http://schemas.microsoft.com/3dmanufacturing/core/2015/02">'
|
|
'<resources/><build/></model>')
|
|
lib.get_mimetype.cache_clear()
|
|
assert lib.get_mimetype(threemf_path) == 'model/3mf'
|
|
|
|
|
|
def test_group_files_by_stem(make_files_structure, test_directory):
|
|
make_files_structure([
|
|
'foo.mp4',
|
|
'foo.txt',
|
|
'foo.info.json',
|
|
'foo.live_chat.json',
|
|
'bar.txt',
|
|
'baz.txt',
|
|
])
|
|
|
|
files = list(test_directory.iterdir())
|
|
assert list(lib.group_files_by_stem(files)) == [
|
|
[test_directory / 'bar.txt'],
|
|
[test_directory / 'baz.txt'],
|
|
[test_directory / 'foo.info.json', test_directory / 'foo.live_chat.json', test_directory / 'foo.mp4',
|
|
test_directory / 'foo.txt'],
|
|
]
|
|
|
|
make_files_structure(['dir_a/s3.bin', 'dir_b/s3.bin'])
|
|
same_stem = [test_directory / 'dir_a/s3.bin', test_directory / 'dir_b/s3.bin']
|
|
assert list(lib.group_files_by_stem(same_stem)) == [
|
|
[test_directory / 'dir_a/s3.bin'],
|
|
[test_directory / 'dir_b/s3.bin'],
|
|
]
|
|
|
|
|
|
def test_get_primary_file(test_directory, video_file, srt_file3, example_epub, example_mobi, example_pdf,
|
|
singlefile_contents_factory, make_files_structure):
|
|
"""Test that the most important file is returned from a list of files."""
|
|
srt_file3 = srt_file3.rename(test_directory / f'{video_file.stem}.srt')
|
|
# Same primary_file no matter the order.
|
|
assert lib.get_primary_file([video_file, srt_file3]) == video_file
|
|
assert lib.get_primary_file([srt_file3, video_file]) == video_file
|
|
|
|
singlefile = test_directory / 'singlefile.html'
|
|
singlefile.write_text(singlefile_contents_factory())
|
|
singlefile_text = test_directory / 'singlefile.txt'
|
|
singlefile_text.write_text('the contents')
|
|
|
|
assert lib.get_primary_file([singlefile, singlefile_text]) == singlefile
|
|
|
|
assert lib.get_primary_file([example_epub, example_mobi]) == example_epub
|
|
assert lib.get_primary_file([example_mobi, example_epub]) == example_epub
|
|
|
|
assert lib.get_primary_file([example_pdf, example_mobi]) == example_pdf
|
|
assert lib.get_primary_file([example_pdf, example_mobi, example_epub]) == example_epub
|
|
|
|
# No primary file in this group.
|
|
foo, bar = make_files_structure({'foo': 'text', 'bar': None})
|
|
with pytest.raises(NoPrimaryFile):
|
|
assert lib.get_primary_file([foo, bar])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_files_no_groups(async_client, test_session, test_directory, make_files_structure,
|
|
zip_file_factory, refresh_files):
|
|
"""Files that share a stem are one FileGroup even when neither is a modeled primary."""
|
|
foo_txt, foo_zip = make_files_structure({
|
|
'foo.txt': 'text',
|
|
'foo.zip': zip_file_factory(),
|
|
})
|
|
assert foo_txt.stat().st_size and foo_zip.stat().st_size
|
|
|
|
await refresh_files()
|
|
|
|
# Same stem is one FileGroup even when neither file is a modeled primary.
|
|
assert test_session.query(FileGroup).count() == 1
|
|
fg = test_session.query(FileGroup).one()
|
|
names = {pathlib.Path(f['path']).name for f in fg.files}
|
|
assert names == {'foo.txt', 'foo.zip'}
|
|
assert fg.size == foo_txt.stat().st_size + foo_zip.stat().st_size
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_directories(test_session, test_directory, assert_directories, await_switches, refresh_files):
|
|
"""
|
|
Directories are stored when they are discovered. They are removed when they can no longer be found.
|
|
"""
|
|
foo = test_directory / 'foo'
|
|
bar = test_directory / 'bar'
|
|
baz = test_directory / 'baz'
|
|
foo.mkdir()
|
|
bar.mkdir()
|
|
baz.mkdir()
|
|
|
|
await refresh_files()
|
|
assert_directories({'foo', 'bar', 'baz'})
|
|
|
|
# Deleted directory is removed.
|
|
foo.rmdir()
|
|
await refresh_files()
|
|
assert_directories({'bar', 'baz'})
|
|
|
|
bar.rmdir()
|
|
await refresh_files([bar])
|
|
assert_directories({'baz', })
|
|
|
|
# A new directory can be refreshed directly.
|
|
foo.mkdir()
|
|
await refresh_files([foo])
|
|
assert_directories({'foo', 'baz'})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_directories_chunks_large_batches(test_session, test_directory):
|
|
"""A media directory with more directories than SQLite can bind in one statement is inserted in chunks.
|
|
|
|
Regression: SQLite limits the number of bound variables per statement, so a single multi-row VALUES
|
|
clause over tens of thousands of directories raised RuntimeError('values_clause: too many parameters')."""
|
|
from wrolpi.db import SQLITE_MAX_VARIABLES
|
|
# upsert_directories binds 3 columns per directory; exceed the limit so the un-chunked path would fail.
|
|
count = SQLITE_MAX_VARIABLES // 3 + 50
|
|
paths = [test_directory / f'd{i:05d}' for i in range(count)]
|
|
for path in paths:
|
|
path.mkdir()
|
|
|
|
lib.upsert_directories([test_directory], paths)
|
|
|
|
stored = {i.path for i in test_session.query(Directory).all()}
|
|
assert set(paths).issubset(stored), 'all discovered directories should be inserted'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_group_merge(async_client, test_session, test_directory, make_files_structure, tag_factory,
|
|
video_bytes, srt_file3):
|
|
"""from_paths on a sidecar finds the existing (directory, stem) FileGroup instead of inserting a duplicate."""
|
|
vid, srt = make_files_structure({
|
|
'vid.mp4': video_bytes,
|
|
'vid.srt': (PROJECT_DIR / 'test/example3.en.srt').read_text(),
|
|
})
|
|
one, two = await tag_factory(), await tag_factory()
|
|
vid_group = FileGroup.from_paths(test_session, vid)
|
|
test_session.flush([vid_group])
|
|
vid_tag_file = vid_group.add_tag(test_session, one.name)
|
|
tag_file_created_at = vid_tag_file.created_at
|
|
test_session.commit()
|
|
|
|
srt_group = FileGroup.from_paths(test_session, srt)
|
|
assert srt_group.id == vid_group.id
|
|
srt_tag_file = srt_group.add_tag(test_session, two.name)
|
|
srt_file_created_at = srt_tag_file.created_at
|
|
test_session.commit()
|
|
|
|
merged = FileGroup.from_paths(test_session, vid, srt)
|
|
test_session.commit()
|
|
assert merged.id == vid_group.id
|
|
assert test_session.query(FileGroup).count() == 1
|
|
assert set(merged.tag_names) == {'one', 'two'}
|
|
assert {i['path'].name for i in merged.my_files()} == {'vid.mp4', 'vid.srt'}
|
|
assert [i for i in merged.tag_files if i.tag.name == 'one'][0].created_at == tag_file_created_at
|
|
assert [i for i in merged.tag_files if i.tag.name == 'two'][0].created_at == srt_file_created_at
|
|
assert merged.mimetype == 'video/mp4'
|
|
filenames = [f['path'] for f in merged.files]
|
|
assert len(filenames) == len(set(filenames)), f"Duplicate files found: {filenames}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_directory(async_client, test_session, test_directory, make_files_structure,
|
|
assert_directories, refresh_files):
|
|
"""A Directory record is deleted when it's directory is deleted."""
|
|
make_files_structure({
|
|
'foo/one.txt': 'one',
|
|
})
|
|
await refresh_files()
|
|
assert_directories({'foo'})
|
|
|
|
bar = test_directory / 'bar'
|
|
await lib.rename(test_session, test_directory / 'foo', 'bar')
|
|
assert (bar / 'one.txt').read_text() == 'one'
|
|
assert_directories({'bar'})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_group_move(async_client, test_session, make_files_structure, test_directory, video_bytes,
|
|
srt_text, refresh_files):
|
|
"""Test FileGroup's move method"""
|
|
video, srt = make_files_structure({
|
|
'video.mp4': video_bytes,
|
|
'video.srt': srt_text,
|
|
})
|
|
await refresh_files()
|
|
file_group: FileGroup = test_session.query(FileGroup).one()
|
|
assert file_group.indexed is True
|
|
|
|
(test_directory / 'foo').mkdir()
|
|
new_path = test_directory / 'foo/video.mp4'
|
|
file_group.move(new_path)
|
|
|
|
new_srt = new_path.with_suffix('.srt')
|
|
assert new_path.read_bytes() == video_bytes
|
|
assert new_srt.read_text() == srt_text
|
|
assert not video.is_file()
|
|
assert not srt.is_file()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_tagged(async_client, test_session, test_directory, make_files_structure, tag_factory, refresh_files,
|
|
await_background_tasks):
|
|
"""A FileGroup's tag is preserved when moved or renamed."""
|
|
from wrolpi.files.worker import file_worker
|
|
|
|
tag = await tag_factory()
|
|
foo, bar, = make_files_structure({
|
|
'foo/foo.txt': 'foo',
|
|
'foo/bar.txt': 'bar',
|
|
})
|
|
await refresh_files()
|
|
bar_file_group, foo_file_group = test_session.query(FileGroup).order_by(FileGroup.primary_path)
|
|
bar_file_group.title = 'custom title' # Should not be overwritten.
|
|
bar_file_group.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
|
|
qux = test_directory / 'qux'
|
|
qux.mkdir()
|
|
|
|
# Move both files into qux. The Tag should also be moved.
|
|
file_worker.queue_move(qux, [bar, foo])
|
|
await await_background_tasks()
|
|
test_session.expire_all()
|
|
|
|
new_foo = qux / 'foo.txt'
|
|
new_bar = qux / 'bar.txt'
|
|
# Files were moved.
|
|
assert new_foo.read_text() == 'foo'
|
|
assert new_bar.read_text() == 'bar'
|
|
assert (test_directory / 'foo').is_dir()
|
|
assert not (test_directory / 'foo/foo.txt').exists()
|
|
assert not (test_directory / 'foo/bar.txt').exists()
|
|
# Tag was moved.
|
|
bar_file_group, foo_file_group = test_session.query(FileGroup).order_by(FileGroup.primary_path)
|
|
assert bar_file_group.primary_path == new_bar
|
|
# files now stores relative filenames; use my_files() to get resolved absolute paths
|
|
assert bar_file_group.my_files()[0]['path'] == new_bar
|
|
assert bar_file_group.primary_path.is_file()
|
|
assert bar_file_group.tag_files
|
|
# foo.txt was not tagged, but was moved.
|
|
assert not foo_file_group.tag_files
|
|
assert foo_file_group.primary_path.is_file()
|
|
assert foo_file_group.primary_path == new_foo
|
|
assert foo_file_group.my_files()[0]['path'] == new_foo
|
|
assert bar_file_group.title == 'custom title', 'Custom title should not have been overwritten.'
|
|
|
|
# Rename "bar.txt" to "baz.txt"
|
|
await lib.rename(test_session, new_bar, 'baz.txt')
|
|
baz = new_bar.with_name('baz.txt')
|
|
assert baz.read_text() == 'bar'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_html_index(async_client, test_session, test_directory, make_files_structure, refresh_files):
|
|
make_files_structure({
|
|
'archive.html': '''<html>
|
|
<title>The Title</title>
|
|
|
|
<style>
|
|
body {
|
|
color: red;
|
|
}
|
|
</style>
|
|
|
|
<script class="sf-hidden" type="application/ld+json">
|
|
{"@context":"http://schema.org", "@type":"NewsArticle",
|
|
"datePublished":"2022-09-27T00:40:19.000Z", "dateModified":"2022-09-27T13:43:47.971Z",
|
|
"author":{"@type":"Person", "name":"BOBBY", "jobTitle":""},
|
|
"creator":{"@type":"Person", "name":"OTHER BOBBY", "jobTitle":""},
|
|
"description": "The article description"}
|
|
</script>
|
|
|
|
<body>
|
|
<h1>Some Header</h1>
|
|
<p>Some Text</p>
|
|
|
|
<span>Span Text</span>
|
|
|
|
Outside element.
|
|
</body>
|
|
|
|
</html>
|
|
'''
|
|
})
|
|
|
|
await refresh_files()
|
|
|
|
assert test_session.query(FileGroup).count() == 1, 'Too many files were refreshed.'
|
|
archive_file: FileGroup = test_session.query(FileGroup).one()
|
|
assert archive_file.a_text == 'The Title'
|
|
assert archive_file.b_text == 'archive html'
|
|
assert archive_file.c_text == 'The article description'
|
|
assert archive_file.d_text == '''Some Header
|
|
Some Text
|
|
Span Text
|
|
Outside element.'''
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_doc_indexer(async_client, test_session, example_doc, refresh_files):
|
|
"""The contents of a Microsoft doc files can be indexed."""
|
|
await refresh_files()
|
|
assert test_session.query(FileGroup).count() == 1
|
|
doc = test_session.query(FileGroup).one()
|
|
assert doc.title == 'example word doc'
|
|
assert doc.a_text == 'example word doc'
|
|
if IS_MACOS:
|
|
assert doc.d_text == 'Example Word Document\nSecond page'
|
|
else:
|
|
assert doc.d_text == 'Example Word Document\n\nSecond page\n'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_docx_indexer(async_client, test_session, example_docx, refresh_files):
|
|
"""The contents of a Microsoft docx files can be indexed."""
|
|
await refresh_files()
|
|
assert test_session.query(FileGroup).count() == 1
|
|
doc = test_session.query(FileGroup).one()
|
|
# The docx file has a "Word Document" title in its core properties.
|
|
assert doc.title == 'Word Document'
|
|
assert doc.a_text == 'Word Document'
|
|
assert doc.d_text == 'Example Word Document Second page'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_search_date_range(async_client, test_session, example_pdf, example_doc, refresh_files):
|
|
"""Test searching FileGroups by their published datetime."""
|
|
await refresh_files()
|
|
doc, pdf = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
|
|
|
|
files, total = lib.search_files(test_session, '', 10, 0)
|
|
assert total == 2, 'Should only be 2 files.'
|
|
|
|
# Past the last page still reports the true total (count is a separate query).
|
|
files, total = lib.search_files(test_session, '', 10, 10)
|
|
assert files == []
|
|
assert total == 2
|
|
|
|
# PDF was published in December.
|
|
files, total = lib.search_files(test_session, '', 10, 0, months=[12, ])
|
|
assert total == 1, 'Only PDF should be from December 2022'
|
|
assert files[0]['id'] == pdf.id
|
|
|
|
# PDF was published in 2022.
|
|
files, total = lib.search_files(test_session, '', 10, 0, from_year=2022, to_year=2022)
|
|
assert total == 1, 'Only PDF should be from December 2022'
|
|
assert files[0]['id'] == pdf.id
|
|
|
|
# PDF was NOT published in 2023.
|
|
files, total = lib.search_files(test_session, '', 10, 0, from_year=2023)
|
|
assert total == 0, 'No example files are published in 2023'
|
|
|
|
files, total = lib.search_files(test_session, '', 10, 0, to_year=2022)
|
|
assert total == 1
|
|
assert files[0]['id'] == pdf.id
|
|
|
|
|
|
def test_search_filter_cache_key():
|
|
"""Totals are cached by filters, not by page or sort direction."""
|
|
a = lib.search_filter_cache_key('videos', deep=False, tag_names=[], search_str='')
|
|
b = lib.search_filter_cache_key('videos')
|
|
assert a == b
|
|
assert lib.search_filter_cache_key('videos', tag_names=['b', 'a']) == \
|
|
lib.search_filter_cache_key('videos', tag_names=['a', 'b'])
|
|
|
|
|
|
def test_replace_file(test_directory):
|
|
file = test_directory / 'foo.txt'
|
|
file.write_text('old contents')
|
|
assert file.read_text() == 'old contents'
|
|
|
|
wrolpi.common.replace_file(file, 'new contents')
|
|
assert file.read_text() == 'new contents'
|
|
assert not (test_directory / 'foo.txt.tmp').exists()
|
|
|
|
wrolpi.common.replace_file(file, b'new bytes')
|
|
assert file.read_bytes() == b'new bytes'
|
|
assert not (test_directory / 'foo.txt.tmp').exists()
|
|
|
|
# Default behavior is to refuse to replace a non-existent file.
|
|
with pytest.raises(FileNotFoundError):
|
|
wrolpi.common.replace_file('does not exist', 'foo')
|
|
|
|
# Non-existent file can be created.
|
|
wrolpi.common.replace_file(test_directory / 'will now exist', 'foo', missing_ok=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_file_video_with_channel(async_client, test_session, test_directory, video_bytes,
|
|
channel_factory, video_factory):
|
|
"""A video file that is uploaded should be assigned to a Channel if it is in the Channel's directory."""
|
|
channel = channel_factory()
|
|
# Put the video file in the Channel's directory.
|
|
video_file: pathlib.Path = channel.directory / 'video.mp4'
|
|
video_file.write_bytes(video_bytes)
|
|
# Put an info json file next to the video file.
|
|
info_json_file = video_file.with_suffix('.info.json')
|
|
info_json_file.write_text(json.dumps({'duration': 5}))
|
|
|
|
# Upsert the file, it should be modeled.
|
|
fg: FileGroup = await lib.upsert_file(video_file)
|
|
assert fg.model == 'video', 'Upserted file should have been modeled.'
|
|
assert fg.my_video_files(), 'Video file was upserted.'
|
|
assert fg.my_json_files(), 'Info json file should have been found near video file.'
|
|
|
|
# The video is in the correct Channel and has its files.
|
|
video: Video = test_session.query(Video).one()
|
|
assert video.channel_id == channel.id, "Videos should be assigned to the Channel their files are in"
|
|
assert video.video_path, 'Video file was upserted'
|
|
assert video.info_json_path, 'Info json file not found near video file.'
|
|
|
|
|
|
@only_macos
|
|
def test_get_real_path_name(test_directory):
|
|
(test_directory / 'foo.txt').touch()
|
|
assert (test_directory / 'foo.txt').exists()
|
|
assert lib.get_real_path_name(test_directory / 'foo.txt') == test_directory / 'foo.txt'
|
|
|
|
assert (test_directory / 'FOO.TXT').exists()
|
|
assert lib.get_real_path_name(test_directory / 'FOO.txt') == test_directory / 'foo.txt'
|
|
|
|
|
|
# --- Bulk Tagging Tests ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_bulk_tag_preview_files(async_client, test_session, make_files_structure, tag_factory):
|
|
"""get_bulk_tag_preview returns correct file count and shared tags for files."""
|
|
foo, bar, baz = make_files_structure({
|
|
'foo.txt': 'foo',
|
|
'bar.txt': 'bar',
|
|
'baz.txt': 'baz',
|
|
})
|
|
|
|
# Create FileGroups
|
|
fg_foo = FileGroup.from_paths(test_session, foo)
|
|
fg_bar = FileGroup.from_paths(test_session, bar)
|
|
fg_baz = FileGroup.from_paths(test_session, baz)
|
|
test_session.commit()
|
|
|
|
# Add shared and non-shared tags
|
|
tag1 = await tag_factory('shared')
|
|
tag2 = await tag_factory('only_foo')
|
|
fg_foo.add_tag(test_session, tag1.id)
|
|
fg_foo.add_tag(test_session, tag2.id)
|
|
fg_bar.add_tag(test_session, tag1.id)
|
|
fg_baz.add_tag(test_session, tag1.id)
|
|
test_session.commit()
|
|
|
|
# Preview for all three files
|
|
preview = lib.get_bulk_tag_preview(test_session, ['foo.txt', 'bar.txt', 'baz.txt'])
|
|
assert preview.file_count == 3
|
|
assert 'shared' in preview.shared_tag_names
|
|
assert 'only_foo' not in preview.shared_tag_names # Not shared by all
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_bulk_tag_preview_directory(test_session, make_files_structure, test_directory, tag_factory):
|
|
"""get_bulk_tag_preview recursively finds files in directories."""
|
|
make_files_structure({
|
|
'mydir/foo.txt': 'foo',
|
|
'mydir/bar.txt': 'bar',
|
|
'mydir/subdir/baz.txt': 'baz',
|
|
})
|
|
|
|
# Preview for the directory - should find all files recursively
|
|
preview = lib.get_bulk_tag_preview(test_session, ['mydir/'])
|
|
assert preview.file_count == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_indexers_concurrent_writer(test_session, make_files_structure, async_client):
|
|
"""apply_indexers must survive other connections committing writes while a batch is indexing.
|
|
|
|
Regression test for the 10.0.0.9 refresh failure: the batch held one ORM session across
|
|
minutes of indexer subprocesses, so its commit had to upgrade a stale read snapshot to a
|
|
write, and SQLite returned "database is locked" (SQLITE_BUSY_SNAPSHOT — busy_timeout does
|
|
not apply) whenever any other connection had committed a write in the meantime."""
|
|
from wrolpi.conftest import production_like_sessions
|
|
|
|
paths = make_files_structure({f'file{i}.txt': f'contents {i}' for i in range(3)})
|
|
for path in paths:
|
|
FileGroup.from_paths(test_session, path)
|
|
test_session.commit()
|
|
|
|
original_create_index = indexers.TextIndexer.create_index
|
|
|
|
with production_like_sessions(test_session) as maker:
|
|
def create_index_and_concurrent_write(path):
|
|
# Another connection commits a write mid-batch, like a modeler or download worker.
|
|
other = maker()
|
|
try:
|
|
other.execute("INSERT INTO directory (path, name, idempotency) "
|
|
"VALUES ('/tmp/concurrent', 'concurrent', CURRENT_TIMESTAMP) "
|
|
"ON CONFLICT (path) DO UPDATE SET name = 'concurrent'")
|
|
other.commit()
|
|
finally:
|
|
other.close()
|
|
return original_create_index(path)
|
|
|
|
with mock.patch.object(indexers.TextIndexer, 'create_index', create_index_and_concurrent_write):
|
|
await lib.apply_indexers()
|
|
|
|
test_session.expire_all()
|
|
file_groups = test_session.query(FileGroup).all()
|
|
assert len(file_groups) == 3
|
|
assert all(fg.indexed for fg in file_groups)
|
|
assert all(fg.d_text for fg in file_groups)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_bulk_tag_preview_many_files(test_session, make_files_structure, test_directory):
|
|
"""Previewing a directory containing well over 1000 files must not fail.
|
|
|
|
Regression test: the stem-prefix lookup OR'd a LIKE per stem, and more than ~1000
|
|
stems exceeded SQLite's expression-tree depth limit ("Expression tree is too large")."""
|
|
make_files_structure({f'mydir/file{i}.txt': str(i) for i in range(1200)})
|
|
|
|
preview = lib.get_bulk_tag_preview(test_session, ['mydir/'])
|
|
assert preview.file_count == 1200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_bulk_tag_preview_multi_file_filegroup(async_client, test_session, make_files_structure,
|
|
test_directory, tag_factory):
|
|
"""get_bulk_tag_preview finds shared tags for multi-file FileGroups.
|
|
|
|
This tests a bug where `get_unique_files_by_stem` may return a non-primary file (e.g. .readability.json)
|
|
but the FileGroup query looks for exact primary_path match (e.g. .html), causing no FileGroup to be found.
|
|
"""
|
|
# Create a multi-file FileGroup like an Archive (primary is .html, but has .readability.json, etc.)
|
|
# SingleFile archives are named with the pattern: %Y-%m-%d-%H-%M-%S_title.html
|
|
html_file, json_file, txt_file = make_files_structure({
|
|
'archive/2025-01-01-12-00-00_Page Title.html': '<html>content</html>',
|
|
'archive/2025-01-01-12-00-00_Page Title.readability.json': '{"title": "Page"}',
|
|
'archive/2025-01-01-12-00-00_Page Title.readability.txt': 'Page content',
|
|
})
|
|
|
|
# Create the FileGroup with .html as the primary path
|
|
fg = FileGroup.from_paths(test_session, html_file, json_file, txt_file)
|
|
test_session.commit()
|
|
|
|
# Verify the primary_path is the .html file
|
|
assert fg.primary_path.suffix == '.html', f'Expected primary_path to be .html, got {fg.primary_path}'
|
|
|
|
# Add a tag to this FileGroup
|
|
tag = await tag_factory('my_tag')
|
|
fg.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
|
|
# Preview for the directory should find the FileGroup and its tag
|
|
preview = lib.get_bulk_tag_preview(test_session, ['archive/'])
|
|
assert preview.file_count == 1, f'Expected 1 FileGroup, got {preview.file_count}'
|
|
assert 'my_tag' in preview.shared_tag_names, f'Expected my_tag in shared tags, got {preview.shared_tag_names}'
|
|
|
|
|
|
def test_get_bulk_tag_preview_empty(test_session, test_directory):
|
|
"""get_bulk_tag_preview returns 0 for empty or non-existent paths."""
|
|
preview = lib.get_bulk_tag_preview(test_session, [])
|
|
assert preview.file_count == 0
|
|
assert preview.shared_tag_names == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('test_case,expected_fg_count', [
|
|
('existing_filegroups', 2),
|
|
('creates_filegroups', 2),
|
|
('recursive_directory', 2),
|
|
('multi_file_filegroup', 1),
|
|
])
|
|
async def test_process_bulk_tag_job_add_tags(test_case, expected_fg_count, async_client, test_session,
|
|
make_files_structure, tag_factory, video_bytes, srt_text,
|
|
test_wrolpi_config):
|
|
"""_process_bulk_tag_job adds tags to files in various scenarios."""
|
|
# Disable ffprobe json file creation to avoid unexpected files in test assertions
|
|
get_wrolpi_config().save_ffprobe_json = False
|
|
|
|
if test_case == 'existing_filegroups':
|
|
foo, bar = make_files_structure({'foo.txt': 'foo', 'bar.txt': 'bar'})
|
|
FileGroup.from_paths(test_session, foo)
|
|
FileGroup.from_paths(test_session, bar)
|
|
test_session.commit()
|
|
job_paths = ['foo.txt', 'bar.txt']
|
|
elif test_case == 'creates_filegroups':
|
|
make_files_structure({'foo.txt': 'foo', 'bar.txt': 'bar'})
|
|
job_paths = ['foo.txt', 'bar.txt']
|
|
elif test_case == 'recursive_directory':
|
|
make_files_structure({'mydir/foo.txt': 'foo', 'mydir/subdir/bar.txt': 'bar'})
|
|
job_paths = ['mydir/']
|
|
elif test_case == 'multi_file_filegroup':
|
|
make_files_structure({
|
|
'video.mp4': video_bytes,
|
|
'video.srt': srt_text,
|
|
'video.info.json': '{"title": "Test Video"}',
|
|
})
|
|
job_paths = ['video.mp4']
|
|
|
|
tag = await tag_factory('new_tag')
|
|
test_session.commit()
|
|
|
|
job = {'paths': job_paths, 'add_tag_names': ['new_tag'], 'remove_tag_names': []}
|
|
await lib._process_bulk_tag_job(job)
|
|
|
|
test_session.expire_all()
|
|
fgs = test_session.query(FileGroup).all()
|
|
assert len(fgs) == expected_fg_count
|
|
for fg in fgs:
|
|
assert 'new_tag' in fg.tag_names
|
|
|
|
# Additional assertion for multi-file FileGroup
|
|
if test_case == 'multi_file_filegroup':
|
|
# files now stores relative filenames as strings; use my_files() to get resolved Paths
|
|
file_paths = {f['path'].name for f in fgs[0].my_files()}
|
|
assert file_paths == {'video.mp4', 'video.srt', 'video.info.json'}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_bulk_tag_job_remove_tags(async_client, test_session, make_files_structure, tag_factory):
|
|
"""_process_bulk_tag_job removes tags from files."""
|
|
foo, bar = make_files_structure({
|
|
'foo.txt': 'foo',
|
|
'bar.txt': 'bar',
|
|
})
|
|
|
|
# Create FileGroups with a tag
|
|
fg_foo = FileGroup.from_paths(test_session, foo)
|
|
fg_bar = FileGroup.from_paths(test_session, bar)
|
|
test_session.commit()
|
|
|
|
tag = await tag_factory('to_remove')
|
|
fg_foo.add_tag(test_session, tag.id)
|
|
fg_bar.add_tag(test_session, tag.id)
|
|
test_session.commit()
|
|
|
|
job = {
|
|
'paths': ['foo.txt', 'bar.txt'],
|
|
'add_tag_names': [],
|
|
'remove_tag_names': ['to_remove'],
|
|
}
|
|
|
|
await lib._process_bulk_tag_job(job)
|
|
|
|
# Verify tags were removed
|
|
test_session.expire_all()
|
|
fgs = test_session.query(FileGroup).all()
|
|
for fg in fgs:
|
|
assert 'to_remove' not in fg.tag_names
|
|
|
|
|
|
def test_sanitize_filename_surrogates_valid_path(test_directory, make_files_structure):
|
|
"""sanitize_filename_surrogates() returns the same path for valid UTF-8 filenames."""
|
|
foo, = make_files_structure({'foo.txt': 'content'})
|
|
|
|
result = lib.sanitize_filename_surrogates(foo)
|
|
|
|
assert result == foo
|
|
assert foo.exists()
|
|
|
|
|
|
def test_sanitize_filename_surrogates_with_emoji(test_directory, make_files_structure):
|
|
"""sanitize_filename_surrogates() handles valid emoji filenames correctly."""
|
|
# Valid emoji filename (this should work on most filesystems)
|
|
files = make_files_structure({'test_emoji_🎉.txt': 'content'})
|
|
emoji_file = files[0]
|
|
|
|
result = lib.sanitize_filename_surrogates(emoji_file)
|
|
|
|
assert result == emoji_file
|
|
assert emoji_file.exists()
|
|
|
|
|
|
def test_sanitize_filename_surrogates_nonexistent_path(test_directory):
|
|
"""sanitize_filename_surrogates() returns the path unchanged if file doesn't exist."""
|
|
nonexistent = test_directory / 'does_not_exist.txt'
|
|
|
|
result = lib.sanitize_filename_surrogates(nonexistent)
|
|
|
|
# Path doesn't exist, so it's returned unchanged (no rename attempted)
|
|
assert result == nonexistent
|
|
|
|
|
|
@pytest.mark.skipif(IS_MACOS, reason="macOS filesystem enforces UTF-8, can't create files with invalid surrogates")
|
|
def test_sanitize_filename_surrogates_renames_file(test_directory):
|
|
"""sanitize_filename_surrogates() renames files when it detects invalid UTF-8.
|
|
|
|
This test only runs on Linux where filesystems allow invalid UTF-8 filenames.
|
|
"""
|
|
import os
|
|
|
|
# Create a file with invalid UTF-8 bytes in the filename using raw bytes
|
|
# \xed\xa0\xbc is an invalid UTF-8 sequence (unpaired surrogate)
|
|
bad_filename_bytes = b'bad_\xed\xa0\xbc_file.txt'
|
|
bad_file_path_bytes = os.fsencode(test_directory) + b'/' + bad_filename_bytes
|
|
|
|
# Write content to the file using raw bytes path
|
|
with open(bad_file_path_bytes, 'wb') as f:
|
|
f.write(b'test content')
|
|
|
|
# Get the path as Python sees it (with surrogates)
|
|
bad_path = test_directory / os.fsdecode(bad_filename_bytes)
|
|
assert bad_path.exists()
|
|
|
|
# Verify the path string contains surrogates (can't be encoded to UTF-8)
|
|
with pytest.raises(UnicodeEncodeError):
|
|
str(bad_path).encode('utf-8')
|
|
|
|
# Call the sanitize function
|
|
result = lib.sanitize_filename_surrogates(bad_path)
|
|
|
|
# The file should have been renamed
|
|
assert result != bad_path
|
|
assert result.exists()
|
|
assert not bad_path.exists()
|
|
# Surrogates should be replaced with underscores - the result should be valid UTF-8
|
|
try:
|
|
str(result).encode('utf-8')
|
|
except UnicodeEncodeError:
|
|
pytest.fail("Result path still contains invalid UTF-8 surrogates")
|
|
# Content should be preserved
|
|
assert result.read_text() == 'test content'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_file_with_associated_files_twice(async_client, test_session, test_directory, make_files_structure,
|
|
video_bytes, srt_text, refresh_files):
|
|
"""
|
|
Renaming a FileGroup twice should rename all associated files both times.
|
|
|
|
Regression test for: When renaming "example.mp4" to "example 2.mp4" and then back to "example.mp4",
|
|
the associated file "example.srt" should be renamed along with the primary file both times.
|
|
"""
|
|
# Create a FileGroup with primary file and associated file
|
|
video, srt = make_files_structure({
|
|
'example.mp4': video_bytes,
|
|
'example.srt': srt_text,
|
|
})
|
|
await refresh_files()
|
|
|
|
# Verify initial state
|
|
file_group = test_session.query(FileGroup).one()
|
|
assert file_group.primary_path == video
|
|
# Video validation creates .ffprobe.json, so we have 3 files
|
|
assert len(file_group.files) == 3
|
|
assert video.is_file()
|
|
assert srt.is_file()
|
|
|
|
# First rename: "example.mp4" -> "example 2.mp4"
|
|
new_path_1 = await lib.rename_file(video, 'example 2.mp4')
|
|
test_session.expire_all()
|
|
|
|
# Verify first rename - both files should be renamed
|
|
new_video_1 = test_directory / 'example 2.mp4'
|
|
new_srt_1 = test_directory / 'example 2.srt'
|
|
assert new_path_1 == new_video_1
|
|
assert new_video_1.is_file(), "Video should be renamed to 'example 2.mp4'"
|
|
assert new_srt_1.is_file(), "SRT should be renamed to 'example 2.srt'"
|
|
assert not video.exists(), "Old video should not exist"
|
|
assert not srt.exists(), "Old SRT should not exist"
|
|
|
|
# Verify FileGroup is updated
|
|
file_group = test_session.query(FileGroup).one()
|
|
assert file_group.primary_path == new_video_1
|
|
file_names = [f['path'] for f in file_group.files]
|
|
assert 'example 2.mp4' in file_names
|
|
assert 'example 2.srt' in file_names
|
|
assert len(file_group.files) >= 2
|
|
|
|
# Second rename: "example 2.mp4" -> "example.mp4"
|
|
new_path_2 = await lib.rename_file(new_video_1, 'example.mp4')
|
|
test_session.expire_all()
|
|
|
|
# Verify second rename - BOTH files should be renamed back
|
|
final_video = test_directory / 'example.mp4'
|
|
final_srt = test_directory / 'example.srt'
|
|
assert new_path_2 == final_video
|
|
assert final_video.is_file(), "Video should be renamed back to 'example.mp4'"
|
|
assert final_srt.is_file(), "SRT should be renamed back to 'example.srt'"
|
|
assert not new_video_1.exists(), "'example 2.mp4' should not exist"
|
|
assert not new_srt_1.exists(), "'example 2.srt' should not exist"
|
|
|
|
# Verify FileGroup is updated
|
|
file_group = test_session.query(FileGroup).one()
|
|
assert file_group.primary_path == final_video
|
|
file_names = [f['path'] for f in file_group.files]
|
|
assert 'example.mp4' in file_names
|
|
assert 'example.srt' in file_names
|
|
assert len(file_group.files) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_non_primary_file_renames_filegroup(async_client, test_session, test_directory,
|
|
make_files_structure, video_bytes, srt_text, refresh_files):
|
|
"""
|
|
Renaming a non-primary file (like a subtitle) should rename the entire FileGroup.
|
|
|
|
If a user renames "example.srt" to "renamed.srt", the primary "example.mp4"
|
|
should also be renamed to "renamed.mp4".
|
|
"""
|
|
# Create a FileGroup with primary file and associated file
|
|
video, srt = make_files_structure({
|
|
'example.mp4': video_bytes,
|
|
'example.srt': srt_text,
|
|
})
|
|
await refresh_files()
|
|
|
|
# Verify initial state
|
|
file_group = test_session.query(FileGroup).one()
|
|
assert file_group.primary_path == video
|
|
# Video validation creates .ffprobe.json, so we may have 2 or 3 files
|
|
assert len(file_group.files) == 3
|
|
|
|
# Rename the non-primary file (the SRT)
|
|
new_path = await lib.rename_file(srt, 'renamed.srt')
|
|
test_session.expire_all()
|
|
|
|
# Verify all files in the FileGroup were renamed
|
|
new_video = test_directory / 'renamed.mp4'
|
|
new_srt = test_directory / 'renamed.srt'
|
|
assert new_path == new_srt
|
|
assert new_video.is_file(), "Primary video should be renamed to 'renamed.mp4'"
|
|
assert new_srt.is_file(), "SRT should be renamed to 'renamed.srt'"
|
|
assert not video.exists(), "Old video should not exist"
|
|
assert not srt.exists(), "Old SRT should not exist"
|
|
|
|
# Verify FileGroup is updated correctly
|
|
file_group = test_session.query(FileGroup).one()
|
|
assert file_group.primary_path == new_video
|
|
file_names = [f['path'] for f in file_group.files]
|
|
assert 'renamed.mp4' in file_names
|
|
assert 'renamed.srt' in file_names
|
|
assert len(file_group.files) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_search_url(test_session, make_files_structure, refresh_files):
|
|
"""Test searching FileGroups by their URL."""
|
|
foo, bar, baz = make_files_structure({
|
|
'foo.txt': 'content',
|
|
'bar.txt': 'content',
|
|
'baz.txt': 'content',
|
|
})
|
|
await refresh_files()
|
|
|
|
foo_fg, bar_fg, baz_fg = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
|
|
foo_fg.url = 'https://example.com/foo'
|
|
bar_fg.url = 'https://example.org/bar'
|
|
baz_fg.url = 'https://example.com/baz/page'
|
|
test_session.commit()
|
|
|
|
# Search by full URL
|
|
files, total = lib.search_files(test_session, '', 10, 0, url='https://example.com/foo')
|
|
assert total == 1
|
|
assert files[0]['id'] == foo_fg.id
|
|
|
|
# Search by domain (partial match)
|
|
files, total = lib.search_files(test_session, '', 10, 0, url='example.com')
|
|
assert total == 2
|
|
assert {f['id'] for f in files} == {foo_fg.id, baz_fg.id}
|
|
|
|
# Case insensitive search
|
|
files, total = lib.search_files(test_session, '', 10, 0, url='EXAMPLE.COM')
|
|
assert total == 2
|
|
|
|
# No results for non-matching URL
|
|
files, total = lib.search_files(test_session, '', 10, 0, url='nonexistent.com')
|
|
assert total == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_search_url_with_null(test_session, make_files_structure, refresh_files):
|
|
"""Test that files without URLs are excluded when URL filter is applied."""
|
|
foo, bar = make_files_structure({
|
|
'foo.txt': 'content',
|
|
'bar.txt': 'content',
|
|
})
|
|
await refresh_files()
|
|
|
|
foo_fg, bar_fg = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
|
|
foo_fg.url = 'https://example.com/foo'
|
|
# bar_fg.url remains None
|
|
test_session.commit()
|
|
|
|
files, total = lib.search_files(test_session, '', 10, 0, url='example.com')
|
|
assert total == 1
|
|
assert files[0]['id'] == foo_fg.id
|
|
|
|
|
|
def test_bulk_update_file_groups_reorganize_handles_datetime(test_session, test_directory):
|
|
"""_bulk_update_file_groups_reorganize should handle datetime objects in files JSON.
|
|
|
|
The files JSON field can contain modification_datetime as a datetime object.
|
|
The function must serialize these correctly to avoid JSON serialization errors.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from wrolpi.files.lib import _bulk_update_file_groups_reorganize
|
|
|
|
# Create a test file and FileGroup
|
|
test_file = test_directory / 'test_video.mp4'
|
|
test_file.touch()
|
|
|
|
fg = FileGroup()
|
|
fg.directory = test_directory
|
|
fg.primary_path = test_file
|
|
fg.files = [{'path': 'test_video.mp4', 'mimetype': 'video/mp4'}]
|
|
test_session.add(fg)
|
|
test_session.commit()
|
|
fg_id = fg.id
|
|
|
|
# Prepare update with datetime object in files (simulating what handle_reorganize does)
|
|
new_files = [{
|
|
'path': 'test_video.mp4',
|
|
'mimetype': 'video/mp4',
|
|
'modification_datetime': datetime.now(timezone.utc), # datetime object, not string
|
|
}]
|
|
|
|
updates = [{
|
|
'id': fg_id,
|
|
'directory': str(test_directory),
|
|
'primary_path': str(test_file),
|
|
'files': new_files,
|
|
}]
|
|
|
|
# This should NOT raise "Object of type datetime is not JSON serializable"
|
|
_bulk_update_file_groups_reorganize(updates)
|
|
|
|
# Verify the update was applied
|
|
test_session.expire_all()
|
|
fg = test_session.query(FileGroup).filter_by(id=fg_id).one()
|
|
assert len(fg.files) == 1
|
|
assert fg.files[0]['path'] == 'test_video.mp4'
|
|
# modification_datetime should be serialized as ISO string
|
|
assert 'modification_datetime' in fg.files[0]
|
|
|
|
|
|
def test_bulk_update_file_groups_reorganize_updates_stem(test_session, test_directory):
|
|
"""Reorganize must write the new stem so a later refresh does not delete+reinsert the row."""
|
|
from wrolpi.files.lib import _bulk_update_file_groups_reorganize
|
|
|
|
old_file = test_directory / 'old title.mp4'
|
|
old_file.touch()
|
|
fg = FileGroup.from_paths(test_session, old_file)
|
|
test_session.commit()
|
|
assert fg.stem == 'old title'
|
|
fg_id = fg.id
|
|
|
|
new_path = test_directory / '2024-01-01 New Title.mp4'
|
|
updates = [{
|
|
'id': fg_id,
|
|
'directory': str(test_directory),
|
|
'primary_path': str(new_path),
|
|
'files': [{'path': new_path.name, 'mimetype': 'video/mp4'}],
|
|
'data': None,
|
|
}]
|
|
_bulk_update_file_groups_reorganize(updates)
|
|
|
|
test_session.expire_all()
|
|
fg = test_session.query(FileGroup).filter_by(id=fg_id).one()
|
|
assert fg.stem == '2024-01-01 New Title'
|
|
assert fg.primary_path == new_path
|