rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Adding config fixture to download manager fixture to avoid calling both in every test.

Browse source
This commit is contained in:
Roland 2022-06-08 12:21:59 -06:00
parent f14a230bc6
commit ff379185fe
No known key found for this signature in database
GPG key ID: 9D6DBD3ACF11AB93
5 changed files with 126 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

4
modules/videos/channel/test/test_api.py
View file

@ -337,9 +337,7 @@ def test_channel_empty_url_doesnt_conflict(test_client, test_session, test_direc
def test_download_channel_no_refresh(test_session, download_channel, video_download_manager):
"""
A Channel cannot be downloaded until it has been refreshed.
"""
"""A Channel cannot be downloaded until it has been refreshed."""
def check_refreshed(expected: bool):
channel = test_session.query(Channel).one()

3
modules/videos/test/test_downloader.py
View file

@ -177,7 +177,8 @@ def test_video_download_no_channel(test_session, video_download_manager, video_f
def test_download_channel(test_session, simple_channel, video_download_manager, video_file,
mock_video_extract_info, mock_video_prepare_filename, mock_video_process_runner):
mock_video_extract_info, mock_video_prepare_filename,
mock_video_process_runner):
"""Downloading (updating the catalog of) a Channel creates download records for all of it's missing videos."""
url = 'https://www.youtube.com/c/LearningSelfReliance/videos'

20
wrolpi/conftest.py
View file

@ -112,7 +112,16 @@ def test_client() -> SanicTestClient:
@pytest.fixture
def test_download_manager():
def test_download_manager_config(test_directory):
(test_directory / 'config').mkdir(exist_ok=True)
config_path = test_directory / 'config/download_manager.yaml'
set_test_download_manager_config(True)
yield config_path
set_test_download_manager_config(False)
@pytest.fixture
def test_download_manager(test_download_manager_config):
manager = DownloadManager()
return manager
@ -134,12 +143,3 @@ def successful_download():
@pytest.fixture
def failed_download():
return DownloadResult(error='pytest.fixture failed_download error', success=False)
@pytest.fixture
def test_download_manager_config(test_directory):
(test_directory / 'config').mkdir(exist_ok=True)
config_path = test_directory / 'config/download_manager.yaml'
set_test_download_manager_config(True)
yield config_path
set_test_download_manager_config(False)

233
wrolpi/test/test_downloader.py
View file

@ -6,18 +6,15 @@ from unittest.mock import MagicMock
import pytest
from wrolpi import downloader
from wrolpi.dates import local_timezone, Seconds
from wrolpi.db import get_db_context
from wrolpi.downloader import Downloader, Download, DownloadFrequency, DownloadResult
from wrolpi.errors import UnrecoverableDownloadError, InvalidDownload, WROLModeEnabled
from wrolpi.test.common import wrap_test_db, TestAPI
from wrolpi.test.common import assert_dict_contains
class PermissiveDownloader(Downloader):
"""
A testing Downloader which always says it's valid.
"""
"""A testing Downloader which always says it's valid."""
name = 'permissive'
def __repr__(self):
@ -28,9 +25,7 @@ class PermissiveDownloader(Downloader):
class HTTPDownloader(Downloader):
"""
A testing Downloader which says its valid when a URL starts with http/https
"""
"""A testing Downloader which says its valid when a URL starts with http/https"""
name = 'http'
def __repr__(self):
@ -40,124 +35,100 @@ class HTTPDownloader(Downloader):
return url.startswith('https://') or url.startswith('http://'), None
class TestDownloader(TestAPI):
def setUp(self) -> None:
super().setUp()
self.mgr = downloader.DownloadManager()
PermissiveDownloader.do_download = MagicMock()
HTTPDownloader.do_download = MagicMock()
def test_register_downloader(test_session, test_download_manager):
"""Downloaders can be registered and have specific priorities."""
assert test_download_manager.instances == tuple()
def test_register_downloader(self):
"""
Downloaders can be registered and have specific priorities.
"""
self.assertEqual(self.mgr.instances, tuple())
http_downloader = HTTPDownloader()
test_download_manager.register_downloader(http_downloader)
assert test_download_manager.instances == (http_downloader,)
http_downloader = HTTPDownloader()
self.mgr.register_downloader(http_downloader)
self.assertEqual(self.mgr.instances, (http_downloader,))
# PermissiveDownloader is first priority.
permissive_downloader = PermissiveDownloader(priority=0)
test_download_manager.register_downloader(permissive_downloader)
assert test_download_manager.instances == (permissive_downloader, http_downloader)
# PermissiveDownloader is first priority.
with pytest.raises(ValueError):
test_download_manager.register_downloader(http_downloader)
with pytest.raises(ValueError):
test_download_manager.register_downloader(permissive_downloader)
def test_do_downloads(test_session, test_download_manager):
_, session = get_db_context()
http_downloader = HTTPDownloader()
http_downloader.do_download = MagicMock()
http_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.register_downloader(http_downloader)
permissive_downloader = PermissiveDownloader(priority=100)
permissive_downloader.do_download = MagicMock()
permissive_downloader.do_download.return_value = DownloadResult(success=False)
test_download_manager.register_downloader(permissive_downloader)
# https is handled by the HTTP Downloader.
test_download_manager.create_download('https://example.com', session)
http_downloader.do_download.assert_called_once()
permissive_downloader.do_download.assert_not_called()
assert test_download_manager.get_download(session, 'https://example.com') is not None
http_downloader.do_download.reset_mock()
# try the permissive download, which returns a failure.
test_download_manager.create_download('foo', session)
http_downloader.do_download.assert_not_called()
permissive_downloader.do_download.assert_called_once()
download = test_download_manager.get_download(session, 'foo')
assert download.attempts == 1
# try again
test_download_manager.create_download('foo', session)
test_download_manager._do_downloads(session)
download = test_download_manager.get_download(session, 'foo')
assert download.attempts == 2
# finally success
permissive_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.create_download('foo', session)
test_download_manager._do_downloads(session)
download = test_download_manager.get_download(session, 'foo')
assert download.status == 'complete'
# No downloads left.
assert list(test_download_manager.get_new_downloads(session)) == []
def test_delete_old_once_downloads(test_session, test_download_manager):
"""Once-downloads over a month old should be deleted."""
with mock.patch('wrolpi.downloader.now') as mock_now:
mock_now.return_value = local_timezone(datetime(2020, 6, 5, 0, 0))
permissive_downloader = PermissiveDownloader(priority=0)
self.mgr.register_downloader(permissive_downloader)
self.assertEqual(self.mgr.instances, (permissive_downloader, http_downloader))
self.assertRaises(ValueError, self.mgr.register_downloader, http_downloader)
self.assertRaises(ValueError, self.mgr.register_downloader, permissive_downloader)
@wrap_test_db
def test_ensure_download(self):
_, session = get_db_context()
self.assertEqual(len(self.mgr.get_downloads(session)), 0)
http_downloader = HTTPDownloader()
http_downloader.do_download.return_value = DownloadResult(success=True)
self.mgr.register_downloader(http_downloader)
self.mgr.create_download('https://example.com', session)
download = self.mgr.get_download(session, 'https://example.com')
self.assertEqual(download.url, 'https://example.com')
self.assertEqual(len(self.mgr.get_downloads(session)), 1)
self.assertIsNotNone(self.mgr.get_download(session, 'https://example.com'))
self.assertIsNone(self.mgr.get_download(session, 'https://example.com/not downloading'))
@wrap_test_db
def test_do_downloads(self):
_, session = get_db_context()
http_downloader = HTTPDownloader()
http_downloader.do_download.return_value = DownloadResult(success=True)
self.mgr.register_downloader(http_downloader)
permissive_downloader = PermissiveDownloader(priority=100)
permissive_downloader.do_download.return_value = DownloadResult(success=False)
self.mgr.register_downloader(permissive_downloader)
# https is handled by the HTTP Downloader.
self.mgr.create_download('https://example.com', session)
http_downloader.do_download.assert_called_once()
permissive_downloader.do_download.assert_not_called()
self.assertIsNotNone(self.mgr.get_download(session, 'https://example.com'))
http_downloader.do_download.reset_mock()
# try the permissive download, which returns a failure.
self.mgr.create_download('foo', session)
http_downloader.do_download.assert_not_called()
permissive_downloader.do_download.assert_called_once()
download = self.mgr.get_download(session, 'foo')
self.assertEqual(download.attempts, 1)
# try again
self.mgr.create_download('foo', session)
self.mgr._do_downloads(session)
download = self.mgr.get_download(session, 'foo')
self.assertEqual(download.attempts, 2)
# finally success
permissive_downloader.do_download.return_value = DownloadResult(success=True)
self.mgr.create_download('foo', session)
self.mgr._do_downloads(session)
download = self.mgr.get_download(session, 'foo')
self.assertEqual(download.status, 'complete')
# No downloads left.
self.assertEqual(list(self.mgr.get_new_downloads(session)), [])
@wrap_test_db
@mock.patch('wrolpi.downloader.now', lambda: local_timezone(datetime(2020, 6, 5, 0, 0)))
def test_delete_old_once_downloads(self):
"""
Once-downloads over a month old should be deleted.
"""
permissive_downloader = PermissiveDownloader(priority=0)
self.mgr.register_downloader(permissive_downloader)
test_download_manager.register_downloader(permissive_downloader)
_, session = get_db_context()
# Recurring downloads should not be deleted.
d1 = self.mgr.create_download('https://example.com/1', session, skip_download=True)
d2 = self.mgr.create_download('https://example.com/2', session, skip_download=True)
d1 = test_download_manager.create_download('https://example.com/1', session, skip_download=True)
d2 = test_download_manager.create_download('https://example.com/2', session, skip_download=True)
d1.frequency = 1
d2.frequency = 1
d2.started()
# Should be deleted.
d3 = self.mgr.create_download('https://example.com/3', session, skip_download=True)
d4 = self.mgr.create_download('https://example.com/4', session, skip_download=True)
d3 = test_download_manager.create_download('https://example.com/3', session, skip_download=True)
d4 = test_download_manager.create_download('https://example.com/4', session, skip_download=True)
d3.complete()
d4.complete()
d3.last_successful_download = local_timezone(datetime(2020, 1, 1, 0, 0, 0))
d4.last_successful_download = local_timezone(datetime(2020, 5, 1, 0, 0, 0))
# Not a month old.
d5 = self.mgr.create_download('https://example.com/5', session, skip_download=True)
d5 = test_download_manager.create_download('https://example.com/5', session, skip_download=True)
d5.last_successful_download = local_timezone(datetime(2020, 6, 1, 0, 0, 0))
# An old, but pending download should not be deleted.
d6 = self.mgr.create_download('https://example.com/6', session, skip_download=True)
d6 = test_download_manager.create_download('https://example.com/6', session, skip_download=True)
d6.last_successful_download = local_timezone(datetime(2020, 4, 1, 0, 0, 0))
d6.started()
self.mgr.delete_old_once_downloads()
test_download_manager.delete_old_once_downloads()
# Two old downloads are deleted.
downloads = session.query(Download).order_by(Download.id).all()
@ -168,26 +139,42 @@ class TestDownloader(TestAPI):
dict(url='https://example.com/6', attempts=1, status='pending'),
]
for download, expected in zip_longest(downloads, expected):
self.assertDictContains(download.dict(), expected)
assert_dict_contains(download.dict(), expected)
@wrap_test_db
def test_create_downloads(self):
"""Multiple downloads can be scheduled using DownloadManager.create_downloads."""
http_downloader = HTTPDownloader(priority=0)
self.mgr.register_downloader(http_downloader)
_, session = get_db_context()
def test_ensure_download(test_session, test_download_manager):
assert len(test_download_manager.get_downloads(test_session)) == 0
# Both URLs are valid and are scheduled.
self.mgr.create_downloads(['https://example.com/1', 'https://example.com/2'], skip_download=True)
downloads = self.mgr.get_downloads(session)
self.assertEqual({i.url for i in downloads}, {'https://example.com/1', 'https://example.com/2'})
http_downloader = HTTPDownloader()
http_downloader.do_download = MagicMock()
http_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.register_downloader(http_downloader)
# One URL is bad, neither should be scheduled.
self.assertRaises(InvalidDownload, self.mgr.create_downloads, ['https://example.com/3', 'bad url should fail'],
skip_download=True)
downloads = self.mgr.get_downloads(session)
self.assertEqual({i.url for i in downloads}, {'https://example.com/1', 'https://example.com/2'})
test_download_manager.create_download('https://example.com', test_session)
download = test_download_manager.get_download(test_session, 'https://example.com')
assert download.url == 'https://example.com'
assert len(test_download_manager.get_downloads(test_session)) == 1
assert test_download_manager.get_download(test_session, 'https://example.com') is not None
assert test_download_manager.get_download(test_session, 'https://example.com/not downloading') is None
def test_create_downloads(test_session, test_download_manager):
"""Multiple downloads can be scheduled using DownloadManager.create_downloads."""
http_downloader = HTTPDownloader(priority=0)
test_download_manager.register_downloader(http_downloader)
# Both URLs are valid and are scheduled.
test_download_manager.create_downloads(['https://example.com/1', 'https://example.com/2'], skip_download=True)
downloads = test_download_manager.get_downloads(test_session)
assert {i.url for i in downloads} == {'https://example.com/1', 'https://example.com/2'}
# One URL is bad, neither should be scheduled.
with pytest.raises(InvalidDownload):
test_download_manager.create_downloads(['https://example.com/3', 'bad url should fail'], skip_download=True)
downloads = test_download_manager.get_downloads(test_session)
assert {i.url for i in downloads} == {'https://example.com/1', 'https://example.com/2'}
def test_downloader_must_have_name():
@ -213,7 +200,7 @@ async def test_download_wrol_mode(test_session, test_download_manager):
await test_download_manager.do_downloads_sync()
def test_download_get_downloader(test_session, test_download_manager):
def test_download_get_downloader(test_session, test_download_manager,):
"""
A Download can find it's Downloader.
"""
@ -359,7 +346,7 @@ def test_max_attempts(test_session, test_download_manager):
assert download.status == 'failed'
def test_skip_urls(test_session, test_download_manager, test_download_manager_config):
def test_skip_urls(test_session, test_download_manager):
"""The DownloadManager will not create downloads for URLs in it's skip list."""
_, session = get_db_context()
from wrolpi.downloader import DOWNLOAD_MANAGER_CONFIG

6
wrolpi/test/test_rss.py
View file

@ -26,7 +26,7 @@ def test_rss_download_invalid(test_session, test_download_manager):
test_download_manager.register_downloader(rss_downloader)
def test_rss_download(test_session, test_download_manager, test_download_manager_config):
def test_rss_download(test_session, test_download_manager):
"""An RSS Downloader will create new Downloads for every link in the feed."""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)
@ -80,7 +80,7 @@ def test_rss_download(test_session, test_download_manager, test_download_manager
])
def test_post_rss_download(test_session, test_client, test_download_manager, test_download_manager_config):
def test_post_rss_download(test_session, test_client, test_download_manager):
"""An RSS can be downloaded in the UI"""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)
@ -101,7 +101,7 @@ def test_post_rss_download(test_session, test_client, test_download_manager, tes
assert download.attempts == 1
def test_rss_no_entries(test_session, test_download_manager, test_download_manager_config):
def test_rss_no_entries(test_session, test_download_manager):
"""An RSS feed with no entries is handled."""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)