mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
673c23188c
A RequestSession no longer expires loaded objects at commit, so a handler serializes what it committed without a reload query. An AsyncSession cannot run that reload at all, so this is groundwork for moving request handling to async. Background sessions (get_db_session) keep SQLAlchemy's default: they change data outside the ORM (raw SQL, triggers) and rely on the reload. Collection, CollectionItem and Channel dict() now run inside the serializer guard. Fixes it found: - Channel.dict() no longer loads the deferred yt-dlp catalog (info_json), which the channel handler loaded only to discard. - get_channel eager-loads what Channel.dict() reads (populate_existing replaces a refresh query). - create_channel / get_or_create_download link through relationships, so a loaded Collection.downloads stays current. - Setting a Repo to Never removes the deleted Download from the loaded collection; the response reported the old frequency. - The downloader no longer calls channel.dict() to preload a Channel; its callers read it inside the session. Tests run each request with expiry off, as production does, and turn it back on for the background work that shares the test Session. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
631 lines
31 KiB
Python
631 lines
31 KiB
Python
import json
|
|
from http import HTTPStatus
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from modules.archive import lib, Archive
|
|
from wrolpi.common import get_relative_to_media_directory
|
|
from wrolpi.test.common import skip_ci, skip_macos
|
|
|
|
|
|
async def check_results(async_client, data, ids):
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
if not response.status_code == HTTPStatus.OK:
|
|
raise AssertionError(str(response.json))
|
|
|
|
if ids:
|
|
file_groups = response.json['file_groups']
|
|
assert file_groups, f'Expected {len(ids)} archives but did not receive any'
|
|
assert [i['id'] for i in response.json['file_groups']] == ids, \
|
|
f'{response.json["file_groups"][0]["id"]}..{response.json["file_groups"][-1]["id"]}'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archives_search_order(test_session, archive_directory, archive_factory, async_client):
|
|
"""Search using all orders."""
|
|
archive_factory('example.com', 'https://example.com/one', 'my archive', 'foo bar qux')
|
|
test_session.commit()
|
|
for order_by in lib.ARCHIVE_ORDERS:
|
|
data = {'search_str': 'foo', 'order_by': order_by}
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
if not response.status_code == HTTPStatus.OK:
|
|
raise AssertionError(str(response.json))
|
|
|
|
data = {'search_str': None, 'order_by': order_by}
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
if not response.status_code == HTTPStatus.OK:
|
|
raise AssertionError(str(response.json))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archives_search(test_session, archive_directory, archive_factory, async_client):
|
|
"""Archives can be searched by their title and their contents."""
|
|
# Search with no archives.
|
|
await check_results(async_client, {'search_str': 'foo'}, [])
|
|
|
|
archive_factory('example.com', 'https://example.com/one', 'my archive', 'foo bar qux')
|
|
archive_factory('example.com', 'https://example.com/one', 'other archive', 'foo baz qux qux')
|
|
archive_factory('example.org', title='archive third', contents='baz qux qux qux')
|
|
archive_factory('example.org') # has no contents
|
|
test_session.commit()
|
|
|
|
# Contents are d_text, so a deep search is required to match them.
|
|
# 1 and 2 contain "foo".
|
|
data = {'search_str': 'foo', 'deep': True}
|
|
await check_results(async_client, data, [2, 1])
|
|
|
|
# The default (fast) search does not match contents.
|
|
data = {'search_str': 'foo'}
|
|
await check_results(async_client, data, [])
|
|
|
|
# 2 and 3 contain "baz".
|
|
data = {'search_str': 'baz', 'deep': True}
|
|
await check_results(async_client, data, [3, 2])
|
|
|
|
# 1 contains "bar".
|
|
data = {'search_str': 'bar', 'deep': True}
|
|
await check_results(async_client, data, [1, ])
|
|
|
|
# No archives contain "huzzah"
|
|
data = {'search_str': 'huzzah', 'deep': True}
|
|
await check_results(async_client, data, [])
|
|
|
|
# Only 3 contains "baz" and is in domain "example.org"
|
|
data = {'search_str': 'baz', 'domain': 'example.org', 'deep': True}
|
|
await check_results(async_client, data, [3, ])
|
|
|
|
# 1's title contains "my". Postgres ignored this stopword; FTS5 has no stopwords, so it matches.
|
|
data = {'search_str': 'my', 'deep': True}
|
|
await check_results(async_client, data, [1, ])
|
|
|
|
# 3's title contains "third", found by the default (fast) search and the deep search.
|
|
data = {'search_str': 'third'}
|
|
await check_results(async_client, data, [3, ])
|
|
data = {'search_str': 'third', 'deep': True}
|
|
await check_results(async_client, data, [3, ])
|
|
|
|
# All contents contain "qux", but they contain different amounts. They are ordered by the amount.
|
|
data = {'search_str': 'qux', 'deep': True}
|
|
await check_results(async_client, data, [3, 2, 1])
|
|
|
|
data = {'search_str': 'qux', 'order_by': 'bad order_by'}
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_archive_tags(test_session, async_client, archive_factory, tag_factory):
|
|
"""Tagged Archives can be searched."""
|
|
tag = await tag_factory()
|
|
archive_factory(domain='example.com', tag_names=[tag.name, ])
|
|
test_session.commit()
|
|
|
|
content = {'search_str': '', 'domain': 'example.com', 'tag_names': [tag.name, ]}
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(content))
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json['file_groups']
|
|
assert response.json['totals']['file_groups'] == 1
|
|
|
|
content = {'search_str': '', 'domain': 'example.com', 'tag_names': ['does not exist', ]}
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(content))
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json['file_groups'] == []
|
|
assert response.json['totals']['file_groups'] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archives_search_headline(test_session, archive_directory, archive_factory, async_client):
|
|
"""Headlines can be requested."""
|
|
archive_factory('example.com', 'https://example.com/one', 'my archive', 'foo bar qux')
|
|
archive_factory('example.com', 'https://example.com/one', 'other archive', 'foo baz qux qux')
|
|
archive_factory('example.org', title='archive third', contents='baz qux qux qux')
|
|
archive_factory('example.org') # has no contents
|
|
test_session.commit()
|
|
|
|
# "foo" is only in the contents (d_text), so a deep search is required to match it.
|
|
content = dict(search_str='foo', deep=True)
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(content))
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
# Headlines are only fetched if requested.
|
|
assert response.json['file_groups'][0]['d_headline'] is None
|
|
assert response.json['file_groups'][1]['d_headline'] is None
|
|
|
|
content = dict(search_str='foo', headline=True, deep=True)
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(content))
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
# FTS headlines use <b>...</b> to highlight matching words.
|
|
assert response.json['file_groups'][0]['d_headline'] == '<b>foo</b> baz qux qux'
|
|
assert response.json['file_groups'][1]['d_headline'] == '<b>foo</b> bar qux'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_offset(test_session, archive_factory, async_client):
|
|
"""Archive search can be offset."""
|
|
for i in range(500):
|
|
archive_factory('example.com', f'https://example.com/{i}', contents='foo bar')
|
|
test_session.commit()
|
|
|
|
data = {'search_str': None, 'offset': 0}
|
|
await check_results(async_client, data, list(range(500, 480, -1)))
|
|
data = {'search_str': None, 'offset': 20}
|
|
await check_results(async_client, data, list(range(480, 460, -1)))
|
|
data = {'search_str': None, 'offset': 100}
|
|
await check_results(async_client, data, list(range(400, 380, -1)))
|
|
data = {'search_str': None, 'offset': 200}
|
|
await check_results(async_client, data, list(range(300, 280, -1)))
|
|
data = {'search_str': None, 'offset': 500}
|
|
await check_results(async_client, data, [])
|
|
# Past the last page still reports the true total (a separate COUNT, not COUNT(*) OVER()).
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json['totals']['file_groups'] == 500
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archives_search_no_query(test_session, archive_factory, async_client):
|
|
"""Archive Search API endpoint does not require data in the body."""
|
|
# Add 100 random archives.
|
|
for _ in range(100):
|
|
archive_factory('example.com')
|
|
test_session.commit()
|
|
|
|
# All archives are returned when no `search_str` is passed.
|
|
request, response = await async_client.post('/api/archive/search', content='{}')
|
|
assert response.status_code == HTTPStatus.OK, response.json
|
|
assert [i['id'] for i in response.json['file_groups']] == list(range(100, 80, -1))
|
|
assert response.json['totals']['file_groups'] == 100
|
|
|
|
# All archives are from "example.com".
|
|
data = dict(domain='example.com')
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.OK, response.json
|
|
assert [i['id'] for i in response.json['file_groups']] == list(range(100, 80, -1))
|
|
assert response.json['totals']['file_groups'] == 100
|
|
|
|
# No archives are from "example.org".
|
|
data = dict(domain='example.org')
|
|
request, response = await async_client.post('/api/archive/search', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.OK, response.json
|
|
assert [i['id'] for i in response.json['file_groups']] == []
|
|
assert response.json['totals']['file_groups'] == 0
|
|
|
|
|
|
@skip_macos
|
|
@skip_ci
|
|
@pytest.mark.asyncio
|
|
async def test_archive_upload(test_session, async_client, singlefile_contents_factory, make_multipart_form,
|
|
await_switches, events_fixture):
|
|
"""Test converting a SingleFile from the SingleFile browser extension to an Archive."""
|
|
singlefile_contents = singlefile_contents_factory(title='upload title', url='https://example.com/single-file-url')
|
|
forms = [
|
|
dict(name='url', value='https://example.com/form-url'),
|
|
dict(name='singlefile_contents', value=singlefile_contents, filename='singlefile_contents')
|
|
]
|
|
body = make_multipart_form(forms)
|
|
headers = {'Content-Type': 'multipart/form-data; boundary=-----------------------------sanic'}
|
|
request, response = await async_client.post('/api/archive/upload', content=body, headers=headers)
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
archive, = test_session.query(Archive).all()
|
|
assert archive.singlefile_path.is_file(), 'Singlefile should have been saved.'
|
|
assert archive.readability_path.is_file(), 'Readability should be extracted.'
|
|
assert archive.readability_json_path.is_file(), 'Readability json should be extracted.'
|
|
assert archive.screenshot_path.is_file(), 'Screenshot should have been created'
|
|
assert archive.file_group.title == 'upload title', 'Title was not correct'
|
|
assert archive.file_group.url == 'https://example.com/single-file-url', 'SingleFile URL should be trusted'
|
|
events_fixture.assert_has_event('upload_archive')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_upload_with_tags(test_session, async_client, singlefile_contents_factory, make_multipart_form,
|
|
await_switches, tag_factory, archive_factory):
|
|
"""Uploading with tagNames applies existing tags and auto-creates new ones.
|
|
|
|
This test mocks `singlefile_to_archive` so it can run on every platform
|
|
(the full archive pipeline depends on a readability/screenshot service
|
|
that is not available on macOS). The mocked Archive stands in for what
|
|
the real pipeline would have produced; the assertion verifies that the
|
|
switch handler applies tags to whichever FileGroup the pipeline returns.
|
|
"""
|
|
existing = await tag_factory('preexisting')
|
|
archive = archive_factory('example.com', 'https://example.com/tagged', 'tagged')
|
|
test_session.commit()
|
|
|
|
async def fake_singlefile_to_archive(singlefile, destination=None):
|
|
return archive
|
|
|
|
singlefile_contents = singlefile_contents_factory(title='tagged', url='https://example.com/tagged')
|
|
forms = [
|
|
dict(name='url', value='https://example.com/tagged'),
|
|
dict(name='singlefile_contents', value=singlefile_contents, filename='singlefile_contents'),
|
|
dict(name='tagNames', value=existing.name),
|
|
dict(name='tagNames', value='brand-new-tag'),
|
|
]
|
|
body = make_multipart_form(forms)
|
|
headers = {'Content-Type': 'multipart/form-data; boundary=-----------------------------sanic'}
|
|
|
|
with patch('modules.archive.lib.singlefile_to_archive', side_effect=fake_singlefile_to_archive):
|
|
request, response = await async_client.post('/api/archive/upload', content=body, headers=headers)
|
|
assert response.status_code == HTTPStatus.OK
|
|
await await_switches()
|
|
|
|
test_session.expire_all()
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
assert sorted(archive.file_group.tag_names) == ['brand-new-tag', 'preexisting']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_upload_without_tags_unchanged(test_session, async_client, singlefile_contents_factory,
|
|
make_multipart_form, await_switches, archive_factory):
|
|
"""Uploads from clients that don't send tagNames (e.g. upstream SingleFile) still work."""
|
|
archive = archive_factory('example.com', 'https://example.com/no-tags', 'no-tags')
|
|
test_session.commit()
|
|
|
|
async def fake_singlefile_to_archive(singlefile, destination=None):
|
|
return archive
|
|
|
|
singlefile_contents = singlefile_contents_factory(title='no-tags', url='https://example.com/no-tags')
|
|
forms = [
|
|
dict(name='url', value='https://example.com/no-tags'),
|
|
dict(name='singlefile_contents', value=singlefile_contents, filename='singlefile_contents'),
|
|
]
|
|
body = make_multipart_form(forms)
|
|
headers = {'Content-Type': 'multipart/form-data; boundary=-----------------------------sanic'}
|
|
|
|
with patch('modules.archive.lib.singlefile_to_archive', side_effect=fake_singlefile_to_archive):
|
|
request, response = await async_client.post('/api/archive/upload', content=body, headers=headers)
|
|
assert response.status_code == HTTPStatus.OK
|
|
await await_switches()
|
|
|
|
test_session.expire_all()
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
assert archive.file_group.tag_names == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_upload_file_tracking(test_session, async_client, archive_directory, archive_factory,
|
|
await_switches, make_multipart_form, image_bytes_factory):
|
|
"""Test that uploading an info.json and image file via /api/files/upload properly tracks them in Archive and FileGroup."""
|
|
archive = archive_factory('example.com', 'https://example.com/test-url', 'tracking test', screenshot=False)
|
|
test_session.commit()
|
|
|
|
# Verify Archive and FileGroup are created, but screenshot does not exist.
|
|
file_group = archive.file_group
|
|
assert not archive.screenshot_path, 'Screenshot should not exist yet.'
|
|
assert file_group.files is not None, 'FileGroup should have files array'
|
|
initial_file_count = len(file_group.files)
|
|
|
|
# Upload info.json via /api/files/upload
|
|
# Extract the prefix from singlefile path (e.g., "2000-01-01-00-00-01_tracking test")
|
|
singlefile_stem = archive.singlefile_path.stem # e.g., "2000-01-01-00-00-01_tracking test"
|
|
|
|
# Create info.json content
|
|
info_json_data = {
|
|
'title': 'test',
|
|
'url': 'https://example.com/test-url',
|
|
'custom_field': 'value'
|
|
}
|
|
info_json_bytes = json.dumps(info_json_data).encode()
|
|
|
|
# Get relative destination directory
|
|
destination = str(get_relative_to_media_directory(archive.singlefile_path.parent))
|
|
filename = f'{singlefile_stem}.info.json'
|
|
|
|
# POST to /api/files/upload
|
|
forms = [
|
|
dict(name='destination', value=destination),
|
|
dict(name='filename', value=filename),
|
|
dict(name='chunkNumber', value='0'),
|
|
dict(name='totalChunks', value='0'),
|
|
dict(name='chunkSize', value=str(len(info_json_bytes))),
|
|
dict(name='overwrite', value='false'),
|
|
dict(name='chunk', value=info_json_bytes, filename='chunk'),
|
|
]
|
|
body = make_multipart_form(forms)
|
|
headers = {'Content-Type': 'multipart/form-data; boundary=-----------------------------sanic'}
|
|
request, response = await async_client.post('/api/files/upload', content=body, headers=headers)
|
|
assert response.status_code == HTTPStatus.CREATED, f'Upload failed with status {response.status_code}: {response.json}'
|
|
|
|
# Step 4: Verify info.json is tracked
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
|
|
# Assert info.json properties exist
|
|
assert archive.info_json_path is not None, 'Archive.info_json_path should exist'
|
|
assert archive.info_json_file is not None, 'Archive.info_json_file should exist'
|
|
|
|
# Assert FileGroup has the correct number of files.
|
|
assert len(archive.file_group.files) == initial_file_count + 1, \
|
|
f'FileGroup should have {initial_file_count + 1} files after adding info.json'
|
|
assert archive.file_group.data.get('info_json_path'), \
|
|
'Archive readability info json should be in FileGroup.data'
|
|
|
|
# Upload an image file and verify it's added to the FileGroup.
|
|
# Store file count before adding image
|
|
file_count_before_image = len(archive.file_group.files)
|
|
|
|
# Create image using factory
|
|
image_data = image_bytes_factory()
|
|
|
|
# Upload image with same stem as the Archive (using .jpg to avoid conflict with factory-created .png)
|
|
image_filename = f'{singlefile_stem}.jpg'
|
|
forms = [
|
|
dict(name='destination', value=destination),
|
|
dict(name='filename', value=image_filename),
|
|
dict(name='chunkNumber', value='0'),
|
|
dict(name='totalChunks', value='0'),
|
|
dict(name='chunkSize', value=str(len(image_data))),
|
|
dict(name='overwrite', value='false'),
|
|
dict(name='chunk', value=image_data, filename='chunk'),
|
|
]
|
|
body = make_multipart_form(forms)
|
|
headers = {'Content-Type': 'multipart/form-data; boundary=-----------------------------sanic'}
|
|
request, response = await async_client.post('/api/files/upload', content=body, headers=headers)
|
|
assert response.status_code == HTTPStatus.CREATED, f'Image upload failed with status {response.status_code}: {response.json}'
|
|
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
|
|
# Assert screenshot properties exist
|
|
assert archive.screenshot_path is not None, 'Archive.screenshot_path should exist'
|
|
assert archive.screenshot_file is not None, 'Archive.screenshot_file should exist'
|
|
# data['screenshot_path'] stores just the filename, screenshot_path resolves to absolute
|
|
assert archive.file_group.data.get(
|
|
'screenshot_path') == archive.screenshot_path.name, 'Archive screenshot should be in FileGroup.data'
|
|
|
|
# Assert image is in FileGroup.files (use my_files() to get resolved absolute paths)
|
|
file_paths = [f['path'] for f in archive.file_group.my_files()]
|
|
assert archive.screenshot_path in file_paths, 'Screenshot should be in FileGroup.files'
|
|
|
|
# Assert FileGroup has correct number of files
|
|
assert len(archive.file_group.files) == file_count_before_image + 1, \
|
|
f'FileGroup should have {file_count_before_image + 1} files after adding image'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_generate_screenshot(test_session, async_client, archive_factory, await_switches,
|
|
image_bytes_factory, wrol_mode_fixture):
|
|
"""Test generating a screenshot for an Archive that doesn't have one."""
|
|
# Mock html_screenshot to avoid Selenium dependency
|
|
mock_screenshot_bytes = image_bytes_factory()
|
|
|
|
with patch('modules.archive.lib.html_screenshot', return_value=mock_screenshot_bytes):
|
|
# Test success case: Archive without screenshot
|
|
archive = archive_factory('example.com', 'https://example.com/test', 'Test Archive', screenshot=False)
|
|
test_session.commit()
|
|
|
|
# Verify archive has no screenshot
|
|
assert archive.screenshot_path is None, 'Archive should not have a screenshot yet'
|
|
assert archive.singlefile_path is not None, 'Archive should have a singlefile'
|
|
|
|
# Request screenshot generation
|
|
request, response = await async_client.post(f'/api/archive/{archive.file_group_id}/generate_screenshot')
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json['message'] == 'Screenshot generation queued'
|
|
|
|
# Verify screenshot was generated
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
assert archive.screenshot_path is not None, 'Screenshot should have been generated'
|
|
assert archive.screenshot_path.is_file(), 'Screenshot file should exist'
|
|
assert archive.screenshot_file is not None, 'Screenshot file should be tracked'
|
|
|
|
# Test error case: Archive not found
|
|
request, response = await async_client.post('/api/archive/99999/generate_screenshot')
|
|
assert response.status_code == HTTPStatus.NOT_FOUND
|
|
assert 'not found' in response.json['error'].lower()
|
|
|
|
# Test success case: Archive already has screenshot - should return OK and ensure tracking
|
|
original_screenshot_path = archive.screenshot_path
|
|
assert original_screenshot_path is not None, 'Archive should have a screenshot'
|
|
assert original_screenshot_path.is_file(), 'Screenshot file should exist'
|
|
|
|
# Note: Previously this test simulated a bug where data['screenshot_path'] could be missing
|
|
# while the file existed. This scenario is now auto-fixed by Archive.validate() which calls
|
|
# rebuild_data(), so the simulation is no longer necessary.
|
|
|
|
request, response = await async_client.post(f'/api/archive/{archive.file_group_id}/generate_screenshot')
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json['message'] == 'Screenshot generation queued'
|
|
|
|
# Verify screenshot is still there and properly tracked
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
assert archive.screenshot_path == original_screenshot_path, 'Screenshot path should be unchanged'
|
|
assert archive.screenshot_path.is_file(), 'Screenshot file should still exist'
|
|
assert archive.screenshot_file is not None, 'Screenshot file should still be tracked'
|
|
# IMPORTANT: Also verify FileGroup.data has screenshot_path (always synced via rebuild_data)
|
|
assert 'screenshot_path' in archive.file_group.data, \
|
|
'screenshot_path should be in FileGroup.data'
|
|
assert archive.file_group.data['screenshot_path'] == original_screenshot_path.name, \
|
|
'screenshot_path in data should be filename only'
|
|
|
|
# Test error case: Archive has no singlefile
|
|
archive_no_singlefile = archive_factory('example.com', 'https://example.com/no-singlefile', screenshot=False)
|
|
test_session.commit()
|
|
# Delete the singlefile to simulate missing file
|
|
if archive_no_singlefile.singlefile_path and archive_no_singlefile.singlefile_path.is_file():
|
|
archive_no_singlefile.singlefile_path.unlink()
|
|
|
|
request, response = await async_client.post(f'/api/archive/{archive_no_singlefile.file_group_id}/generate_screenshot')
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
assert 'no singlefile' in response.json['error'].lower()
|
|
|
|
# Test WROL mode case
|
|
await wrol_mode_fixture(True)
|
|
archive_wrol = archive_factory('example.com', 'https://example.com/wrol-test', screenshot=False)
|
|
test_session.commit()
|
|
|
|
request, response = await async_client.post(f'/api/archive/{archive_wrol.file_group_id}/generate_screenshot')
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
await wrol_mode_fixture(False)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_file_format_preview(async_client):
|
|
"""Test the archive file format preview endpoint."""
|
|
# Test valid format with download_datetime (default format)
|
|
data = {'archive_file_format': '%(download_datetime)s_%(title)s.%(ext)s'}
|
|
request, response = await async_client.post('/api/archive/file_format', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert 'preview' in response.json
|
|
assert response.json['preview'].endswith('.html')
|
|
assert 'Example Page Title' in response.json['preview']
|
|
# Verify datetime format YYYY-MM-DD-HH-MM-SS
|
|
import re
|
|
assert re.search(r'\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\d{2}', response.json['preview'])
|
|
|
|
# Test valid format with download_date (date only)
|
|
data = {'archive_file_format': '%(download_date)s_%(title)s.%(ext)s'}
|
|
request, response = await async_client.post('/api/archive/file_format', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert re.search(r'\d{4}-\d{2}-\d{2}_', response.json['preview'])
|
|
|
|
# Test format with year subdirectory
|
|
data = {'archive_file_format': '%(download_year)s/%(download_datetime)s_%(title)s.%(ext)s'}
|
|
request, response = await async_client.post('/api/archive/file_format', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert '/' in response.json['preview'] # Contains subdirectory
|
|
|
|
# Test invalid format - missing .%(ext)s
|
|
data = {'archive_file_format': '%(download_date)s_%(title)s'}
|
|
request, response = await async_client.post('/api/archive/file_format', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
assert 'error' in response.json
|
|
|
|
# Test invalid format - unknown variable
|
|
data = {'archive_file_format': '%(invalid_var)s_%(title)s.%(ext)s'}
|
|
request, response = await async_client.post('/api/archive/file_format', content=json.dumps(data))
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
assert 'error' in response.json
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_browsers_endpoint(async_client):
|
|
"""Test the browsers endpoint returns installed browsers."""
|
|
# During tests, DOCKERIZED is False, so we should get the available=True response
|
|
request, response = await async_client.get('/api/archive/browsers')
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert 'browsers' in response.json
|
|
assert 'available' in response.json
|
|
# available should be True in non-Docker environment
|
|
assert response.json['available'] is True
|
|
# browsers is a list
|
|
assert isinstance(response.json['browsers'], list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_browsers_endpoint_dockerized(async_client):
|
|
"""Test the browsers endpoint returns not available when dockerized."""
|
|
with patch('modules.archive.api.DOCKERIZED', True):
|
|
request, response = await async_client.get('/api/archive/browsers')
|
|
assert response.status_code == HTTPStatus.OK
|
|
assert response.json['browsers'] == []
|
|
assert response.json['available'] is False
|
|
assert 'message' in response.json
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_archive_untagged(async_client, test_session, archive_factory):
|
|
"""An untagged Archive can be deleted via the unified delete endpoint without force."""
|
|
archive = archive_factory('example.com', 'https://example.com/1')
|
|
test_session.commit()
|
|
paths = list(archive.my_paths())
|
|
assert paths and all(p.is_file() for p in paths)
|
|
|
|
body = json.dumps({'file_group_ids': [archive.file_group_id]})
|
|
request, response = await async_client.post('/api/files/delete_groups', content=body)
|
|
assert response.status_code == HTTPStatus.NO_CONTENT
|
|
assert all(not p.exists() for p in paths)
|
|
assert test_session.query(Archive).count() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_archive_tagged_requires_force(async_client, test_session, archive_factory, tag_factory,
|
|
await_switches):
|
|
"""A tagged Archive returns 409 with FILE_GROUP_IS_TAGGED until force=true is set."""
|
|
tag = await tag_factory()
|
|
archive = archive_factory('example.com', 'https://example.com/1', tag_names=[tag.name])
|
|
test_session.commit()
|
|
paths = list(archive.my_paths())
|
|
|
|
body = json.dumps({'file_group_ids': [archive.file_group_id]})
|
|
request, response = await async_client.post('/api/files/delete_groups', content=body)
|
|
assert response.status_code == HTTPStatus.CONFLICT
|
|
assert response.json['code'] == 'FILE_GROUP_IS_TAGGED'
|
|
assert len(response.json['file_groups']) == 1
|
|
returned = response.json['file_groups'][0]
|
|
assert returned['id'] == archive.file_group_id
|
|
assert tag.name in returned['tags']
|
|
# Nothing was deleted.
|
|
assert all(p.is_file() for p in paths)
|
|
assert test_session.query(Archive).count() == 1
|
|
|
|
# With force=true, the Archive is deleted.
|
|
body = json.dumps({'file_group_ids': [archive.file_group_id], 'force': True})
|
|
request, response = await async_client.post('/api/files/delete_groups', content=body)
|
|
assert response.status_code == HTTPStatus.NO_CONTENT
|
|
assert all(not p.exists() for p in paths)
|
|
assert test_session.query(Archive).count() == 0
|
|
await await_switches()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_archive_atomic_when_some_tagged(async_client, test_session, archive_factory, tag_factory):
|
|
"""When deleting multiple Archives and any are tagged, NONE are deleted until force=true."""
|
|
tag = await tag_factory()
|
|
untagged = archive_factory('example.com', 'https://example.com/1')
|
|
tagged = archive_factory('example.com', 'https://example.com/2', tag_names=[tag.name])
|
|
test_session.commit()
|
|
untagged_paths = list(untagged.my_paths())
|
|
tagged_paths = list(tagged.my_paths())
|
|
|
|
body = json.dumps({'file_group_ids': [untagged.file_group_id, tagged.file_group_id]})
|
|
request, response = await async_client.post('/api/files/delete_groups', content=body)
|
|
assert response.status_code == HTTPStatus.CONFLICT
|
|
assert response.json['code'] == 'FILE_GROUP_IS_TAGGED'
|
|
# Only the tagged FileGroup is in the response.
|
|
returned_ids = {fg['id'] for fg in response.json['file_groups']}
|
|
assert returned_ids == {tagged.file_group_id}
|
|
# Nothing was deleted.
|
|
assert all(p.is_file() for p in untagged_paths)
|
|
assert all(p.is_file() for p in tagged_paths)
|
|
assert test_session.query(Archive).count() == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_archive_loads_history_eagerly(test_session, archive_factory, async_client, tag_factory):
|
|
"""An Archive's page returns its history, with each Archive's Tags, without a query per history Archive."""
|
|
from sqlalchemy import event
|
|
from sqlalchemy.engine import Engine
|
|
|
|
tag = await tag_factory()
|
|
short = [archive_factory('example.com', 'https://example.com/short', tag_names=[tag.name]) for _ in range(2)]
|
|
long = [archive_factory('example.com', 'https://example.com/long', tag_names=[tag.name]) for _ in range(6)]
|
|
test_session.commit()
|
|
|
|
async def get(archive):
|
|
url = f'/api/archive/{archive.file_group_id}?skip_viewed=true'
|
|
test_session.expire_all() # Each request starts from the same Session state.
|
|
statements = []
|
|
|
|
def count(conn, cursor, statement, *_):
|
|
statements.append(statement)
|
|
|
|
event.listen(Engine, 'before_cursor_execute', count)
|
|
try:
|
|
request, response = await async_client.get(url)
|
|
finally:
|
|
event.remove(Engine, 'before_cursor_execute', count)
|
|
assert response.status_code == HTTPStatus.OK, response.json
|
|
return response.json, len(statements)
|
|
|
|
await get(short[-1]) # The first request does one-time work; measure the next ones.
|
|
short_json, short_statements = await get(short[-1])
|
|
long_json, long_statements = await get(long[-1])
|
|
|
|
assert len(short_json['history']) == 1
|
|
assert len(long_json['history']) == 5
|
|
assert long_json['file_group']['tags'] == [tag.name]
|
|
assert all(i['tags'] == [tag.name] for i in long_json['history'])
|
|
# Five history Archives cost no more queries than one.
|
|
assert long_statements == short_statements, (short_statements, long_statements)
|