import json
import pathlib
import tempfile
from datetime import datetime, timedelta
from http import HTTPStatus
import pytest
import pytz
from PIL import Image
from pytz import utc
from modules.archive import lib
from modules.archive.lib import get_or_create_domain_collection, get_new_archive_files, delete_archives, \
model_archive_result, get_domains
from modules.archive.models import Archive
from wrolpi.api_utils import CustomJSONEncoder
from wrolpi.collections import Collection
from wrolpi.common import get_wrolpi_config
from wrolpi.db import get_db_session
from wrolpi.files import lib as files_lib
from wrolpi.files.models import FileGroup
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ジにてこちら\nsome 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(async_client, test_directory, 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(async_client, test_directory, 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(async_client, test_session, test_directory):
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(async_client, test_session, example_singlefile):
with get_db_session(commit=True) as session:
url = 'https://wrolpi.org:443'
collection = get_or_create_domain_collection(session, url)
archive = Archive.from_paths(test_session, example_singlefile)
archive.url = url
archive.collection_id = collection.id
session.add(archive)
session.flush()
assert archive.collection == collection
assert archive.domain == 'wrolpi.org'
@pytest.mark.asyncio
async def test_archive_title(async_client, 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'),
)
async def reset_and_get_archive():
test_session.query(FileGroup).delete()
test_session.commit()
await files_lib.refresh_files()
return test_session.query(Archive).one()
archive1: Archive = await reset_and_get_archive()
assert test_session.query(FileGroup).count() == 1
assert test_session.query(Archive).count() == 1
# Readability JSON is trusted first.
assert archive1.file_group.url == 'https://example.com/json-url'
assert archive1.file_group.title == 'json title'
# Delete JSON file.
archive1.readability_json_path.unlink()
archive1: Archive = await reset_and_get_archive()
assert archive1.file_group.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('', ''))
archive1: Archive = await reset_and_get_archive()
# No title/URL could be found.
assert archive1.file_group.url is None
assert archive1.file_group.title is None
@pytest.mark.asyncio
async def test_archive_refresh_deleted_archive(async_client, test_session, archive_directory, archive_factory):
"""Archives/domain collections 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(Collection).filter_by(
kind='domain').count() == domain_count, 'domain collection count does not match'
# All 5 archives are already in the DB.
check_counts(archive_count=5, domain_count=2)
await async_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()
await async_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()
await async_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()
await async_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()
await async_client.post('/api/files/refresh')
check_counts(archive_count=0, domain_count=0)
@pytest.mark.asyncio
async def test_fills_contents_with_refresh(async_client, 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
@pytest.mark.asyncio
async def test_delete_archive(async_client, 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
# Save paths before deletion (archives become detached after delete)
archive1_paths = list(archive1.my_paths())
archive2_paths = list(archive2.my_paths())
archive3_paths = list(archive3.my_paths())
# 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_paths and not any(i.is_file() for i in archive1_paths)
assert archive3_paths and not any(i.is_file() for i in archive3_paths)
# Archive2 is untouched
assert archive2_paths and all(i.is_file() for i in archive2_paths)
# Delete the last archive. The domain collection should also be deleted.
delete_archives(archive2.id)
assert test_session.query(Archive).count() == 0
domain = test_session.query(Collection).filter_by(kind='domain').one_or_none()
assert domain is None
# All Files were deleted.
assert test_session.query(FileGroup).count() == 0
assert archive2_paths and not any(i.is_file() for i in archive2_paths)
def test_get_domains(test_session, archive_factory):
"""
`get_domains` gets only domain collections 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()] == []
@skip_circleci
@pytest.mark.asyncio
async def test_new_archive(test_session, test_directory, 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.file_group.download_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.
# Note: We check for content presence rather than exact formatting because BeautifulSoup's
# prettify() can produce different output on different platforms (Mac vs CI).
singlefile_text = archive1.singlefile_path.read_text()
assert 'Page saved with SingleFile' in singlefile_text
assert 'test single-file' in singlefile_text
assert 'ジにてこちら' in singlefile_text
assert '' in singlefile_text
assert 'some title' in singlefile_text
assert singlefile_text.count('\n') >= 5 # Verify it's properly formatted with newlines
readability_text = archive1.readability_path.read_text()
assert 'test readability content' in readability_text
assert readability_text.count('\n') >= 3 # Verify it's properly formatted with newlines
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 collection is reused.
assert archive1.domain == archive2.domain
@pytest.mark.asyncio
async def test_get_title_from_html(test_directory, 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 = '<html>\ntest single-file\nジにてこちら\n<title>some title