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
wrolpi/modules/archive/test/test_lib.py
Roland b6b04f2739
Removing unused code. Updating comments. 
Removing timeout.  Updating tests now that downloads work.

Bumping version for async downloads.

Downloads run asynchronously.  Using aiohttp to make archive http request so ArchiveDownloader.do_download can be asynchronous.

Removing session for create_download(s) usages.

Reworkingk create_downloads to use optional_session.

Canceling download workers after tests and when stopping the manager.

All tests passing.

Tests pass!

Downloader tests pass.

DownloadManager uses multiple workers to download domains in parallel.
2022-07-22 16:35:42 -06:00

537 lines
23 KiB
Python
Raw Blame History

import json
import pathlib
from datetime import datetime
import mock
import pytest
from modules.archive import lib
from modules.archive.lib import get_or_create_domain, _refresh_archives, \
get_new_archive_files, delete_archive, do_archive, get_domains, get_archive_directory, group_archive_files, \
ArchiveFiles
from modules.archive.models import Archive, Domain
from wrolpi.dates import local_timezone
from wrolpi.db import get_db_session
from wrolpi.media_path import MediaPath
from wrolpi.root_api import CustomJSONEncoder
from wrolpi.test.common import skip_circleci
def make_fake_request_archive(readability=True, screenshot=True, title=True):
async def fake_request_archive(_):
singlefile = '<html>\ntest single-file\nジにてこちら\n<title>some title</title></html>'
r = dict(
content=f'<html>test readability content</html>',
textContent='<html>test readability textContent</html>',
title='ジにてこちら' if title else None,
) if readability else None
s = b'screenshot data' if screenshot else None
return singlefile, r, s
return fake_request_archive
@pytest.mark.asyncio
async def test_no_screenshot(test_session):
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive(screenshot=False)):
archive = await do_archive('https://example.com')
assert isinstance(archive.singlefile_path, MediaPath)
assert isinstance(archive.readability_path, MediaPath)
# Screenshot was empty
assert archive.screenshot_path is None
@pytest.mark.asyncio
async def test_no_readability(test_session):
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive(readability=False)):
archive = await do_archive('https://example.com')
assert isinstance(archive.singlefile_path, MediaPath)
assert isinstance(archive.screenshot_path, MediaPath)
# Readability empty
assert archive.readability_path is None
assert archive.readability_txt_path is None
@pytest.mark.asyncio
async def test_dict(test_session):
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive()):
d = (await do_archive('https://example.com')).dict()
assert isinstance(d, dict)
json.dumps(d, cls=CustomJSONEncoder)
@pytest.mark.asyncio
async def test_relationships(test_session, test_directory):
with get_db_session(commit=True) as session:
url = 'https://wrolpi.org:443'
domain = get_or_create_domain(session, url)
archive = Archive(
singlefile_path=f'{test_directory.name}/wrolpi.org:443/foo',
title='bar',
url=url,
domain_id=domain.id,
)
session.add(archive)
session.flush()
assert archive.domain == domain
@pytest.mark.asyncio
async def test_validate_paths(test_session):
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive()):
archive = await do_archive('https://example.com')
try:
with get_db_session(commit=True):
archive.singlefile_path = 'asdf'
except ValueError as e:
assert 'relative' in str(e), f'Relative path error was not raised'
def test_archive_refresh_deleted_archive(test_session, archive_directory, archive_factory):
"""
Archives/Domains should be deleted when archive files are deleted.
"""
archive1 = archive_factory('example.com', 'https://example.com/1')
archive2 = archive_factory('example.com', 'https://example.com/1')
archive3 = archive_factory('example.com')
archive4 = archive_factory('example.org')
archive5 = archive_factory()
# Empty directories should be ignored.
empty = archive_directory / 'empty'
empty.mkdir()
def check_counts(archive_count, domain_count):
assert test_session.query(Archive).count() == archive_count
assert test_session.query(Domain).count() == domain_count
# All 3 archives are already in the DB.
check_counts(archive_count=5, domain_count=3)
_refresh_archives()
check_counts(archive_count=5, domain_count=3)
# Delete archive2's files, it's the latest for 'https://example.com/1'
for path in archive2.my_paths():
path.unlink()
_refresh_archives()
check_counts(archive_count=4, domain_count=3)
# Delete archive1's files, now the URL is empty.
for path in archive1.my_paths():
path.unlink()
_refresh_archives()
check_counts(archive_count=3, domain_count=3)
# Delete archive3, now there is now example.com domain
for path in archive3.my_paths():
path.unlink()
_refresh_archives()
check_counts(archive_count=2, domain_count=2)
# Delete all the rest of the archives
for path in archive4.my_paths():
path.unlink()
for path in archive5.my_paths():
path.unlink()
_refresh_archives()
check_counts(archive_count=0, domain_count=0)
def test_fills_contents_with_refresh(test_session, archive_factory, test_client):
"""Refreshing archives fills in any missing contents."""
archive1 = archive_factory('example.com', 'https://example.com/one')
archive2 = archive_factory('example.com', 'https://example.com/one')
archive3 = archive_factory('example.org')
archive4 = archive_factory('example.org') # has no contents
# Clear the title, this will be filled in.
archive1.title = archive2.title = archive3.title = archive4.title = None
# Title can be found in archive3's singlefile
archive3.singlefile_path.path.write_text('<html> some stuff<title>last title</title> other stuff</html>')
contents_title_archive = [
('foo bar qux', 'my archive', archive1),
('foo baz qux qux', 'other archive', archive2),
('baz qux qux qux', None, archive3),
]
for contents, json_title, archive in contents_title_archive:
with archive.readability_txt_path.path.open('wt') as fh:
fh.write(contents)
if json_title:
archive.readability_json_path.path.write_text(json.dumps({'title': json_title, 'url': archive.url}))
test_session.commit()
# Contents are empty. Archives are all missing the title.
assert not archive1.contents
assert not archive2.contents
assert not archive3.contents
assert not archive4.contents
# Fill the contents.
_refresh_archives()
# The archives will be renamed with their title.
archive1, archive2, archive3, archive4 = test_session.query(Archive).order_by(Archive.id)
assert archive1.contents
assert archive2.contents
assert archive3.contents
assert not archive4.contents
# Missing title is filled.
assert archive1.title == 'my archive'
assert archive2.title == 'other archive'
assert archive3.title == 'last title' # from json
def test_delete_archive(test_session, archive_factory):
archive1 = archive_factory('example.com', 'https://example.com/1')
archive2 = archive_factory('example.com', 'https://example.com/1')
archive3 = archive_factory('example.com', 'https://example.com/1')
# Delete the oldest.
delete_archive(archive1.id)
# Delete the latest.
delete_archive(archive3.id)
# Delete the last archive. The Domain should also be deleted.
delete_archive(archive2.id)
domain = test_session.query(Domain).one_or_none()
assert domain is None
def test_get_domains(test_session, archive_factory):
"""
`get_domains` gets only Domains with Archives.
"""
archive1 = archive_factory('example.com')
archive2 = archive_factory('example.com')
archive3 = archive_factory('example.org')
assert [i['domain'] for i in get_domains()] == ['example.com', 'example.org']
archive2.delete()
test_session.commit()
assert [i['domain'] for i in get_domains()] == ['example.com', 'example.org']
archive1.delete()
test_session.commit()
assert [i['domain'] for i in get_domains()] == ['example.org']
archive3.delete()
test_session.commit()
assert [i['domain'] for i in get_domains()] == []
@pytest.mark.asycio
async def test_new_archive(test_session, fake_now):
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive()):
fake_now(datetime(2000, 1, 1))
archive1 = await do_archive('https://example.com')
# Everything is filled out.
assert isinstance(archive1, Archive)
assert archive1.archive_datetime is not None
assert isinstance(archive1.singlefile_path, MediaPath)
assert isinstance(archive1.readability_path, MediaPath)
assert isinstance(archive1.readability_txt_path, MediaPath)
assert isinstance(archive1.screenshot_path, MediaPath)
assert archive1.title == 'ジにてこちら'
assert archive1.url is not None
assert archive1.domain is not None
# The actual files were dumped and read correctly.
with open(archive1.singlefile_path.path) as fh:
assert fh.read() == '<html>\ntest single-file\nジにてこちら\n<title>some title</title></html>'
with open(archive1.readability_path.path) as fh:
assert fh.read() == '<html>test readability content</html>'
with open(archive1.readability_txt_path.path) as fh:
assert fh.read() == '<html>test readability textContent</html>'
with open(archive1.readability_json_path.path) as fh:
assert json.load(fh) == {'title': 'ジにてこちら', 'url': 'https://example.com'}
fake_now(datetime(2000, 1, 2))
archive2 = await do_archive('https://example.com')
# Domain is reused.
assert archive1.domain == archive2.domain
@pytest.mark.asyncio
async def test_get_title_from_html(test_session, fake_now):
fake_now(datetime(2000, 1, 1))
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive()):
archive = await do_archive('example.com')
assert archive.title == 'ジにてこちら'
async def fake_request_archive(_):
singlefile = '<html>\ntest single-file\nジにてこちら\n<title>some title</title></html>'
r = dict(
content=f'<html>test readability content</html>',
textContent='<html>test readability textContent</html>',
)
s = b'screenshot data'
return singlefile, r, s
fake_now(datetime(2000, 1, 2))
with mock.patch('modules.archive.lib.request_archive', fake_request_archive):
archive = await do_archive('example.com')
assert archive.title == 'some title'
async def fake_request_archive(_):
singlefile = '<html></html>'
r = dict(
content=f'<html>missing a title</html>',
textContent='',
)
s = b'screenshot data'
return singlefile, r, s
fake_now(datetime(2000, 1, 3))
with mock.patch('modules.archive.lib.request_archive', fake_request_archive):
archive = await do_archive('example.com')
assert archive.title is None
@skip_circleci
def test_get_new_archive_files(fake_now):
"""Archive files have a specific format so they are sorted by datetime, and are near each other."""
fake_now(datetime(2001, 1, 1))
archive_files = get_new_archive_files('https://example.com/two', None)
assert str(archive_files.singlefile).endswith('archive/example.com/2001-01-01-00-00-00_NA.html')
assert str(archive_files.readability).endswith('archive/example.com/2001-01-01-00-00-00_NA.readability.html')
assert str(archive_files.readability_txt).endswith('archive/example.com/2001-01-01-00-00-00_NA.readability.txt')
assert str(archive_files.readability_json).endswith('archive/example.com/2001-01-01-00-00-00_NA.readability.json')
assert str(archive_files.screenshot).endswith('archive/example.com/2001-01-01-00-00-00_NA.png')
archive_files = get_new_archive_files('https://www.example.com/one', 'Title')
# Leading www. is removed.
assert str(archive_files.singlefile).endswith('archive/example.com/2001-01-01-00-00-00_Title.html')
assert str(archive_files.readability).endswith('archive/example.com/2001-01-01-00-00-00_Title.readability.html')
assert str(archive_files.readability_txt).endswith('archive/example.com/2001-01-01-00-00-00_Title.readability.txt')
assert str(archive_files.readability_json).endswith(
'archive/example.com/2001-01-01-00-00-00_Title.readability.json')
assert str(archive_files.screenshot).endswith('archive/example.com/2001-01-01-00-00-00_Title.png')
@skip_circleci
@pytest.mark.asyncio
async def test_title_in_filename(test_session, fake_now, test_directory):
"""
The Archive files have the title in the path.
"""
fake_now(datetime(2000, 1, 1))
with mock.patch('modules.archive.lib.request_archive', make_fake_request_archive()):
archive1 = await do_archive('example.com')
assert archive1.title == 'ジにてこちら'
assert str(archive1.singlefile_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_ジにてこちら.html'
assert str(archive1.readability_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_ジにてこちら.readability.html'
assert str(archive1.readability_txt_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_ジにてこちら.readability.txt'
assert str(archive1.readability_json_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_ジにてこちら.readability.json'
assert str(archive1.screenshot_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_ジにてこちら.png'
async def fake_request_archive(_):
singlefile = '<html>\ntest single-file\nジにてこちら\n</html>' # no title in HTML
r = dict(
# No Title from Readability.
content=f'<html>test readability content</html>',
textContent='<html>test readability textContent</html>',
)
s = b'screenshot data'
return singlefile, r, s
with mock.patch('modules.archive.lib.request_archive', fake_request_archive):
archive2 = await do_archive('example.com')
assert str(archive2.singlefile_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_NA.html'
assert str(archive2.readability_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_NA.readability.html'
assert str(archive2.readability_txt_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_NA.readability.txt'
assert str(archive2.readability_json_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_NA.readability.json'
assert str(archive2.screenshot_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_NA.png'
async def fake_request_archive(_):
singlefile = '<html>\ntest single-file\nジにてこちら\n<title>dangerous ;\\//_title</html></html>'
r = dict(
# No Title from Readability.
content=f'<html>test readability content</html>',
textContent='<html>test readability textContent</html>',
)
s = b'screenshot data'
return singlefile, r, s
with mock.patch('modules.archive.lib.request_archive', fake_request_archive):
archive3 = await do_archive('example.com')
assert str(archive3.singlefile_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_dangerous ;_title.html'
assert str(archive3.readability_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_dangerous ;_title.readability.html'
assert str(archive3.readability_txt_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_dangerous ;_title.readability.txt'
assert str(archive3.readability_json_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_dangerous ;_title.readability.json'
assert str(archive3.screenshot_path.path.relative_to(test_directory)) == \
'archive/2000-01-01-00-00-00_dangerous ;_title.png'
@skip_circleci
def test_refresh_archives(test_session, test_directory):
"""Archives can be found and put in the database. A single Archive will have multiple files."""
archive_directory = get_archive_directory()
# Make some test files to refresh.
example_dir = archive_directory / 'example.com'
example_dir.mkdir()
(example_dir / '2021-10-05 16:20:10.346823.html').touch() # renames to 2021-10-05-16-20-10_NA.html
(example_dir / '2021-10-05 16:20:10.346823.png').touch()
(example_dir / '2021-10-05 16:20:10.346823-readability.txt').touch()
(example_dir / '2021-10-05 16:20:10.346823-readability.json').touch()
(example_dir / '2021-10-05 16:20:10.346823-readability.html').touch()
with (example_dir / '2021-10-05 16:20:10.346823-readability.json').open('wt') as fh:
json.dump({'url': 'foo'}, fh)
# These should log an error because they are missing the singlefile path.
(example_dir / '2021-10-05 16:20:10.readability.html').touch()
(example_dir / '2021-10-05 16:20:11.readability.json').touch()
# Bad files should also be ignored.
(example_dir / 'random file').touch()
(example_dir / '2021-10-05 16:20:10.346827-something.html').mkdir()
(example_dir / '2021-10-05 16:20:10 no extension').touch()
# The single archive is found.
_refresh_archives()
with get_db_session() as session:
assert session.query(Archive).count() == 1
# Running the refresh does not result in a new archive.
_refresh_archives()
with get_db_session() as session:
assert session.query(Archive).count() == 1
# Archives file format was changed, lets check the new formats are found.
(example_dir / '2021-10-05-16-20-11_The Title.html').touch()
(example_dir / '2021-10-05-16-20-11_The Title.readability.json').write_text(json.dumps({'url': 'bar'}))
_refresh_archives()
# The old formatted archive above is renamed.
assert (example_dir / '2021-10-05-16-20-10_NA.html').is_file()
with get_db_session() as session:
assert session.query(Archive).count() == 2
def test_group_archive_files(test_directory):
"""Archive files should be grouped together when they are the same Archive."""
files = [
pathlib.Path('2021-10-05 16:20:10.346823.html'),
pathlib.Path('2021-10-05 16:20:10.346823.readability.json'),
pathlib.Path('2000-01-01-00-00-00_Title.html'),
pathlib.Path('2000-01-01-00-00-00_Title.readability.json'),
pathlib.Path('not an archive'),
pathlib.Path('2000-01-01-00-00-01_Missing a singlefile.readability.json'),
]
group1 = ArchiveFiles(
singlefile=pathlib.Path('2000-01-01-00-00-00_Title.html'),
readability_json=pathlib.Path('2000-01-01-00-00-00_Title.readability.json')
)
group2 = ArchiveFiles(
singlefile=pathlib.Path('2021-10-05 16:20:10.346823.html'),
readability_json=pathlib.Path('2021-10-05 16:20:10.346823.readability.json'),
)
assert list(group_archive_files(files)) == [
(local_timezone(datetime(2000, 1, 1, 0, 0, 0)), group1),
(local_timezone(datetime(2021, 10, 5, 16, 20, 10)), group2),
]
@pytest.mark.parametrize(
'name,expected', [
('foo', False),
('2000-01-01-00-00-00_Some Title.html', True),
('2000-01-01-00-00-00_Some Title.readability.json', True),
('2000-01-01-00-00-00_Some Title.readability.html', True),
('2000-01-01-00-00-00_Some Title.readability.txt', True),
('2000-01-01-00-00-00_Some Title.png', True),
('2000-01-01-00-00-00_Some Title.jpg', True),
('2000-01-01-00-00-00_Some Title.jpeg', True),
('2000-01-01-00-00-00_Some NA.html', True),
('2000-01-01-00-00-00_NA.readability.json', True),
('2000-01-01-00-00-00_NA.readability.html', True),
('2000-01-01-00-00-00_NA.readability.txt', True),
('2000-01-01-00-00-00_NA.png', True),
('2000-01-01-00-00-00_NA.jpg', True),
('2000-01-01-00-00-00_NA.jpeg', True),
('2000-01-01-00-00-00_ジにてこちら.html', True),
('2000-01-01-00-00-00no underscore.html', False),
]
)
def test_is_archive_file(name, expected, test_directory):
path = test_directory / name
path.touch()
assert lib.is_archive_file(path) == expected
@skip_circleci
def test_migrate_archive_files(test_session, archive_directory, archive_factory):
"""Test that migration of all old Archive formatted files are migrated."""
# Create some updated archives that should be ignored.
archive1 = archive_factory('example.com', title='the title')
archive2 = archive_factory('example.com')
archive3 = archive_factory('example.org')
assert archive1.singlefile_path.path.is_file()
assert archive2.singlefile_path.path.is_file()
assert archive3.singlefile_path.path.is_file()
example_com_directory = archive_directory / 'example.com'
example_org_directory = archive_directory / 'example.org'
# Archive4
(example_com_directory / '2021-10-05 16:20:10.346823.html').touch()
path = example_com_directory / '2021-10-05 16:20:10.346823.readability.json'
path.write_text(json.dumps({'title': 'my title'}))
# Archive5
(example_com_directory / '2000-01-01 00:00:00.000000.html').touch()
path = example_com_directory / '2000-01-01 00:00:00.000000.readability.json'
path.write_text(json.dumps({'title': 'invalid#*: title'}))
# Archive6
(example_org_directory / '2000-01-01 00:00:00.000000.html').touch()
path = example_org_directory / '2000-01-01 00:00:00.000000.readability.json'
path.write_text(json.dumps({'title': 'A' * 500})) # Really long file name.
# Migration forms a plan, then performs it.
plan = lib.migrate_archive_files()
for old, new in plan:
# All files are moved.
assert not old.is_file()
assert new.is_file()
# Check that only the outdated files are moved.
plan = [(i.relative_to(archive_directory), j.relative_to(archive_directory)) for i, j in plan]
assert plan == [
(pathlib.Path('example.com/2000-01-01 00:00:00.000000.html'),
pathlib.Path('example.com/2000-01-01-00-00-00_invalid# title.html')),
(pathlib.Path('example.com/2000-01-01 00:00:00.000000.readability.json'),
pathlib.Path('example.com/2000-01-01-00-00-00_invalid# title.readability.json')),
(pathlib.Path('example.com/2021-10-05 16:20:10.346823.html'),
pathlib.Path('example.com/2021-10-05-16-20-10_my title.html')),
(pathlib.Path('example.com/2021-10-05 16:20:10.346823.readability.json'),
pathlib.Path('example.com/2021-10-05-16-20-10_my title.readability.json')),
(pathlib.Path('example.org/2000-01-01 00:00:00.000000.html'),
pathlib.Path('example.org/2000-01-01-00-00-00_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.html')),
(pathlib.Path('example.org/2000-01-01 00:00:00.000000.readability.json'),
pathlib.Path(
'example.org/2000-01-01-00-00-00_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.readability.json')),
]
# New Archive files still exist.
assert all(i.is_file() for i in archive1.my_paths())
assert all(i.is_file() for i in archive2.my_paths())
assert all(i.is_file() for i in archive3.my_paths())
Reference in a new issue View git blame Copy permalink