rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Database configs import by diffing; download_manager.yaml first

Browse source 
Step 2 of the config import rework.  A database config can now declare a
ConfigSection for each kind of row it saves, and ConfigFile.import_config runs
a fixed sequence for it:

- read and validate the file once;
- decide from the versions (wrolpi.config_version) what the import may do;
- snapshot each section's rows as plain data keyed the way the file names them,
  and diff: added, changed, removed (none from an empty section, or without a
  current version);
- prepare (files, network) with no transaction open;
- write only the differences, and the file's version, in one short write.  A
  failure writes nothing.  If the database's version changed since the
  snapshot, the import decides again (and saves a newer database).

Subclasses override the section methods (file_rows, db_rows, upsert, delete)
and the prepare/after_import hooks.  The other database configs keep their
importers until their own steps.

download_manager.yaml is the first: an unchanged config imports with no write
(was 5,000 statements for 2,500 Downloads), new Downloads are one INSERT, and a
missing ZimSubscription is restored without rewriting the rest.  Settings equal
to the global defaults compare as unset, as they are saved.  A destination
written relative to the media directory is stored absolute.  Completed
once-downloads are never removed (they are not saved; #679).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-10-24 15:22:46 -06:00
parent 946043450e
commit 949d8b4923
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
5 changed files with 624 additions and 135 deletions
Download patch file Download diff file Expand all files Collapse all files

120
wrolpi/common.py
View file

@ -494,6 +494,10 @@ DB_CONFIG_FILE_NAMES ={'tags.yaml', 'channels.yaml', 'domains.yaml', 'download_m
'playlists.yaml', 'repos.yaml'}
class _DatabaseChanged(Exception):
"""The database changed a config's rows between an import's snapshot and its write."""
class ConfigFile:
"""
This is used to keep the DB and config files in sync. Importing the config reads the config from file, then
@ -759,8 +763,15 @@ class ConfigFile:
def get_relative_file(self):
return get_relative_to_media_directory(self.get_file())
# A database config that imports by diffing (see wrolpi.config_import) lists a `ConfigSection` for each kind of
# row it saves, parents first. Its `import_config` is the fixed sequence in `import_sections`.
sections: tuple = ()
def import_config(self, file: pathlib.Path = None, send_events=False):
"""Read config file data, apply it to the in-memory config and database."""
if self.sections:
return self.import_sections(file, send_events)
file = file or self.get_file()
file_str = str(get_relative_to_media_directory(file))
# Caller will set to successful if it works.
@ -782,6 +793,104 @@ class ConfigFile:
else:
logger_.error(f'Failed to import {file_str} because it does not exist.')
def import_sections(self, file: pathlib.Path = None, send_events=False):
"""Import a config with `sections`: read and validate the file once, decide what its version allows (see
wrolpi.config_version), diff each section against a snapshot of the database, prepare with no transaction
open, then write only the differences, and the file's version, in one short write."""
from wrolpi.events import Events
file = file or self.get_file()
file_str = str(get_relative_to_media_directory(file))
forced, self.force_import = self.force_import, False
self.successful_import = False
self.import_skipped = 0
try:
if not file.is_file():
# A missing config is not an empty one: import nothing, delete nothing.
logger_.info(f'No config file, skipping import: {file_str}')
self.successful_import = True
return
data = self.read_config_file(file) if file.stat().st_size else None
if not data or not self.validate(data):
raise InvalidConfig(f'Config is {"invalid" if data else "empty"}: {file_str}')
self._config.update({k: v for k, v in data.items() if k in self.default_config})
# Another writer may change the database between the snapshot and the write; then decide again.
for _ in range(3):
if self._import_sections_once(forced):
break
else:
raise RuntimeError(f'The database kept changing while importing {file_str}')
self.successful_import = True
except Exception as e:
self.successful_import = False
message = f'Failed to import {file_str}'
logger_.error(message, exc_info=e)
if send_events:
Events.send_config_import_failed(message)
raise
def _import_sections_once(self, forced: bool) -> bool:
"""One pass of `import_sections`. Returns False when the database's version changed before the write."""
from wrolpi.config_import import diff_section
from wrolpi.config_version import IMPORT, SAVE, get_config_version, set_config_version
from wrolpi.db import get_db_session
decision, db_version = self.version_decision(forced)
if decision == SAVE:
self.save_database_instead()
return True
file_rows = {i.name: i.file_rows(self._config) for i in self.sections}
with get_db_session() as session:
db_rows = {i.name: i.db_rows(session) for i in self.sections}
diffs = {i.name: diff_section(*file_rows[i.name], db_rows[i.name], deletes=decision == IMPORT)
for i in self.sections}
self.prepare(diffs)
file_version = self.version or 0
if any(diffs.values()) or (decision == IMPORT and db_version != file_version):
try:
with get_db_session(commit=True) as session:
try:
if get_config_version(session, self.file_name) != db_version:
raise _DatabaseChanged()
for section in self.sections:
if rows := {**diffs[section.name].added, **diffs[section.name].changed}:
section.upsert(session, rows)
for section in reversed(self.sections):
if diffs[section.name].removed:
section.delete(session, diffs[section.name].removed)
if decision == IMPORT:
# After the writes above, which counted as changes.
session.flush()
set_config_version(session, self.file_name, file_version)
except BaseException:
# None of it, and release the write lock now.
session.rollback()
raise
except _DatabaseChanged:
logger_.info(f'{self.file_name} changed in the database while importing; deciding again')
return False
self.import_skipped = sum(len(invalid) for _, invalid in file_rows.values())
changed = {name: diff for name, diff in diffs.items() if diff}
logger_.info(f'Imported {self.file_name} ({decision}): '
+ (', '.join(f'{name}: {len(diff.added)} added, {len(diff.changed)} changed, '
f'{len(diff.removed)} removed' for name, diff in changed.items()) or 'no changes'))
self.after_import(diffs)
if decision != IMPORT:
# The database may have rows the file does not list; save them (see `record_import_version`).
self.background_dump.activate_switch()
return True
def prepare(self, diffs: dict):
"""Slow work the write needs (files, network), run with no transaction open. `diffs` maps each section's
name to its `SectionDiff`."""
def after_import(self, diffs: dict):
"""Run after the write: activate switches, send events, start background tasks."""
def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False):
"""Dump database data, copy it to this the in-memory config, then write it to the config file."""
# Copy the data as-is by default. Other classes will overwrite this and dump the database before saving.
@ -830,6 +939,17 @@ class ConfigFile:
self.last_import_decision = decision
return decision
def version_decision(self, forced: bool = False) -> Tuple[str, Optional[int]]:
"""What importing this config's file (already read) may do, and the database version that decided it. For
`import_sections`, which records the version in its own write (`import_decision` is for the others)."""
from wrolpi.config_version import IMPORT, decide, get_config_version
from wrolpi.db import get_db_session
with get_db_session() as session:
db_version = get_config_version(session, self.file_name)
decision = IMPORT if forced else decide(self.version or 0, db_version)
logger_.debug(f'{self.file_name}: file version {self.version}, database version {db_version}: {decision}')
return decision, db_version
def save_database_instead(self):
"""The database has changes this config's file does not (see `import_decision`); save them."""
logger_.warning(f'{self.file_name} is older than the database; saving the database instead of importing it')

73
wrolpi/config_import.py Normal file
View file

@ -0,0 +1,73 @@
"""Importing a database config: read its file, diff it against a snapshot of the database, then write only the
differences in one short write (see `ConfigFile.import_config`).
A config declares a `ConfigSection` for each kind of row it saves (tags.yaml has Tags and the files they tag).
Rows are compared as plain data keyed the way the file names them (a URL, a name, a path), so the snapshot needs no
ORM objects and the read can finish before any slow preparation starts.
"""
from dataclasses import dataclass, field
from typing import Any, Dict, Hashable, Iterable, List, Set, Tuple
from sqlalchemy.orm import Session
@dataclass
class SectionDiff:
"""What applying a config file changes in one section. Rows are the file's, as `ConfigSection.file_rows`
returns them."""
added: Dict[Hashable, dict] = field(default_factory=dict)
changed: Dict[Hashable, dict] = field(default_factory=dict)
removed: Set[Hashable] = field(default_factory=set)
def __bool__(self):
return bool(self.added or self.changed or self.removed)
class ConfigSection:
"""One kind of row a database config saves.
Subclasses define how a file row and a database row become the same comparable dict (the values the config
saves, keyed by `key`), and how to write the differences."""
name: str = None
def file_rows(self, config: dict) -> Tuple[Dict[Hashable, dict], Set[Hashable]]:
"""The rows this section of `config` (the file's data) lists, keyed, plus the keys of rows that are invalid.
An invalid row is skipped, but its key counts as listed, so it is never deleted."""
raise NotImplementedError()
def db_rows(self, session: Session) -> Dict[Hashable, dict]:
"""The rows the config saves, keyed and compared like `file_rows`. Rows the config never saves are left
out: they are never "removed"."""
raise NotImplementedError()
def upsert(self, session: Session, rows: Dict[Hashable, dict]):
"""Write these added or changed rows. Fetch the current rows by key first: one may have been created or
deleted since the snapshot."""
raise NotImplementedError()
def delete(self, session: Session, keys: Set[Hashable]):
"""Delete the rows with these keys. A row deleted since the snapshot, or no longer saved by the config, is
left alone."""
raise NotImplementedError()
def diff_section(file_rows: Dict[Hashable, dict], invalid_keys: Set[Hashable], db_rows: Dict[Hashable, dict],
deletes: bool) -> SectionDiff:
"""Compare a section's file rows with its database rows. `deletes`: rows the file no longer lists are removed.
An empty section never removes anything: it cannot be told apart from a lost or blanked file."""
diff = SectionDiff()
for key, row in file_rows.items():
if key not in db_rows:
diff.added[key] = row
elif db_rows[key] != row:
diff.changed[key] = row
if deletes and (file_rows or invalid_keys):
diff.removed = db_rows.keys() - file_rows.keys() - invalid_keys
return diff
def chunks(items: Iterable[Any], size: int = 500) -> Iterable[List[Any]]:
"""Split `items` for `IN (...)` queries (SQLite limits the parameters of one statement)."""
items = list(items)
for i in range(0, len(items), size):
yield items[i:i + size]

268
wrolpi/downloader.py
View file

@ -17,14 +17,14 @@ from enum import Enum
from http import HTTPStatus
from itertools import filterfalse
from typing import List, Dict, Generator, Iterable, Coroutine, Any
from typing import Tuple, Optional, Callable, Awaitable
from typing import Tuple, Optional, Callable, Awaitable, Set
from urllib.parse import urlparse
from zoneinfo import ZoneInfo
import feedparser
import pytz
from feedparser import FeedParserDict
from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON, text
from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON, insert, or_, text
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Mapped, Session, column_property, relationship
@ -34,7 +34,8 @@ from wrolpi.cmd import which, run_command, CommandResult
from wrolpi.common import loaded, Base, ModelHelper, logger, wrol_mode_check, zig_zag, ConfigFile, \
wrol_mode_enabled, background_task, get_absolute_media_path, timer, aiohttp_get, \
get_download_info, trim_file_name, get_wrolpi_config, TRACE_LEVEL, normalize_domain
from wrolpi.config_version import IMPORT, SAVE, collection_kinds, count_config_change, register_ownership
from wrolpi.config_import import ConfigSection, chunks
from wrolpi.config_version import collection_kinds, count_config_change, register_ownership
from wrolpi.dates import TZDateTime, now, Seconds
from wrolpi.db import get_db_session, session_curs, serializer
from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \
@ -413,6 +414,13 @@ class Download(ModelHelper, Base): # noqa
return download
raise UnknownDownload(f'Cannot find Download with id: {id_}')
@staticmethod
def saved_by_config_filter() -> tuple:
"""Filters the Downloads download_manager.yaml saves. A completed once-download is not saved (its
`last_download_attempt` counts toward the daily download limits), so it is never removed by an import."""
return or_(Download.last_successful_download.is_(None),
Download.frequency.isnot(None) & (Download.frequency != 0)),
@staticmethod
def get_by_url(session: Session, url: str) -> Optional['Download']:
download = session.query(Download).filter(Download.url == url).one_or_none()
@ -2040,8 +2048,129 @@ class DownloadManagerConfigValidator:
skip_urls: list[str] = field(default_factory=list)
def inheritable_video_defaults() -> dict:
from modules.videos.lib import _extract_global_inheritable, get_videos_downloader_config
return _extract_global_inheritable(get_videos_downloader_config())
def _utc(value) -> Optional[datetime]:
"""A config's datetime (or ISO string), in UTC; one without a timezone is assumed to be UTC."""
if not value:
return None
if isinstance(value, str):
value = datetime.fromisoformat(value)
return (value if value.tzinfo else value.replace(tzinfo=pytz.utc)).astimezone(pytz.utc)
def is_zim_subscription_download(url: str, frequency: Optional[int]) -> bool:
"""A recurring Kiwix Download belongs to a ZimSubscription."""
return bool(frequency) and url.startswith('https://download.kiwix.org/')
class DownloadsSection(ConfigSection):
"""download_manager.yaml's Downloads, keyed by URL. A recurring Kiwix Download's ZimSubscription is derived
from it: `subscribed` compares whether the subscription exists, so a missing one is restored."""
name = 'downloads'
@staticmethod
def _row(url, destination, downloader, frequency, last_successful_download, next_download, settings, status,
sub_downloader, tag_names, subscribed, defaults: dict) -> dict:
return dict(
destination=str(destination) if destination else None,
downloader=downloader,
frequency=frequency or None,
last_successful_download=_utc(last_successful_download),
next_download=_utc(next_download),
# Settings equal to the global defaults are inherited, and not saved.
settings=DownloadManagerConfig._filter_settings_for_yaml(settings, defaults) or None,
status=status or DownloadStatus.new,
sub_downloader=sub_downloader,
tag_names=list(tag_names) if tag_names else None,
subscribed=subscribed,
)
def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]:
defaults = inheritable_video_defaults()
rows, invalid = dict(), set()
for data in config.get('downloads') or []:
url = data.get('url')
try:
settings = dict(data.get('settings') or {})
# An old config kept `destination` and `tag_names` in the settings.
destination = data.get('destination') or settings.pop('destination', None)
tag_names = data.get('tag_names') or settings.pop('tag_names', None)
settings.pop('destination', None)
settings.pop('tag_names', None)
# Stored absolute (relative to the media directory), as the database compares it.
destination = get_absolute_media_path(destination) if destination else None
rows[url] = self._row(
url, destination, data['downloader'], data.get('frequency'),
data.get('last_successful_download'), data.get('next_download'), settings, data.get('status'),
data.get('sub_downloader'), tag_names, is_zim_subscription_download(url, data.get('frequency')),
defaults)
except Exception as e:
logger.error(f'Skipping invalid Download in config: {data}', exc_info=e)
invalid.add(url)
return rows, invalid
def db_rows(self, session: Session) -> Dict[str, dict]:
from modules.zim.models import ZimSubscription
defaults = inheritable_video_defaults()
query = session.query(
Download.url, Download.destination, Download.downloader, Download.frequency,
Download.last_successful_download, Download.next_download, Download.settings, Download.status,
Download.sub_downloader, Download.tag_names, ZimSubscription.id,
).outerjoin(ZimSubscription, ZimSubscription.download_id == Download.id) \
.filter(*Download.saved_by_config_filter())
return {url: self._row(url, *values, subscribed=subscription_id is not None, defaults=defaults)
for url, *values, subscription_id in query}
def upsert(self, session: Session, rows: Dict[str, dict]):
from modules.zim.lib import zim_download_url_to_name
from modules.zim.models import ZimSubscription
existing = dict()
for urls in chunks(rows):
existing.update({i.url: i for i in session.query(Download).filter(Download.url.in_(urls))})
values = {url: dict(
destination=row['destination'],
downloader=row['downloader'],
frequency=row['frequency'],
last_successful_download=row['last_successful_download'],
next_download=row['next_download'],
settings=row['settings'] or dict(),
status=row['status'],
sub_downloader=row['sub_downloader'],
tag_names=row['tag_names'],
) for url, row in rows.items()}
if new := [dict(url=url, **i) for url, i in values.items() if url not in existing]:
# One statement for all new Downloads (a rebuilt database has thousands). Not flushed: count it.
session.execute(insert(Download), new)
count_config_change(session, DownloadManagerConfig.file_name)
for url, download in existing.items():
for name, value in values[url].items():
setattr(download, name, value)
session.flush()
for url, row in rows.items():
if row['subscribed']:
name, language = zim_download_url_to_name(url)
subscription = session.query(ZimSubscription).filter_by(name=name).one_or_none() \
or ZimSubscription(name=name, language=language)
subscription.change_download(session, url, row['frequency'])
session.add(subscription)
def delete(self, session: Session, keys: Set[str]):
for urls in chunks(keys):
# Re-checked: a once-download may have completed since the snapshot; it is no longer saved.
for download in session.query(Download).filter(Download.url.in_(urls),
*Download.saved_by_config_filter()):
logger.warning(f'Deleting Download {download.url} because it is no longer in the config')
session.delete(download)
class DownloadManagerConfig(ConfigFile):
file_name = 'download_manager.yaml'
sections = (DownloadsSection(),)
default_config = dict(
downloads=[],
skip_urls=[],
@ -2066,14 +2195,13 @@ class DownloadManagerConfig(ConfigFile):
self.update({'downloads': value})
@staticmethod
def _filter_settings_for_yaml(settings: dict) -> dict:
def _filter_settings_for_yaml(settings: dict, global_defaults: dict = None) -> dict:
"""Remove inheritable video settings that match current global defaults.
Keeps YAML minimal — only overrides are persisted."""
if not settings:
return settings
from modules.videos.lib import INHERITABLE_VIDEO_SETTINGS, _extract_global_inheritable, \
get_videos_downloader_config
global_defaults = _extract_global_inheritable(get_videos_downloader_config())
from modules.videos.lib import INHERITABLE_VIDEO_SETTINGS
global_defaults = global_defaults if global_defaults is not None else inheritable_video_defaults()
filtered = {}
for k, v in settings.items():
if k in INHERITABLE_VIDEO_SETTINGS and v == global_defaults.get(k):
@ -2087,10 +2215,7 @@ class DownloadManagerConfig(ConfigFile):
version = self.db_version_for_dump(session)
downloads: Iterable[Download] = session.query(Download).order_by(Download.url)
new_downloads = []
for download in downloads:
if download.last_successful_download and not download.frequency:
# This once-download has completed, do not save it.
continue
for download in downloads.filter(*Download.saved_by_config_filter()):
destination = download.destination
new_download = dict(
destination=str(destination) if destination else None,
@ -2113,127 +2238,6 @@ class DownloadManagerConfig(ConfigFile):
if send_events:
Events.send_config_save_failed(message)
def import_config(self, file: pathlib.Path = None, send_events=False):
file = file or self.get_file()
# If config file doesn't exist, mark import as successful (nothing to import)
if not file.is_file():
logger.info('No downloads config file, skipping import')
self.successful_import = True
return
super().import_config(file)
decision = self.import_decision()
if decision == SAVE:
self.save_database_instead()
return
downloads_data = self.downloads
# Empty downloads list = never delete DB records
if not downloads_data:
logger.info(f'No downloads in config, preserving existing DB downloads')
self.successful_import = True
return
with get_db_session(commit=True) as session:
from modules.zim.lib import zim_download_url_to_name
from modules.zim.models import ZimSubscription
try:
logger.warning('Importing downloads in config')
downloads_by_url = {i['url']: i for i in downloads_data}
existing_downloads: List[Download] = list(session.query(Download))
for existing in existing_downloads:
download = downloads_by_url.pop(existing.url, None)
if download:
# Download in config already exists, update the DB record.
# The config is the source of truth.
existing.downloader = download['downloader']
existing.destination = download['destination']
existing.frequency = download['frequency']
existing.last_successful_download = download.get('last_successful_download')
existing.next_download = download.get('next_download')
existing.status = download['status']
existing.sub_downloader = download['sub_downloader']
existing.settings = download.get('settings') or dict()
if 'destination' in existing.settings:
# `destination` may be from old config, or from ChannelDownloader.
destination = existing.settings.pop('destination')
existing.destination = existing.destination or destination
if 'tag_names' in existing.settings:
# `tag_names` may be from old config, or from ChannelDownloader.
tag_names = existing.settings.pop('tag_names')
existing.tag_names = existing.tag_names or tag_names
existing.flush()
elif existing.is_complete and not existing.frequency:
# Completed once-downloads are never written to the config (see `dump_config`), so their
# absence does not mean the user removed them. Keep them: their `last_download_attempt`
# is what the daily download limits count, and that count must survive a restart.
continue
elif decision == IMPORT:
logger.warning(f'Deleting Download {existing.url} because it is no longer in the config')
session.delete(existing)
for download in downloads_by_url.values():
# These downloads are new, import them.
download = Download(
destination=download['destination'],
downloader=download['downloader'],
frequency=download['frequency'],
last_successful_download=download.get('last_successful_download'),
next_download=download.get('next_download'),
settings=download['settings'] or dict(),
status=download['status'],
sub_downloader=download['sub_downloader'],
tag_names=download['tag_names'],
url=(url := download['url']),
)
session.add(download)
logger.info(f'Adding new download {url}')
session.commit()
self.successful_import = True
except Exception as e:
self.successful_import = False
message = f'Failed to import {self.file_name}'
logger.error(message, exc_info=e)
if send_events:
Events.send_config_import_failed(message)
raise
try:
# Claim any Kiwix subscriptions for ZimSubscription(s).
downloads: List[Download] = session.query(Download)
need_commit = False
for download in downloads:
if not download.url.startswith('https://download.kiwix.org/') or not download.frequency:
# ZimSubscription requires download.kiwix.org AND a frequency.
continue
name, language = zim_download_url_to_name(download.url)
subscription = session.query(ZimSubscription).filter_by(name=name, language=language).one_or_none()
if not subscription:
subscription = ZimSubscription(name=name, language=language)
subscription.change_download(session, download.url, download.frequency)
session.add(subscription)
need_commit = True
if need_commit:
session.commit()
self.record_import_version()
self.successful_import = True
except Exception as e:
self.successful_import = False
message = 'Failed to restore ZimSubscriptions'
logger.error(message, exc_info=e)
if send_events:
Events.send_config_import_failed(message)
raise
self.successful_import = True
def preview_backup_import(self, backup_date: str, mode: str) -> dict:
backup_file = self._get_backup_file(backup_date)
backup_data = self.read_config_file(backup_file)

292
wrolpi/test/test_config_import.py Normal file
View file

@ -0,0 +1,292 @@
"""A database config imports by reading its file, diffing it against a snapshot of the database, and writing only the
differences in one short write (see wrolpi.config_import). download_manager.yaml is the first config to import this
way."""
import sqlite3
import threading
from contextlib import contextmanager
from datetime import timedelta
import pytest
import yaml
from sqlalchemy import event, text
from sqlalchemy.engine import Engine
from modules.zim.models import ZimSubscription
from wrolpi import conftest
from wrolpi.config_version import get_config_version
from wrolpi.dates import now
from wrolpi.db import get_db_session
from wrolpi.downloader import Download, DownloadManagerConfig, get_download_manager_config, import_downloads_config
WRITES = ('INSERT', 'UPDATE', 'DELETE', 'BEGIN IMMEDIATE')
KIWIX_URL = 'https://download.kiwix.org/zim/wikisource/wikisource_en_all_maxi_'
@contextmanager
def statements():
"""Collect the SQL statements run in this block."""
collected = []
def collect(conn, cursor, statement, *_):
collected.append(statement.lstrip())
event.listen(Engine, 'before_cursor_execute', collect)
try:
yield collected
finally:
event.remove(Engine, 'before_cursor_execute', collect)
def writes(collected):
return [i for i in collected if i.upper().startswith(WRITES)]
def database_path(test_session) -> str:
return test_session.get_bind().url.database
@pytest.fixture
def production_like_sessions(test_session, saved_downloads):
"""Each `get_db_session` gets its own connection, as in production (after `saved_downloads` is committed)."""
test_session.commit()
with conftest.production_like_sessions(test_session) as maker:
yield maker
@pytest.fixture
async def saved_downloads(test_session, test_directory, test_download_manager_config, await_switches):
"""Downloads of every shape download_manager.yaml saves, saved to the config (which records its version)."""
destination = test_directory / 'videos/channel'
yesterday = now() - timedelta(days=1)
test_session.add_all([
Download(url='https://example.com/once', downloader='archive', status='new'),
Download(url='https://example.com/recurring', downloader='video_channel', frequency=86400,
sub_downloader='video', destination=destination, tag_names=['one', 'two'],
settings=dict(video_resolutions=['720p']), last_successful_download=yesterday,
next_download=now() + timedelta(days=1), status='complete'),
Download(url='https://example.com/removed', downloader='rss', frequency=3600, status='new'),
# Not saved (completed once-downloads count toward the daily limit), so never "removed".
Download(url='https://example.com/completed', downloader='archive', status='complete',
last_successful_download=yesterday),
])
test_session.commit()
from modules.zim import lib as zim_lib
await zim_lib.subscribe(test_session, 'Wikisource', 'en')
test_session.commit()
get_download_manager_config().dump_config()
await await_switches()
assert get_config_version(test_session, 'download_manager.yaml') is not None
return test_download_manager_config
def edit_config(path, change):
data = yaml.safe_load(path.read_text())
change(data)
path.write_text(yaml.dump(data))
def urls(test_session):
test_session.expire_all()
return {i.url for i in test_session.query(Download)}
@pytest.mark.asyncio
async def test_unchanged_import_writes_nothing(test_session, saved_downloads):
"""Importing a config the database already matches opens no write transaction."""
config = get_download_manager_config()
with statements() as collected:
config.import_config()
assert config.successful_import is True
assert writes(collected) == []
# The version, then one snapshot of the Downloads (with their Zim subscriptions).
assert len(collected) <= 4, collected
@pytest.mark.asyncio
async def test_import_writes_only_the_differences(test_session, saved_downloads):
"""An added Download is inserted, a changed one is updated, a removed one is deleted; the others are untouched.
Completed once-downloads are not saved, so they are never removed."""
def change(data):
downloads = {i['url']: i for i in data['downloads']}
downloads['https://example.com/recurring']['frequency'] = 604800
del downloads['https://example.com/removed']
downloads['https://example.com/added'] = dict(url='https://example.com/added', downloader='archive',
status='new', last_successful_download=None,
next_download=None)
data['downloads'] = list(downloads.values())
edit_config(saved_downloads, change)
config = get_download_manager_config()
with statements() as collected:
config.import_config()
assert config.successful_import is True
assert urls(test_session) == {'https://example.com/once', 'https://example.com/recurring',
'https://example.com/added', 'https://example.com/completed', KIWIX_URL}
assert Download.get_by_url(test_session, 'https://example.com/recurring').frequency == 604800
download_writes = [i for i in writes(collected) if ' download' in i.lower()]
assert len(download_writes) == 3, download_writes
@pytest.mark.asyncio
async def test_zim_subscription_is_restored(test_session, saved_downloads):
"""A Kiwix Download's Zim subscription is derived; a missing one is restored, an existing one is left alone."""
test_session.query(ZimSubscription).delete()
test_session.commit()
get_download_manager_config().import_config()
subscription = test_session.query(ZimSubscription).one()
assert (subscription.name, subscription.language, subscription.download.url) == ('Wikisource', 'en', KIWIX_URL)
@pytest.mark.asyncio
async def test_failed_write_changes_nothing(test_session, saved_downloads, production_like_sessions, monkeypatch):
"""The import's changes and its version are one write: a failure leaves neither."""
from wrolpi.downloader import DownloadsSection
edit_config(saved_downloads, lambda data: data['downloads'].append(dict(
url='https://example.com/added', downloader='archive', status='new', last_successful_download=None,
next_download=None)))
edit_config(saved_downloads, lambda data: data.update(
version=data['version'] + 1,
downloads=[i for i in data['downloads'] if i['url'] != 'https://example.com/removed']))
before = get_config_version(test_session, 'download_manager.yaml')
def fail(*args, **kwargs):
raise RuntimeError('Write failed')
monkeypatch.setattr(DownloadsSection, 'delete', fail)
config = get_download_manager_config()
with pytest.raises(RuntimeError):
config.import_config()
assert config.successful_import is False
assert 'https://example.com/added' not in urls(test_session)
assert 'https://example.com/removed' in urls(test_session)
assert get_config_version(test_session, 'download_manager.yaml') == before
@pytest.mark.asyncio
async def test_import_waits_for_a_concurrent_writer(test_session, saved_downloads, production_like_sessions):
"""Another connection holding the write lock delays the import's write; it does not fail it."""
edit_config(saved_downloads, lambda data: data['downloads'].append(dict(
url='https://example.com/added', downloader='archive', status='new', last_successful_download=None,
next_download=None)))
locked, release = threading.Event(), threading.Event()
def hold_lock():
conn = sqlite3.connect(database_path(test_session), isolation_level=None)
conn.execute('BEGIN IMMEDIATE')
locked.set()
release.wait(5)
conn.execute('COMMIT')
conn.close()
thread = threading.Thread(target=hold_lock)
thread.start()
locked.wait(5)
threading.Timer(0.5, release.set).start()
try:
get_download_manager_config().import_config()
finally:
release.set()
thread.join()
with get_db_session() as session:
assert Download.get_by_url(session, 'https://example.com/added')
@pytest.mark.asyncio
async def test_write_lock_is_free_while_preparing(test_session, saved_downloads, production_like_sessions,
monkeypatch):
"""Slow preparation (files, network) runs with no transaction open: other writers are not blocked."""
edit_config(saved_downloads, lambda data: data['downloads'].append(dict(
url='https://example.com/added', downloader='archive', status='new', last_successful_download=None,
next_download=None)))
took_lock = []
def prepare(self, diffs):
conn = sqlite3.connect(database_path(test_session), isolation_level=None, timeout=0)
conn.execute('BEGIN IMMEDIATE')
conn.execute('COMMIT')
conn.close()
took_lock.append(True)
monkeypatch.setattr(DownloadManagerConfig, 'prepare', prepare)
get_download_manager_config().import_config()
assert took_lock == [True]
@pytest.mark.asyncio
async def test_database_change_during_import_is_kept(test_session, saved_downloads, production_like_sessions,
monkeypatch, await_switches):
"""A Download added after the import read the database (and before it wrote) is not deleted by the older
file: the import decides again, finds the database newer, and saves it."""
edit_config(saved_downloads, lambda data: data['downloads'].pop())
prepared = []
def prepare(self, diffs):
if not prepared:
with get_db_session(commit=True) as session:
session.add(Download(url='https://example.com/concurrent', downloader='archive', status='new'))
prepared.append(True)
monkeypatch.setattr(DownloadManagerConfig, 'prepare', prepare)
config = get_download_manager_config()
config.import_config()
await await_switches()
assert config.successful_import is True
with get_db_session() as session:
assert Download.get_by_url(session, 'https://example.com/concurrent')
saved = {i['url'] for i in yaml.safe_load(saved_downloads.read_text())['downloads']}
assert 'https://example.com/concurrent' in saved
@pytest.mark.asyncio
async def test_unknown_version_imports_without_deletes(test_session, saved_downloads, await_switches):
"""Without a database version, a Download the file does not list is kept (and then saved into the file)."""
test_session.execute(text('DELETE FROM config_version'))
test_session.commit()
edit_config(saved_downloads, lambda data: data.update(
downloads=[i for i in data['downloads'] if i['url'] != 'https://example.com/removed']))
await import_downloads_config()
await await_switches()
assert 'https://example.com/removed' in urls(test_session)
saved = {i['url'] for i in yaml.safe_load(saved_downloads.read_text())['downloads']}
assert 'https://example.com/removed' in saved
assert get_config_version(test_session, 'download_manager.yaml') is not None
@pytest.mark.asyncio
async def test_import_button(test_session, saved_downloads, async_client):
"""The Import button applies the file whatever its version (the user chose it), inside the request's write."""
test_session.add(Download(url='https://example.com/unsaved', downloader='archive', status='new'))
test_session.commit() # The database is now newer than the file.
request, response = await async_client.post('/api/config/import', json=dict(file_name='download_manager.yaml'))
assert response.status_code == 204
assert 'https://example.com/unsaved' not in urls(test_session)
@pytest.mark.asyncio
async def test_relative_destination(test_session, test_directory, saved_downloads):
"""A destination written relative to the media directory is stored absolute, and is then in step."""
edit_config(saved_downloads, lambda data: data['downloads'].append(dict(
url='https://example.com/relative', downloader='archive', destination='archive/example.com',
status='new', last_successful_download=None, next_download=None)))
config = get_download_manager_config()
config.import_config()
test_session.expire_all()
assert Download.get_by_url(test_session, 'https://example.com/relative').destination \
== test_directory / 'archive/example.com'
with statements() as collected:
config.import_config()
assert writes(collected) == []

6
wrolpi/test/test_downloader.py
View file

@ -1054,7 +1054,7 @@ def test_download_settings_accepts_valid_download_order(order):
@pytest.mark.asyncio
async def test_downloads_config(test_session, test_download_manager, test_download_manager_config,
async def test_downloads_config(test_session, test_directory, test_download_manager, test_download_manager_config,
test_downloader, assert_downloads, tag_factory, await_switches):
# Can import with an empty config.
await import_downloads_config()
@ -1128,8 +1128,8 @@ async def test_downloads_config(test_session, test_download_manager, test_downlo
frequency=None,
next_download=datetime(2000, 1, 2, 0, 0, 0, tzinfo=pytz.UTC),
downloader='test_downloader',
# The destination was imported from the config.
destination=pathlib.Path('a different directory'),
# The destination was imported from the config, relative to the media directory.
destination=test_directory / 'a different directory',
sub_downloader=None,
tag_names=None, # Tags were removed.
),