import json
import pathlib
import tempfile
from datetime import datetime, timedelta
from http import HTTPStatus
import pytest
from PIL import Image
from modules.archive import lib
from modules.archive.lib import get_or_create_domain, get_new_archive_files, delete_archives, model_archive_result, \
get_domains
from modules.archive.models import Archive, Domain
from wrolpi.db import get_db_session
from wrolpi.files import lib as files_lib
from wrolpi.files.models import FileGroup
from wrolpi.root_api import CustomJSONEncoder
from wrolpi.test.common import skip_circleci
def make_fake_archive_result(readability=True, screenshot=True, title=True):
with tempfile.NamedTemporaryFile(suffix='.png') as fh:
Image.new('RGB', (1, 1), color='grey').save(fh)
fh.seek(0)
image = fh.read()
singlefile = '\ntest single-file\nジにてこちら\n
some title'
r = dict(
content=f'test readability content',
textContent='test readability textContent',
title='ジにてこちら' if title else None,
) if readability else None
s = image if screenshot else None
return singlefile, r, s
@pytest.mark.asyncio
async def test_no_screenshot(test_session):
singlefile, readability, screenshot = make_fake_archive_result(screenshot=False)
archive = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert isinstance(archive.singlefile_path, pathlib.Path)
assert isinstance(archive.readability_path, pathlib.Path)
# Screenshot was empty
assert archive.screenshot_path is None
@pytest.mark.asyncio
async def test_no_readability(test_session):
singlefile, readability, screenshot = make_fake_archive_result(readability=False)
archive = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert isinstance(archive.singlefile_path, pathlib.Path)
assert isinstance(archive.screenshot_path, pathlib.Path)
# Readability empty
assert archive.readability_path is None
assert archive.readability_txt_path is None
@pytest.mark.asyncio
async def test_dict(test_session):
singlefile, readability, screenshot = make_fake_archive_result()
d = (await model_archive_result('https://example.com', singlefile, readability, screenshot)).dict()
assert isinstance(d, dict)
json.dumps(d, cls=CustomJSONEncoder)
@pytest.mark.asyncio
async def test_relationships(test_session, test_directory, example_singlefile):
with get_db_session(commit=True) as session:
url = 'https://wrolpi.org:443'
domain = get_or_create_domain(session, url)
archive = Archive.from_paths(test_session, example_singlefile)
archive.url = url
archive.domain_id = domain.id
session.add(archive)
session.flush()
assert archive.domain == domain
@pytest.mark.asyncio
async def test_archive_title(test_session, archive_factory, singlefile_contents_factory):
"""An Archive's title can be fetched in multiple ways. This tests from most to least reliable."""
# Create some test files, delete all records for a fresh refresh.
archive_factory(
domain='example.com',
url='https://example.com/json-url',
title='json title',
singlefile_contents=singlefile_contents_factory('singlefile title', 'https://example.com/singlefile-url'),
)
test_session.query(FileGroup).delete()
test_session.commit()
await files_lib.refresh_files()
assert test_session.query(FileGroup).count() == 1
assert test_session.query(Archive).count() == 1
archive1: Archive = test_session.query(Archive).one()
# Readability JSON is trusted first.
assert archive1.url == 'https://example.com/json-url'
assert archive1.file_group.title == 'json title'
# Delete JSON file.
archive1.readability_json_path.unlink()
test_session.delete(archive1)
await files_lib.refresh_files()
archive1: Archive = test_session.query(Archive).one()
assert archive1.url == 'https://example.com/singlefile-url'
assert archive1.file_group.title == 'singlefile title'
# Clear out title and URL from Singlefile.
archive1.singlefile_path.write_text(singlefile_contents_factory('', ''))
test_session.delete(archive1)
await files_lib.refresh_files()
# No title/URL could be found.
archive1: Archive = test_session.query(Archive).one()
assert archive1.url is None
assert archive1.file_group.title is None
def test_archive_refresh_deleted_archive(test_client, 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()
test_session.commit()
# Empty directories should be ignored.
(archive_directory / 'empty').mkdir()
def check_counts(archive_count, domain_count):
assert test_session.query(Archive).count() == archive_count, 'Archive count does not match'
assert test_session.query(Domain).count() == domain_count, 'Domain count does not match'
# All 5 archives are already in the DB.
check_counts(archive_count=5, domain_count=2)
test_client.post('/api/files/refresh')
check_counts(archive_count=5, domain_count=1)
# Delete archive2's files, it's the latest for 'https://example.com/1'
for path in archive2.my_paths():
path.unlink()
test_client.post('/api/files/refresh')
check_counts(archive_count=4, domain_count=1)
# Delete archive1's files, now the URL is empty.
for path in archive1.my_paths():
path.unlink()
test_client.post('/api/files/refresh')
check_counts(archive_count=3, domain_count=0)
# Delete archive3, now there is now example.com domain
for path in archive3.my_paths():
path.unlink()
test_client.post('/api/files/refresh')
check_counts(archive_count=2, domain_count=0)
# Delete all the rest of the archives
for path in archive4.my_paths():
path.unlink()
for path in archive5.my_paths():
path.unlink()
test_client.post('/api/files/refresh')
check_counts(archive_count=0, domain_count=0)
@pytest.mark.asyncio
async def test_fills_contents_with_refresh(test_session, archive_factory, singlefile_contents_factory):
"""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')
# Title can be found in archive3's singlefile
archive3 = archive_factory('example.org', singlefile_contents=singlefile_contents_factory('last title'))
archive4 = archive_factory('example.org') # has no contents
# Clear the title, this will be filled in.
archive1.file_group.title = archive2.file_group.title = archive3.file_group.title = archive4.file_group.title = None
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:
archive: Archive
readability_txt_path = archive.singlefile_path.with_suffix('.readability.txt')
with readability_txt_path.open('wt') as fh:
fh.write(contents)
if json_title:
readability_json_path = archive.singlefile_path.with_suffix('.readability.json')
readability_json_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.file_group.d_text
assert not archive2.file_group.d_text
assert not archive3.file_group.d_text
assert not archive4.file_group.d_text
# Fill the contents.
await files_lib.refresh_files()
# The archives will be renamed with their title.
archive1, archive2, archive3, archive4 = test_session.query(Archive).order_by(Archive.id)
assert not archive4.file_group.d_text
# Missing titles are filled.
assert archive1.file_group.title == 'my archive'
assert archive2.file_group.title == 'other archive'
assert archive3.file_group.title == 'last title' # from singlefile HTML
def test_delete_archive(test_session, archive_factory):
"""Archives can be deleted."""
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')
test_session.commit()
# Files are real.
assert archive1.my_paths() and all(i.is_file() for i in archive1.my_paths())
assert archive2.my_paths() and all(i.is_file() for i in archive2.my_paths())
assert archive3.my_paths() and all(i.is_file() for i in archive3.my_paths())
assert test_session.query(Archive).count() == 3
assert test_session.query(FileGroup).count() == 3
# Delete the oldest.
delete_archives(archive1.id, archive3.id)
assert test_session.query(Archive).count() == 1
assert test_session.query(FileGroup).count() == 1
# Files were deleted.
assert archive1.my_paths() and not any(i.is_file() for i in archive1.my_paths())
assert archive3.my_paths() and not any(i.is_file() for i in archive3.my_paths())
# Archive2 is untouched
assert archive2.my_paths() and all(i.is_file() for i in archive2.my_paths())
# Delete the last archive. The Domain should also be deleted.
delete_archives(archive2.id)
assert test_session.query(Archive).count() == 0
domain = test_session.query(Domain).one_or_none()
assert domain is None
# All Files were deleted.
assert test_session.query(FileGroup).count() == 0
assert archive2.my_paths() and not any(i.is_file() for i in archive2.my_paths())
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')
test_session.commit()
assert archive1.domain
assert archive2.domain
assert archive3.domain
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.asyncio
async def test_new_archive(test_session, fake_now):
singlefile, readability, screenshot = make_fake_archive_result()
fake_now(datetime(2000, 1, 1))
archive1 = await model_archive_result('https://example.com', singlefile, readability, screenshot)
# Everything is filled out.
assert isinstance(archive1, Archive)
assert archive1.archive_datetime is not None
assert isinstance(archive1.singlefile_path, pathlib.Path)
assert isinstance(archive1.readability_path, pathlib.Path)
assert isinstance(archive1.readability_txt_path, pathlib.Path)
assert isinstance(archive1.screenshot_path, pathlib.Path)
assert archive1.file_group.title == 'ジにてこちら'
assert archive1.url is not None
assert archive1.domain is not None
# The actual files were dumped and read correctly. The HTML/JSON files are reformatted.
assert archive1.singlefile_path.read_text() == '''
test single-file
ジにてこちら
some title
'''
assert archive1.readability_path.read_text() == '''
test readability content
'''
with open(archive1.readability_txt_path) as fh:
assert fh.read() == 'test readability textContent'
assert archive1.readability_json_path.read_text() == '''{
"title": "\\u30b8\\u306b\\u3066\\u3053\\u3061\\u3089",
"url": "https://example.com"
}'''
fake_now(datetime(2000, 1, 2))
archive2 = await model_archive_result('https://example.com', singlefile, readability, screenshot)
# 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))
singlefile, readability, screenshot = make_fake_archive_result()
archive = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert archive.file_group.title == 'ジにてこちら'
singlefile = '\ntest single-file\nジにてこちら\nsome title'
readability = dict(
content=f'test readability content',
textContent='test readability textContent',
)
screenshot = b'screenshot data'
fake_now(datetime(2000, 1, 2))
archive = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert archive.file_group.title == 'some title'
singlefile = ''
readability = dict(
content=f'missing a title',
textContent='',
)
screenshot = b'screenshot data'
fake_now(datetime(2000, 1, 3))
archive = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert archive.file_group.title is None
@skip_circleci
def test_get_new_archive_files_length(fake_now):
"""Archive titles are truncated to fit file system length limit. (255-character limit for most file systems)"""
fake_now(datetime(2001, 1, 1))
archive_files = get_new_archive_files('https://example.com', 'a' * 500)
assert len(str(archive_files.singlefile.name)) == 225
assert len(str(archive_files.readability.name)) == 237
assert len(str(archive_files.readability_txt.name)) == 236
assert len(str(archive_files.readability_json.name)) == 237
assert len(str(archive_files.screenshot.name)) == 224
@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, image_bytes_factory):
"""
The Archive files have the title in the path.
"""
fake_now(datetime(2000, 1, 1))
singlefile, readability, screenshot = make_fake_archive_result()
archive1 = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert archive1.file_group.title == 'ジにてこちら'
assert str(archive1.singlefile_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_ジにてこちら.html'
assert str(archive1.readability_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_ジにてこちら.readability.html'
assert str(archive1.readability_txt_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_ジにてこちら.readability.txt'
assert str(archive1.readability_json_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_ジにてこちら.readability.json'
assert str(archive1.screenshot_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_ジにてこちら.png'
singlefile = '\ntest single-file\nジにてこちら\n' # no title in HTML
readability = dict(
# No Title from Readability.
content=f'test readability content',
textContent='test readability textContent',
)
screenshot = image_bytes_factory()
archive2 = await model_archive_result('https://example.com', singlefile, readability, screenshot)
assert str(archive2.singlefile_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_NA.html'
assert str(archive2.readability_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_NA.readability.html'
assert str(archive2.readability_txt_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_NA.readability.txt'
assert str(archive2.readability_json_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_NA.readability.json'
assert str(archive2.screenshot_path.relative_to(test_directory)) == \
'archive/example.com/2000-01-01-00-00-00_NA.png'
singlefile = '\ntest single-file\nジにてこちら\ndangerous ;\\//_title