diff --git a/wrolpi/common.py b/wrolpi/common.py index c04d60ff..3ea5233f 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -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') diff --git a/wrolpi/config_import.py b/wrolpi/config_import.py new file mode 100644 index 00000000..d05798ff --- /dev/null +++ b/wrolpi/config_import.py @@ -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] diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index bd51c21f..61270675 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -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) diff --git a/wrolpi/test/test_config_import.py b/wrolpi/test/test_config_import.py new file mode 100644 index 00000000..65e1bcfb --- /dev/null +++ b/wrolpi/test/test_config_import.py @@ -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) == [] diff --git a/wrolpi/test/test_downloader.py b/wrolpi/test/test_downloader.py index 0f54b6a3..cfce9d55 100644 --- a/wrolpi/test/test_downloader.py +++ b/wrolpi/test/test_downloader.py @@ -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. ),