mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 02:48:38 -04:00
221 lines
9 KiB
Python
221 lines
9 KiB
Python
import json
|
|
import pathlib
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from modules.archive import Archive
|
|
from wrolpi.common import get_media_directory, get_wrolpi_config
|
|
from wrolpi.errors import UnknownArchive
|
|
from wrolpi.files.models import FileGroup
|
|
|
|
|
|
def test_archive_get_by_id(test_session, archive_factory):
|
|
"""Archive.get_by_id should accept session as first argument, id as second."""
|
|
archive = archive_factory()
|
|
test_session.commit()
|
|
|
|
# Session must be first argument (session-first pattern)
|
|
found = Archive.get_by_id(test_session, archive.id)
|
|
assert found is not None
|
|
assert found.id == archive.id
|
|
|
|
# Non-existent ID should return None
|
|
not_found = Archive.get_by_id(test_session, 99999)
|
|
assert not_found is None
|
|
|
|
|
|
def test_archive_find_by_id(test_session, archive_factory):
|
|
"""Archive.find_by_id should accept session as first argument, id as second."""
|
|
archive = archive_factory()
|
|
test_session.commit()
|
|
|
|
# Session must be first argument (session-first pattern)
|
|
found = Archive.find_by_id(test_session, archive.id)
|
|
assert found is not None
|
|
assert found.id == archive.id
|
|
|
|
# Non-existent ID should raise UnknownArchive
|
|
with pytest.raises(UnknownArchive):
|
|
Archive.find_by_id(test_session, 99999)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_download_destination(async_client, test_session, test_directory, archive_factory, fake_now):
|
|
fake_now(datetime(2000, 1, 2))
|
|
|
|
wrolpi_config = get_wrolpi_config()
|
|
|
|
archive = archive_factory(domain='wrolpi.org')
|
|
test_session.commit()
|
|
|
|
# Test the default download directory.
|
|
assert str(archive.download_directory) == str(test_directory / 'archive/wrolpi.org')
|
|
|
|
# Year of download is supported.
|
|
wrolpi_config.archive_destination = 'archives/%(domain_tag)s/%(domain)s/%(year)s'
|
|
assert str(archive.download_directory) == str(test_directory / 'archives/wrolpi.org/2000')
|
|
|
|
# More download date is supported.
|
|
wrolpi_config.archive_destination = 'archive/%(domain_tag)s/%(domain)s/%(year)s/%(month)s/%(day)s'
|
|
assert str(archive.download_directory) == str(test_directory / 'archive/wrolpi.org/2000/1/2')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_screenshot(async_client, test_session, archive_factory, image_bytes_factory):
|
|
"""Test Archive.set_screenshot() method with various scenarios."""
|
|
|
|
# ========== Test 1: Successfully set screenshot on archive without one ==========
|
|
archive = archive_factory('example.com', 'https://example.com/test', 'Test Archive', screenshot=False)
|
|
test_session.commit()
|
|
|
|
# Verify no screenshot initially
|
|
assert archive.screenshot_path is None
|
|
|
|
# Create a screenshot file
|
|
screenshot_path = archive.singlefile_path.parent / 'test_screenshot.png'
|
|
screenshot_path.write_bytes(image_bytes_factory())
|
|
|
|
# Set the screenshot
|
|
archive.set_screenshot(screenshot_path)
|
|
|
|
# Verify data was set before commit (stores filename only, not absolute path)
|
|
assert archive.file_group.data['screenshot_path'] == screenshot_path.name, \
|
|
f"Data not set correctly before commit: {archive.file_group.data.get('screenshot_path')}"
|
|
|
|
test_session.commit()
|
|
|
|
# Refresh the archive to get latest data from DB
|
|
test_session.expire_all()
|
|
archive = test_session.query(Archive).filter_by(id=archive.id).one()
|
|
|
|
# Verify screenshot was set (screenshot_path property resolves to absolute via my_files())
|
|
assert archive.screenshot_path == screenshot_path
|
|
assert archive.screenshot_file is not None
|
|
# Verify FileGroup.data was persisted to DB (now stores filename only)
|
|
assert 'screenshot_path' in archive.file_group.data
|
|
assert archive.file_group.data['screenshot_path'] == screenshot_path.name
|
|
|
|
# Verify file is tracked in FileGroup.files
|
|
screenshot_files = archive.file_group.my_files('image/')
|
|
assert len(screenshot_files) > 0
|
|
assert screenshot_files[0]['path'] == screenshot_path
|
|
|
|
# ========== Test 2: Error when trying to set screenshot on archive that already has one ==========
|
|
original_screenshot = archive.screenshot_path
|
|
assert original_screenshot is not None
|
|
|
|
# Try to set another screenshot
|
|
new_screenshot_path = archive.singlefile_path.parent / 'another_screenshot.png'
|
|
new_screenshot_path.write_bytes(b'fake image data')
|
|
|
|
# Should raise ValueError
|
|
with pytest.raises(ValueError, match='already has a screenshot'):
|
|
archive.set_screenshot(new_screenshot_path)
|
|
|
|
# Verify original screenshot is unchanged
|
|
assert archive.screenshot_path == original_screenshot
|
|
|
|
# ========== Test 3: Error when trying to set non-existent screenshot file ==========
|
|
archive_no_screenshot = archive_factory('example.com', 'https://example.com/test2', 'Test Archive 2',
|
|
screenshot=False)
|
|
test_session.commit()
|
|
|
|
# Try to set a screenshot that doesn't exist
|
|
nonexistent_path = pathlib.Path('/tmp/does_not_exist.png')
|
|
|
|
# Should raise ValueError
|
|
with pytest.raises(ValueError, match='does not exist'):
|
|
archive_no_screenshot.set_screenshot(nonexistent_path)
|
|
|
|
# Verify no screenshot was set
|
|
assert archive_no_screenshot.screenshot_path is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_validate_rebuilds_data_after_filename_change(async_client, test_session, test_directory,
|
|
archive_factory):
|
|
"""Archive.validate() should rebuild data field when filenames change.
|
|
|
|
This tests the bug where reorganization updates files[] but not data{}.
|
|
The data field contains cached path names that must stay in sync.
|
|
|
|
The bug scenario: files are renamed during reorganization (e.g., fixing separator
|
|
from '___' to '_'), files[] is updated but data{} is not. The UI then uses
|
|
stale data.singlefile_path and gets 404 errors.
|
|
"""
|
|
# Create an archive using the factory
|
|
archive = archive_factory(domain='test.com', url='https://test.com/article')
|
|
test_session.commit()
|
|
|
|
# Verify initial data field has singlefile_path
|
|
assert archive.file_group.data is not None
|
|
original_singlefile = archive.file_group.data.get('singlefile_path')
|
|
assert original_singlefile is not None
|
|
|
|
# Simulate what reorganization does: update files[] but not data{}
|
|
# (This mimics _bulk_update_file_groups_reorganize behavior)
|
|
# Use a realistic rename: change the title part of the filename
|
|
# Original: '2000-01-01-07-00-01_NA.html' -> '2000-01-01-07-00-01_New-Title.html'
|
|
new_filename = original_singlefile.replace('_NA.html', '_New-Title.html')
|
|
old_files = archive.file_group.files
|
|
new_files = []
|
|
for f in old_files:
|
|
new_f = dict(f)
|
|
if f['path'].endswith('.html') and not f['path'].endswith('.readability.html'):
|
|
new_f['path'] = new_filename
|
|
new_files.append(new_f)
|
|
archive.file_group.files = new_files
|
|
archive.file_group.primary_path = archive.file_group.directory / new_filename
|
|
|
|
# Rename the actual file on disk to match the new filename
|
|
old_path = archive.file_group.directory / original_singlefile
|
|
new_path = archive.file_group.directory / new_filename
|
|
old_path.rename(new_path)
|
|
|
|
# data field still has OLD filename (this is the bug)
|
|
assert archive.file_group.data['singlefile_path'] == original_singlefile
|
|
|
|
# After validate(), data field should have NEW filename
|
|
archive.validate()
|
|
|
|
assert archive.file_group.data['singlefile_path'] == new_filename
|
|
|
|
|
|
def test_update_wrolpi_json_prefers_info_json(test_session, make_files_structure):
|
|
"""FileGroup.update_wrolpi_json prefers .info.json over .readability.json when both exist."""
|
|
make_files_structure({'page.html': '<html>test</html>'})
|
|
fg = FileGroup.from_paths(test_session, get_media_directory() / 'page.html')
|
|
test_session.add(fg)
|
|
test_session.flush()
|
|
|
|
fg.replace_info_json({'title': 'Info Title'})
|
|
fg.replace_readability_json({'title': 'Readability Title'})
|
|
|
|
fg.update_wrolpi_json({'key': 'value'})
|
|
|
|
# wrolpi section should be in .info.json, not .readability.json.
|
|
info_json = json.loads((get_media_directory() / 'page.info.json').read_text())
|
|
assert info_json['wrolpi'] == {'key': 'value'}
|
|
|
|
readability_json = json.loads((get_media_directory() / 'page.readability.json').read_text())
|
|
assert 'wrolpi' not in readability_json
|
|
|
|
|
|
def test_metadata_json_path(test_session, make_files_structure):
|
|
"""FileGroup.metadata_json_path returns .info.json first, then .readability.json."""
|
|
make_files_structure({'page.html': '<html>test</html>'})
|
|
fg = FileGroup.from_paths(test_session, get_media_directory() / 'page.html')
|
|
test_session.add(fg)
|
|
test_session.flush()
|
|
|
|
# No metadata JSON yet.
|
|
assert fg.metadata_json_path is None
|
|
|
|
# Add readability JSON.
|
|
fg.replace_readability_json({'title': 'Test'})
|
|
assert fg.metadata_json_path == get_media_directory() / 'page.readability.json'
|
|
|
|
# Add info JSON - it should now be preferred.
|
|
fg.replace_info_json({'title': 'Test'})
|
|
assert fg.metadata_json_path == get_media_directory() / 'page.info.json'
|