wrolpi/modules/archive/test/test_collection_id_bug.py
lrnselfreliance dab333124e
Upgrade to SQLAlchemy 2.0 (synchronous) (#803)
SQLAlchemy 1.3 has been end-of-life since 2021, and alembic was held at 1.10
because newer releases need SQLAlchemy 2.  This is a mechanical port; the code
stays synchronous.

- SQLAlchemy[asyncio]==2.0.54, alembic==1.20.0.
- Model annotations use Mapped[...] (2.0 rejects bare annotations on mapped
  attributes); declarative_base moves to sqlalchemy.orm.
- Plain SQL strings go through text() or exec_driver_sql; schema_ddl.execute_sql
  accepts a SQLAlchemy Connection or a raw sqlite3 connection, so rebuild_fts
  works from the baseline migration again.
- Session.get replaces Query.get; session.in_transaction() replaces
  session.transaction; cache_ok on MediaPathType and FancyJSON.
- The tag-directory sync query joins through TagFile's relationships instead of
  the legacy two-entity outerjoin.
- Section-hint rows use attribute access (2.0 rows are tuples).
- Migrations run on an AUTOCOMMIT connection, matching 1.3, where each
  statement committed on its own.  2.0 autobegins, so the unique-stem
  migration's read snapshot could not become a writer after its helpers wrote
  on another connection ("database is locked").  The begin listener emits no
  BEGIN for AUTOCOMMIT connections.
- Newer alembic compares column types: WrolpiKV now describes the TEXT table
  its migration created.
- The test database installs its DDL inside engine.begin(); 2.0 connections
  no longer autocommit, so the FTS tables were rolled back.
- Dependabot now holds SQLAlchemy at 2.0 instead of 1.3.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 15:08:40 -06:00

202 lines
7.5 KiB
Python

"""Test for collection_id bug when modeling archives from discovered files.
This test reproduces the bug where Archives created by model_archive()
don't have collection_id set, causing a NOT NULL constraint violation.
The bug happens because:
1. The archive_factory sets collection_id explicitly (masking the bug)
2. But model_archive() (used in production) doesn't set collection_id
3. This causes NULL constraint violations when indexing real files
"""
import pathlib
import pytest
from modules.archive import model_archive
from modules.archive.lib import archive_strftime
from wrolpi.collections import Collection
@pytest.mark.asyncio
async def test_model_archive_sets_collection_id(async_client, test_session, test_directory, make_files_structure):
"""
Test that model_archive() creates Archives with collection_id set.
This test uses model_archive() (the production code path) instead of
archive_factory, so it will detect the collection_id bug.
"""
import datetime
import pytz
# Create archive files with a URL in the singlefile
domain = 'example.com'
url = f'https://{domain}/test-page'
archive_dir = test_directory / 'archives' / domain
archive_dir.mkdir(parents=True)
timestamp = archive_strftime(datetime.datetime(2000, 1, 1, 0, 0, 0).astimezone(pytz.UTC))
title = 'Test Page'
# Create a minimal singlefile with URL embedded (note trailing space after SingleFile)
singlefile_content = f'''<!DOCTYPE html> <html lang="en"><!--
Page saved with SingleFile
url: {url}
saved date: Mon May 16 2022 23:51:35 GMT+0000 (Coordinated Universal Time)
--><head><meta charset="utf-8">
<meta name="generator" content="SingleFile">
<title>{title}</title>
</head>
<body>
<h1>{title}</h1>
<p>Test content</p>
</body>
</html>'''
files = make_files_structure({
str(archive_dir / f'{timestamp}_{title}.html'): singlefile_content.strip(),
str(archive_dir / f'{timestamp}_{title}.readability.json'): '{"title": "' + title + '"}',
})
from wrolpi.files.models import FileGroup
# Create FileGroup from the files (simulating what refresh does)
file_paths = [pathlib.Path(f) for f in files]
file_group = FileGroup.from_paths(test_session, *file_paths)
# Model the archive using the production code path
# This should create an Archive with collection_id set
archive = model_archive(test_session, file_group)
# The bug: collection_id is None, causing NOT NULL constraint violation
assert archive is not None, "Archive should be created"
assert archive.collection_id is not None, \
"Archive.collection_id should be set (this is the bug!)"
# Verify the collection was created
collection = test_session.query(Collection).filter_by(
name=domain,
kind='domain'
).one_or_none()
assert collection is not None, "Domain collection should be created"
assert archive.collection_id == collection.id, \
"Archive should be linked to the domain collection"
@pytest.mark.asyncio
async def test_model_archive_extracts_url_from_singlefile(async_client, test_session, test_directory,
make_files_structure):
"""
Test that model_archive() can extract URL from singlefile when file_group.url is None.
This is the second part of the fix - if the URL isn't in the file_group yet,
we need to extract it from the singlefile content.
"""
import datetime
import pytz
# Create archive files
domain = 'test.org'
url = f'https://{domain}/article'
archive_dir = test_directory / 'archives' / domain
archive_dir.mkdir(parents=True)
timestamp = archive_strftime(datetime.datetime(2000, 1, 1, 0, 0, 0).astimezone(pytz.UTC))
# Create singlefile with URL in saved-from comment
singlefile_content = f'''<!DOCTYPE html> <html lang="en"><!--
Page saved with SingleFile
url: {url}
saved date: Mon May 16 2022 23:51:35 GMT+0000 (Coordinated Universal Time)
--><head><meta charset="utf-8">
<meta name="generator" content="SingleFile">
<title>Test Article</title>
</head>
<body><p>Content</p></body>
</html>'''
files = make_files_structure({
str(archive_dir / f'{timestamp}_Test.html'): singlefile_content.strip(),
})
from wrolpi.files.models import FileGroup
# Create FileGroup from the files (simulating what refresh does)
file_paths = [pathlib.Path(f) for f in files]
file_group = FileGroup.from_paths(test_session, *file_paths)
# Model the archive - it should extract the URL from the singlefile
archive = model_archive(test_session, file_group)
assert archive is not None
assert archive.file_group.url == url, \
"URL should be extracted from singlefile content"
assert archive.collection_id is not None, \
"collection_id should be set after URL extraction"
# Verify collection was created with correct domain
collection = test_session.get(Collection, archive.collection_id)
assert collection.name == domain
assert collection.kind == 'domain'
@pytest.mark.asyncio
async def test_archive_outside_directory_keeps_collection_directory(async_client, test_session, test_directory,
make_files_structure):
"""
Test that model_archive() does NOT clear collection.directory when archive is outside it.
Bug: Previously, if an archive was found outside its domain collection's directory,
the code would clear collection.directory = None. This was wrong because:
1. Archives should still be associated with domain collection based on URL
2. The collection's directory should remain intact for other archives
"""
import datetime
import pytz
domain = 'zerohedge.com'
url = f'https://{domain}/article'
# Create a domain collection WITH a specific directory
collection_dir = 'archive/News/zerohedge.com'
collection = Collection(name=domain, kind='domain', directory=collection_dir)
test_session.add(collection)
test_session.commit()
collection_id = collection.id
# Create the collection directory
(test_directory / collection_dir).mkdir(parents=True, exist_ok=True)
# Create archive files in a DIFFERENT directory (simulating manual file move)
old_archive_dir = test_directory / 'archive' / domain
old_archive_dir.mkdir(parents=True)
timestamp = archive_strftime(datetime.datetime(2000, 1, 1, 0, 0, 0).astimezone(pytz.UTC))
title = 'Test Article'
singlefile_content = f'''<!DOCTYPE html><html><!--
Page saved with SingleFile
url: {url}
--><head><title>{title}</title></head>
<body><p>Content</p></body></html>'''
files = make_files_structure({
str(old_archive_dir / f'{timestamp}_{title}.html'): singlefile_content.strip(),
str(old_archive_dir / f'{timestamp}_{title}.readability.json'): '{"title": "' + title + '"}',
})
from wrolpi.files.models import FileGroup
file_paths = [pathlib.Path(f) for f in files]
file_group = FileGroup.from_paths(test_session, *file_paths)
# Model the archive - archive is in archive/zerohedge.com but collection dir is archive/News/zerohedge.com
archive = model_archive(test_session, file_group)
assert archive is not None
assert archive.collection_id == collection_id, "Archive should be associated with domain collection"
# CRITICAL: The collection's directory should NOT be cleared
test_session.refresh(collection)
assert str(collection.directory) == collection_dir, \
f"Collection directory should NOT be cleared! Expected '{collection_dir}', got '{collection.directory}'"