mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
aa203cc102
Removing default download freqeuncy so channels will not be downloaded without a url and frequency. Adding channel unique url constraint. Handling when a Channel's directory may still be relative in DB. Adding TODOs to remove old favorites. Removing outdated favorites from config after favorites have been created. Testing that old-style favorited videos in the config are imported. Removing channels_by_link. Using list to store channels in config. Moving channel import function into lib. Supporting old channel import with links. Bumping version for channel.link removal. Removing all reference to Channel.link in app and api. Adding relative property to MediaPathType. Testing that a channel can be downloaded after it has a frequency added. All tests pass. Video and Channel tests pass without channel.link. Removing channel.link usage in channel api. Changing video Statistics page to About.
415 lines
16 KiB
Python
415 lines
16 KiB
Python
import pathlib
|
|
import subprocess
|
|
import tempfile
|
|
from datetime import datetime
|
|
from typing import List
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from modules.videos.models import Channel, Video
|
|
from wrolpi.common import get_absolute_media_path, sanitize_link
|
|
from wrolpi.dates import local_timezone, now
|
|
from wrolpi.downloader import Download, DownloadFrequency
|
|
from wrolpi.test.common import build_test_directories, TestAPI
|
|
from wrolpi.vars import PROJECT_DIR
|
|
from ..common import get_matching_directories, convert_image, remove_duplicate_video_paths, \
|
|
apply_info_json, get_video_duration, generate_video_poster, is_valid_poster
|
|
from ..lib import save_channels_config, get_channels_config, import_channels_config
|
|
|
|
|
|
class TestCommon(TestAPI):
|
|
|
|
def test_matching_directories(self):
|
|
structure = [
|
|
'foo/qux/',
|
|
'Bar/',
|
|
'baz/baz'
|
|
'barr',
|
|
'bazz',
|
|
]
|
|
|
|
with build_test_directories(structure) as temp_dir:
|
|
temp_dir = pathlib.Path(temp_dir)
|
|
|
|
# No directories have c
|
|
matches = get_matching_directories(temp_dir / 'c')
|
|
assert matches == []
|
|
|
|
# Get all directories starting with f
|
|
matches = get_matching_directories(temp_dir / 'f')
|
|
assert matches == [str(temp_dir / 'foo')]
|
|
|
|
# Get all directories starting with b, ignore case
|
|
matches = get_matching_directories(temp_dir / 'b')
|
|
assert matches == [str(temp_dir / 'Bar'), str(temp_dir / 'baz')]
|
|
|
|
# baz matches, but it has no subdirectories
|
|
matches = get_matching_directories(temp_dir / 'baz')
|
|
assert matches == [str(temp_dir / 'baz')]
|
|
|
|
# foo is an exact match, return subdirectories
|
|
matches = get_matching_directories(temp_dir / 'foo')
|
|
assert matches == [str(temp_dir / 'foo/qux')]
|
|
|
|
|
|
def test_get_absolute_media_path():
|
|
path = get_absolute_media_path('videos')
|
|
assert str(path).endswith('videos')
|
|
|
|
|
|
def test_get_video_duration(test_directory):
|
|
"""
|
|
Video duration can be retrieved from the video file.
|
|
"""
|
|
video_path = PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4'
|
|
assert get_video_duration(video_path) == 5
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
get_video_duration(PROJECT_DIR / 'test/does not exist.mp4')
|
|
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
empty_file = test_directory / 'empty.mp4'
|
|
empty_file.touch()
|
|
get_video_duration(empty_file)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'paths,expected',
|
|
(
|
|
(['1.mp4', ], ['1.mp4', ]),
|
|
(['1.mp4', '2.mp4'], ['1.mp4', '2.mp4']),
|
|
(['1.mp4', '1.ogg'], ['1.mp4']),
|
|
(['1.bad_ext', '1.other_ext'], ['1.bad_ext']),
|
|
)
|
|
)
|
|
def test_remove_duplicate_video_paths(paths, expected):
|
|
result = remove_duplicate_video_paths(map(pathlib.Path, paths))
|
|
assert sorted(result) == sorted(map(pathlib.Path, expected))
|
|
|
|
|
|
def test_convert_image():
|
|
"""
|
|
An image's format can be changed. This tests that convert_image() converts from a WEBP format, to a JPEG format.
|
|
"""
|
|
foo = Image.new('RGB', (25, 25), color='grey')
|
|
|
|
with tempfile.TemporaryDirectory() as tempdir:
|
|
tempdir = pathlib.Path(tempdir)
|
|
|
|
# Save the new image to "foo.webp".
|
|
existing_path = tempdir / 'foo.webp'
|
|
foo.save(existing_path)
|
|
assert Image.open(existing_path).format == 'WEBP'
|
|
|
|
destination_path = tempdir / 'foo.jpg'
|
|
assert not destination_path.is_file()
|
|
|
|
# Convert the WEBP to a JPEG. The WEBP image should be removed.
|
|
convert_image(existing_path, destination_path)
|
|
assert not existing_path.is_file()
|
|
assert destination_path.is_file()
|
|
assert Image.open(destination_path).format == 'JPEG'
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'ext,expected',
|
|
[
|
|
('jpg', True),
|
|
('jpeg', True),
|
|
('png', False),
|
|
('webp', False),
|
|
('tif', False),
|
|
]
|
|
)
|
|
def test_is_valid_poster(ext, expected, test_directory):
|
|
foo = Image.new('RGB', (25, 25), color='blue')
|
|
image_path = test_directory / f'image.{ext}'
|
|
foo.save(image_path)
|
|
assert is_valid_poster(image_path) == expected, f'is_valid_poster({image_path}) should be {expected}'
|
|
|
|
|
|
def test_update_view_count(test_session, channel_factory, video_factory):
|
|
def check_view_counts(view_counts):
|
|
for source_id, view_count in view_counts.items():
|
|
video = test_session.query(Video).filter_by(source_id=source_id).one()
|
|
assert video.view_count == view_count
|
|
|
|
channel1, channel2, channel3 = channel_factory(), channel_factory(), channel_factory()
|
|
video_factory(channel_id=channel1.id, with_poster_ext='jpg', title='vid1')
|
|
video_factory(channel_id=channel2.id, title='vid2')
|
|
video_factory(channel_id=channel3.id, with_poster_ext='jpg', title='vid3')
|
|
video_factory(channel_id=channel3.id, title='vid4')
|
|
channel1.info_json = {'entries': [{'id': 'vid1', 'view_count': 10}]}
|
|
channel2.info_json = {'entries': [{'id': 'vid2', 'view_count': 11}, {'id': 'bad_id', 'view_count': 12}]}
|
|
channel3.info_json = {'entries': [{'id': 'vid3', 'view_count': 13}, {'id': 'vid4', 'view_count': 14}]}
|
|
test_session.commit()
|
|
|
|
# Check all videos are empty.
|
|
check_view_counts({'vid1': None, 'vid2': None, 'vid3': None, 'vid4': None})
|
|
|
|
# Channel 1 is updated, the other channels are left alone.
|
|
apply_info_json(channel1.id)
|
|
check_view_counts({'vid1': 10, 'vid2': None, 'vid3': None, 'vid4': None})
|
|
|
|
# Channel 2 is updated, the other channels are left alone. The 'bad_id' video is ignored.
|
|
apply_info_json(channel2.id)
|
|
check_view_counts({'vid1': 10, 'vid2': 11, 'vid3': None, 'vid4': None})
|
|
|
|
# All videos are updated.
|
|
apply_info_json(channel3.id)
|
|
check_view_counts({'vid1': 10, 'vid2': 11, 'vid3': 13, 'vid4': 14})
|
|
|
|
# An outdated view count will be overwritten.
|
|
vid = test_session.query(Video).filter_by(id=1).one()
|
|
vid.view_count = 8
|
|
check_view_counts({'vid1': 8, 'vid2': 11, 'vid3': 13, 'vid4': 14})
|
|
apply_info_json(channel1.id)
|
|
check_view_counts({'vid1': 10, 'vid2': 11, 'vid3': 13, 'vid4': 14})
|
|
|
|
|
|
def test_generate_video_poster(video_file):
|
|
"""
|
|
A poster can be generated from a video file.
|
|
"""
|
|
poster_path = video_file.with_suffix('.jpg')
|
|
generate_video_poster(video_file)
|
|
assert poster_path.is_file(), f'{poster_path} was not created!'
|
|
assert poster_path.stat().st_size > 0
|
|
|
|
|
|
def test_update_censored_videos(test_session, video_factory, simple_channel):
|
|
vid1 = video_factory(channel_id=simple_channel.id)
|
|
vid2 = video_factory(channel_id=simple_channel.id)
|
|
vid3 = video_factory(channel_id=simple_channel.id)
|
|
vid4 = video_factory() # should not be censored because it has no channel.
|
|
test_session.commit()
|
|
|
|
def check_censored(expected):
|
|
for video_id, censored in expected:
|
|
video = test_session.query(Video).filter_by(id=video_id).one()
|
|
assert video.censored == censored
|
|
|
|
# Videos are not censored by default.
|
|
test_session.commit()
|
|
apply_info_json(simple_channel.id)
|
|
check_censored([(vid1.id, False), (vid2.id, False), (vid3.id, False), (vid4.id, False)])
|
|
|
|
# All videos are in the info_json.
|
|
simple_channel.info_json = {
|
|
'entries': [
|
|
dict(id=vid1.source_id, view_count=0),
|
|
dict(id=vid2.source_id, view_count=0),
|
|
dict(id=vid3.source_id, view_count=0),
|
|
]
|
|
}
|
|
test_session.commit()
|
|
|
|
apply_info_json(simple_channel.id)
|
|
check_censored([(vid1.id, False), (vid2.id, False), (vid3.id, False), (vid4.id, False)])
|
|
|
|
simple_channel.info_json = {
|
|
'entries': [
|
|
dict(id=vid1.source_id, view_count=0), # vid2 is missing
|
|
dict(id=vid3.source_id, view_count=0),
|
|
]
|
|
}
|
|
test_session.commit()
|
|
apply_info_json(simple_channel.id)
|
|
check_censored([(vid1.id, False), (vid2.id, True), (vid3.id, False), (vid4.id, False)])
|
|
|
|
simple_channel.info_json = {
|
|
'entries': [
|
|
dict(id=vid1.source_id, view_count=0), # vid2 is back, vid3 is missing.
|
|
dict(id=vid2.source_id, view_count=0),
|
|
]
|
|
}
|
|
test_session.commit()
|
|
apply_info_json(simple_channel.id)
|
|
check_censored([(vid1.id, False), (vid2.id, False), (vid3.id, True), (vid4.id, False)])
|
|
|
|
simple_channel.info_json = {
|
|
'entries': [] # all videos gone
|
|
}
|
|
test_session.commit()
|
|
apply_info_json(simple_channel.id)
|
|
check_censored([(vid1.id, True), (vid2.id, True), (vid3.id, True), (vid4.id, False)])
|
|
|
|
# Channels without info json preserve their last censored.
|
|
simple_channel.info_json = None
|
|
test_session.commit()
|
|
apply_info_json(simple_channel.id)
|
|
check_censored([(vid1.id, True), (vid2.id, True), (vid3.id, True), (vid4.id, False)])
|
|
|
|
|
|
def test_import_favorites(test_session, simple_channel, video_factory, test_channels_config):
|
|
"""
|
|
A favorited Video will be preserved through everything (channel deletion, DB wipe) and can be imported and will be
|
|
favorited again.
|
|
|
|
A Video's favorited status can only be cleared by calling `Video.delete`.
|
|
"""
|
|
video_factory(channel_id=simple_channel.id) # never favorited
|
|
vid2 = video_factory(channel_id=None) # has no channel
|
|
vid3 = video_factory(channel_id=simple_channel.id)
|
|
favorite = vid2.favorite = local_timezone(datetime(2000, 1, 1, 0, 0, 0))
|
|
vid3.favorite = favorite
|
|
test_session.commit()
|
|
vid2_video_path = vid2.video_path.path
|
|
|
|
favorites = {
|
|
str(simple_channel.directory.relative): {vid3.video_path.path.name: {'favorite': favorite}},
|
|
'NO CHANNEL': {vid2.video_path.path.name: {'favorite': favorite}},
|
|
}
|
|
|
|
# Save config, verify that favorite is set.
|
|
save_channels_config()
|
|
config = get_channels_config()
|
|
assert config.favorites == favorites
|
|
|
|
def import_and_verify(favorited_ids: List[int]):
|
|
with mock.patch('modules.videos.lib.get_channel_source_id', lambda i: 'foo'):
|
|
import_channels_config()
|
|
for video in test_session.query(Video).all():
|
|
if video.id in favorited_ids:
|
|
assert video.favorite
|
|
else:
|
|
assert not video.favorite
|
|
|
|
# Clear the favorite (as if the DB was wiped), verify that the favorite is imported and set.
|
|
vid2.favorite = None
|
|
import_and_verify([vid2.id, vid3.id])
|
|
|
|
# Removing the video does not delete the favorite. If the DB is wiped, we do not want to lose our favorites!
|
|
test_session.query(Video).filter_by(id=vid2.id).delete()
|
|
test_session.commit()
|
|
save_channels_config()
|
|
config = get_channels_config()
|
|
assert config.favorites == favorites
|
|
import_and_verify([vid3.id])
|
|
|
|
# Add vid2 again, it should be favorited on import.
|
|
vid2 = Video(video_path=vid2_video_path)
|
|
test_session.add(vid2)
|
|
test_session.commit()
|
|
import_and_verify([vid3.id, vid2.id])
|
|
|
|
# Deleting the Video in the model really removes the favorite status.
|
|
vid3.delete()
|
|
config = get_channels_config()
|
|
assert config.favorites == {'NO CHANNEL': {vid2.video_path.path.name: {'favorite': favorite}}}
|
|
import_and_verify([vid2.id])
|
|
|
|
|
|
def test_import_channel_downloads(test_session, channel_factory, test_channels_config):
|
|
"""Importing the Channels' config should create any missing download records"""
|
|
channel1 = channel_factory(source_id='foo')
|
|
channel2 = channel_factory(source_id='bar')
|
|
# channel2 has no url, but has a frequency. Import should not create a download record.
|
|
channel2.url = None
|
|
channel2.download_frequency = DownloadFrequency.biweekly
|
|
test_session.commit()
|
|
assert channel1.download_frequency is None
|
|
assert len(test_session.query(Channel).all()) == 2
|
|
assert test_session.query(Download).all() == []
|
|
|
|
def update_channel_config(conf, source_id, d):
|
|
for c in conf.channels:
|
|
if c['source_id'] == source_id:
|
|
c.update(d)
|
|
conf.save()
|
|
|
|
# Config has no channels with a download_frequency.
|
|
save_channels_config()
|
|
import_channels_config()
|
|
assert channel1.download_frequency is None
|
|
assert len(test_session.query(Channel).all()) == 2
|
|
assert test_session.query(Download).all() == []
|
|
|
|
# Add a frequency to the Channel.
|
|
channels_config = get_channels_config()
|
|
update_channel_config(channels_config, 'foo', {'download_frequency': DownloadFrequency.biweekly})
|
|
|
|
# Download record is created on import.
|
|
import_channels_config()
|
|
assert channel1.download_frequency is not None
|
|
assert len(test_session.query(Channel).all()) == 2
|
|
download: Download = test_session.query(Download).one()
|
|
assert download.url == channel1.url
|
|
assert download.frequency == channel1.download_frequency
|
|
|
|
# Download frequency is adjusted when config file changes.
|
|
update_channel_config(channels_config, 'foo', {'download_frequency': DownloadFrequency.weekly})
|
|
import_channels_config()
|
|
assert len(test_session.query(Channel).all()) == 2
|
|
assert download.url == channel1.url
|
|
assert download.frequency == channel1.download_frequency
|
|
assert download.downloader == 'video_channel'
|
|
assert download.next_download
|
|
|
|
next_download = str(download.next_download)
|
|
import_channels_config()
|
|
download: Download = test_session.query(Download).one()
|
|
assert next_download == str(download.next_download)
|
|
|
|
|
|
def test_import_channels_config_outdated(test_session, test_directory, channel_factory, test_channels_config,
|
|
video_factory):
|
|
"""The Channels' config used to use a "link" to differentiate channels, test that old configs can be imported."""
|
|
channel1 = channel_factory(name='Channel1', download_frequency=DownloadFrequency.weekly)
|
|
channel2 = channel_factory(name='Channel2', download_frequency=DownloadFrequency.weekly)
|
|
test_session.commit()
|
|
|
|
vid1 = video_factory(title='vid1', channel_id=channel1.id)
|
|
vid2 = video_factory(title='vid2', channel_id=channel2.id)
|
|
vid3 = video_factory(title='vid3')
|
|
vid4 = video_factory(title='vid4')
|
|
test_session.commit()
|
|
|
|
channel1_name, channel_1_directory = channel1.name, channel1.directory.path
|
|
channel2_name, channel_2_directory = channel2.name, channel2.directory.path
|
|
|
|
assert channel1.directory.path.exists() and channel1.directory.path.is_absolute()
|
|
assert channel2.directory.path.exists() and channel2.directory.path.is_absolute()
|
|
|
|
# Channel's used to have a relative directory in the DB.
|
|
test_session.execute('UPDATE channel SET directory=:directory WHERE id=2',
|
|
dict(directory=str(channel_2_directory.relative_to(test_directory))),
|
|
)
|
|
|
|
save_channels_config()
|
|
|
|
# The config has a list of Channels, use the old method of a "dict" with the sanitized name as the key.
|
|
channels_config = get_channels_config()
|
|
channels_config.channels = {sanitize_link(i['name']): i for i in channels_config.channels}
|
|
# Change frequency to verify that the channels are updated.
|
|
for link, channel in channels_config.channels.items():
|
|
channel['download_frequency'] = DownloadFrequency.biweekly
|
|
# Change the favorites to the old "link" method as well.
|
|
channels_config.favorites = {
|
|
sanitize_link(channel1.name): {str(vid1.video_path.path): {'favorite': now()}},
|
|
sanitize_link(channel2.name): {str(vid2.video_path.path): {'favorite': now()}},
|
|
'NO CHANNEL': {str(vid3.video_path.path): {'favorite': now()}},
|
|
}
|
|
channels_config.save()
|
|
|
|
assert not vid1.favorite and not vid2.favorite and not vid3.favorite and not vid3.favorite
|
|
|
|
assert test_session.query(Channel).count() == 2
|
|
|
|
with mock.patch('modules.videos.lib.get_channel_source_id') as mock_get_channel_source_id:
|
|
mock_get_channel_source_id.return_value = 'some source id'
|
|
import_channels_config()
|
|
|
|
# No new channels were created. Existing channels were updated.
|
|
assert test_session.query(Channel).count() == 2
|
|
channel1, channel2 = test_session.query(Channel).order_by(Channel.id)
|
|
assert channel1.name == channel1_name and str(channel1.directory.path) == str(channel_1_directory)
|
|
# Channel2's directory is now absolute.
|
|
assert channel2.name == channel2_name and str(channel2.directory.path) == str(channel_2_directory)
|
|
assert channel1.download_frequency == DownloadFrequency.biweekly
|
|
assert channel2.download_frequency == DownloadFrequency.biweekly
|
|
# Videos were favorited by their channel "link".
|
|
assert vid1.favorite and vid2.favorite and vid3.favorite
|
|
assert not vid4.favorite
|