From 95a7e64dd71cebb053ef6913c97f15fcb3358b52 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Wed, 2 Sep 2026 04:34:22 -0700 Subject: [PATCH] Centralize queued plugin execution --- archivebox/cli/archivebox_extract.py | 13 +- archivebox/plugins/discovery.py | 72 +------ archivebox/plugins/hooks.py | 176 +----------------- archivebox/progressmonitor/views.py | 8 +- archivebox/search/backends.py | 34 ++-- archivebox/search/query.py | 53 ++++-- archivebox/services/process_service.py | 14 +- archivebox/services/runner.py | 139 ++++++-------- archivebox/tests/conftest.py | 73 ++++++++ archivebox/tests/test_cli_piping.py | 4 +- archivebox/tests/test_cli_run.py | 19 +- .../test_cli_update_reindex_snapshots.py | 6 +- archivebox/tests/test_config_MAX_limits.py | 4 +- archivebox/tests/test_hooks.py | 91 ++------- .../tests/test_process_runtime_paths.py | 4 +- archivebox/tests/test_search.py | 21 +-- archivebox/tests/test_ui_admin_links.py | 12 +- archivebox/tests/test_ui_admin_machine.py | 4 +- archivebox/tests/test_ui_live_progress.py | 12 +- 19 files changed, 260 insertions(+), 499 deletions(-) diff --git a/archivebox/cli/archivebox_extract.py b/archivebox/cli/archivebox_extract.py index f5f915fd..88a6351c 100644 --- a/archivebox/cli/archivebox_extract.py +++ b/archivebox/cli/archivebox_extract.py @@ -294,9 +294,8 @@ def run_plugins( queue_at = timezone.now() if existing_snapshot_ids: if requested_rows: - # Search indexing on a sealed Snapshot is the only targeted hook - # allowed to bypass the normal lifecycle. Every other requested - # plugin requeues its Snapshot through the unified lifecycle. + # Explicit plugin retries are maintenance on the existing snapshot; + # preserve a sealed lifecycle while making its queued rows due. affected_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in rows_to_queue} if preserve_queued and queued_rows: queued_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in queued_rows} @@ -319,13 +318,7 @@ def run_plugins( for snapshot in Snapshot.objects.filter(id__in=affected_snapshot_ids).only("id", "status", "modified_at"): # Guard the read-time status so we never bump retry_at on a # row that's been re-queued / started by a concurrent runner. - plugin_names = requested_plugins_by_id.get(str(snapshot.id), set()) - sealed_search_backfill = ( - snapshot.status == Snapshot.StatusChoices.SEALED - and plugin_names - and all(plugin_name.startswith("search_backend_") for plugin_name in plugin_names) - ) - if sealed_search_backfill: + if snapshot.status == Snapshot.StatusChoices.SEALED and requested_plugins_by_id.get(str(snapshot.id)): snapshot.safe_update( {"retry_at": queue_at, "modified_at": queue_at}, refresh=False, diff --git a/archivebox/plugins/discovery.py b/archivebox/plugins/discovery.py index ae167c00..869a75cd 100644 --- a/archivebox/plugins/discovery.py +++ b/archivebox/plugins/discovery.py @@ -88,70 +88,14 @@ def get_enabled_plugins(config: ConfigLookup | None = None, **config_kwargs: Any return get_plugin_config_resolver().enabled_plugin_names_from_flat(dict(config.items())) -def discover_plugins_that_provide_interface( - module_name: str, - required_attrs: list[str], - plugin_prefix: str | None = None, -) -> dict[str, Any]: - """ - Discover plugins that provide a specific Python module with required interface. - - This enables dynamic plugin discovery for features like search backends, - storage backends, etc. without hardcoding imports. - """ - import importlib.util - - backends = {} - - for plugin_dir in iter_plugin_dirs(): - plugin_name = plugin_dir.name - if plugin_prefix and not plugin_name.startswith(plugin_prefix): - continue - - module_path = plugin_dir / f"{module_name}.py" - if not module_path.exists(): - continue - - try: - spec = importlib.util.spec_from_file_location( - f"archivebox.dynamic_plugins.{plugin_name}.{module_name}", - module_path, - ) - if spec is None or spec.loader is None: - continue - - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - - if not all(attr in vars(module) for attr in required_attrs): - continue - - if plugin_prefix: - backend_name = plugin_name[len(plugin_prefix) :] - else: - backend_name = plugin_name - - backends[backend_name] = module - - except Exception: - continue - - return backends - - -def get_search_backends() -> dict[str, Any]: - """ - Discover all available search backend plugins. - - Search backends must provide a search.py module with: - - search(query: str) -> List[str] (returns snapshot IDs) - - flush(snapshot_ids: Iterable[str]) -> None - """ - return discover_plugins_that_provide_interface( - module_name="search", - required_attrs=["search", "flush"], - plugin_prefix="search_backend_", - ) +def get_search_backends(): + """Return plugins that declare both standalone search commands.""" + catalog = get_plugin_catalog() + return { + plugin.name.removeprefix("search_backend_"): plugin + for plugin in catalog.values() + if catalog.command(plugin.name, "search") is not None and catalog.command(plugin.name, "flush") is not None + } @lru_cache(maxsize=1) diff --git a/archivebox/plugins/hooks.py b/archivebox/plugins/hooks.py index 2f7364d1..62a6c06d 100644 --- a/archivebox/plugins/hooks.py +++ b/archivebox/plugins/hooks.py @@ -8,40 +8,13 @@ from __future__ import annotations __package__ = "archivebox.plugins" -import os -from collections.abc import Mapping from pathlib import Path -from typing import TYPE_CHECKING, Any, Protocol, TypeGuard, runtime_checkable +from typing import Any -from asgiref.sync import async_to_sync +from abx_dl.models import parse_hook_filename -from abx_dl.execution import execute_hook -from abx_dl.models import Hook, parse_hook_filename - -from archivebox.config.constants import CONSTANTS -from archivebox.config.version import VERSION from archivebox.misc.util import fix_url_from_markdown, sanitize_extracted_url -from archivebox.plugins.discovery import ConfigLookup, get_enabled_plugins, get_plugin_catalog, get_plugin_special_config - -if TYPE_CHECKING: - from archivebox.machine.models import Process - - -@runtime_checkable -class ConfigDump(Protocol): - def as_dict(self) -> dict[str, Any]: ... - - -def _has_config_dump(config: object) -> TypeGuard[ConfigDump]: - return isinstance(config, ConfigDump) - - -def _config_to_overrides(config: ConfigLookup | Mapping[str, Any] | None) -> dict[str, Any]: - if config is None: - return {} - if _has_config_dump(config): - return dict(config.as_dict()) - return dict(config.items()) +from archivebox.plugins.discovery import ConfigLookup, get_enabled_plugins, get_plugin_catalog def is_background_hook(hook_name: str) -> bool: @@ -76,149 +49,6 @@ def discover_hooks( return [hook.path for _plugin, hook in get_plugin_catalog().hooks(normalized, names=names)] -def _catalog_hook(script: Path) -> Hook: - script = script.resolve() - for plugin in get_plugin_catalog().values(): - for hook in plugin.hooks: - if hook.path.resolve() == script: - return hook - parsed = parse_hook_filename(script.name) - if parsed is None: - raise ValueError(f"Not a valid plugin hook filename: {script.name}") - event, order, is_background = parsed - return Hook( - name=script.name, - event=event, - plugin_name=script.parent.name, - path=script, - order=order, - is_background=is_background, - ) - - -def _hook_environment(config: ConfigLookup | Mapping[str, Any] | None, **config_scope: Any) -> tuple[dict[str, str], Any]: - from archivebox.config.common import ( - ArchiveBoxConfig, - _archivebox_config_input_names, - get_config, - normalize_runtime_config, - ) - - overrides = _config_to_overrides(config) - resolved = get_config(overrides=overrides, **config_scope) - runtime = normalize_runtime_config( - resolved.for_crawl_runtime(runtime_overrides=overrides), - json_safe=False, - ) - runtime.update(normalize_runtime_config(overrides, json_safe=False)) - - env = os.environ.copy() - config_input_names = _archivebox_config_input_names() - for key in config_input_names: - env.pop(key, None) - env.pop("PLUGINS", None) - env["PATH"] = os.environ.get("PATH", "") - env["DATA_DIR"] = str(CONSTANTS.DATA_DIR) - env["LIBRARY_VERSION"] = VERSION - env.setdefault("MACHINE_ID", os.environ.get("MACHINE_ID", CONSTANTS.MACHINE_ID)) - - canonical_config_keys = set(ArchiveBoxConfig.model_fields) - for key, value in runtime.items(): - if key == "PATH" or value is None: - continue - if key in config_input_names and key not in canonical_config_keys: - continue - if isinstance(value, bool): - env[key] = "true" if value else "false" - elif isinstance(value, (dict, list)): - import json - - env[key] = json.dumps(value) - else: - env[key] = str(value) - - node_modules_dir = runtime.get("NODE_MODULES_DIR") - lib_dir = runtime.get("ABXPKG_LIB_DIR") - if not node_modules_dir and lib_dir: - node_modules_dir = Path(lib_dir) / "pnpm" / "packages" / "chrome" / "node_modules" - if node_modules_dir: - env["NODE_MODULES_DIR"] = str(node_modules_dir) - env["NODE_MODULE_DIR"] = str(node_modules_dir) - node_path = [part for part in str(runtime.get("NODE_PATH") or "").split(os.pathsep) if part] - if str(node_modules_dir) not in node_path: - node_path.append(str(node_modules_dir)) - env["NODE_PATH"] = os.pathsep.join(node_path) - return env, resolved - - -def run_hook( - script: Path, - output_dir: Path, - config: ConfigLookup | Mapping[str, Any] | None = None, - timeout: int | None = None, - parent: Process | None = None, - **kwargs: Any, -) -> Process: - """Compatibility adapter for finite direct calls; abx-dl owns execution.""" - from archivebox.machine.models import Process - from archivebox.services.process_service import ProcessService as PersistedProcessService - from archivebox.services.process_service import parse_event_datetime - from abx_dl.orchestrator import create_bus - - # Preserve the old direct-call contract: hooks are children of an explicit - # parent, or of the current ArchiveBox process when one can be identified. - # This belongs on the DB projection adapter, not in hook CLI arguments. - if parent is None: - try: - parent = Process.current() - except Exception: - parent = None - config_scope = {key.removeprefix("config_"): kwargs.pop(key) for key in list(kwargs) if key.startswith("config_")} - env, resolved = _hook_environment(config, **config_scope) - hook = _catalog_hook(script) - if timeout is None: - timeout = get_plugin_special_config(hook.plugin_name, resolved)["timeout"] - timeout = min(int(timeout or 300), int(CONSTANTS.MAX_HOOK_RUNTIME_SECONDS)) - - bus = create_bus(name=f"ArchiveBoxHook_{hook.plugin_name}", total_timeout=float(timeout) + 30.0) - PersistedProcessService(bus, parent_process_id=str(parent.id) if parent is not None else None) - - async def execute_and_close(): - try: - return await execute_hook( - hook, - output_dir=output_dir, - env=env, - arguments=kwargs, - timeout=timeout, - bus=bus, - process_type=Process.TypeChoices.HOOK, - ) - finally: - await bus.wait_until_idle() - await bus.destroy(clear=False) - - completed = async_to_sync(execute_and_close)() - started_at = parse_event_datetime(completed.start_ts) - process = Process.objects.filter(pid=completed.pid or None, started_at=started_at).order_by("-modified_at").first() - if process is None: - raise RuntimeError(f"Hook {hook.full_name} completed without an ArchiveBox Process projection") - return process - - -def extract_records_from_process(process: Process) -> list[dict[str, Any]]: - """Return hook JSONL records with generic catalog identity attached.""" - records = process.get_records() - plugin_name = Path(process.pwd).name if process.pwd else "unknown" - plugin_hook = next((str(arg) for arg in process.cmd if Path(str(arg)).name.startswith("on_")), "") - hook_name = Path(plugin_hook).name if plugin_hook else "unknown" - for record in records: - record.setdefault("plugin", plugin_name) - record.setdefault("hook_name", hook_name) - record.setdefault("plugin_hook", plugin_hook) - return records - - def collect_urls_from_plugins(snapshot_dir: Path) -> list[dict[str, Any]]: """Read the durable urls.jsonl interface emitted by parser plugins.""" urls: list[dict[str, Any]] = [] diff --git a/archivebox/progressmonitor/views.py b/archivebox/progressmonitor/views.py index a33bfc84..4f0c96d3 100644 --- a/archivebox/progressmonitor/views.py +++ b/archivebox/progressmonitor/views.py @@ -30,12 +30,16 @@ def progress_endpoint(scope: Literal["crawl", "snapshot"] | None = None, object_ @lru_cache(maxsize=1) def _live_progress_plugin_names() -> tuple[frozenset[str], frozenset[str]]: plugin_configs = discover_plugin_configs() + indexing_plugin_names = frozenset( + plugin_name + for plugin_name, plugin_config in plugin_configs.items() + if {"search", "flush"}.issubset(plugin_config.get("commands", {})) + ) download_plugin_names = frozenset( plugin_name for plugin_name, plugin_config in plugin_configs.items() - if plugin_config.get("output_mimetypes") and not plugin_name.startswith("search_backend_") + if plugin_config.get("output_mimetypes") and plugin_name not in indexing_plugin_names ) - indexing_plugin_names = frozenset(plugin_name for plugin_name in plugin_configs if plugin_name.startswith("search_backend_")) return download_plugin_names, indexing_plugin_names diff --git a/archivebox/search/backends.py b/archivebox/search/backends.py index fba5ab5a..8311437d 100644 --- a/archivebox/search/backends.py +++ b/archivebox/search/backends.py @@ -1,7 +1,7 @@ __package__ = "archivebox.search" +import json import os -from contextlib import contextmanager from typing import Any from archivebox.config.common import get_config @@ -10,29 +10,21 @@ from archivebox.config.common import get_config _search_backends_cache: dict | None = None -@contextmanager -def search_backend_env(config: dict[str, Any] | None = None, **config_kwargs: Any): - """Temporarily expose resolved search config through os.environ for backend code.""" +def search_backend_command_env(config: dict[str, Any] | None = None, **config_kwargs: Any) -> dict[str, str]: + """Serialize resolved application config for a standalone plugin command.""" config = config or get_config(**config_kwargs) - updates = {} + env = os.environ.copy() for key, value in config.items(): key = str(key) - if not (key.startswith("SEARCH_BACKEND_") or key.endswith("_BINARY")): - continue if value is None: continue - if isinstance(value, (str, int, float, bool, os.PathLike)): - updates[key] = str(value) - previous = {key: os.environ.get(key) for key in updates} - os.environ.update(updates) - try: - yield - finally: - for key, value in previous.items(): - if value is None: - os.environ.pop(key, None) - else: - os.environ[key] = value + if isinstance(value, bool): + env[key] = "true" if value else "false" + elif isinstance(value, (dict, list, tuple)): + env[key] = json.dumps(value) + elif isinstance(value, (str, int, float, os.PathLike)): + env[key] = str(value) + return env def normalize_search_backend_name(backend_name: str | None) -> str: @@ -41,7 +33,7 @@ def normalize_search_backend_name(backend_name: str | None) -> str: def get_available_backends() -> dict: - """Discover search backend plugin modules and cache them in memory.""" + """Discover search-capable plugins and cache their catalog entries.""" global _search_backends_cache if _search_backends_cache is None: @@ -53,7 +45,7 @@ def get_available_backends() -> dict: def get_backend(config: dict[str, Any] | None = None, **config_kwargs: Any) -> Any: - """Resolve the configured search backend module.""" + """Resolve the configured search-capable plugin.""" config = config or get_config(**config_kwargs) backend_name = normalize_search_backend_name(config.SEARCH_BACKEND_ENGINE) backends = get_available_backends() diff --git a/archivebox/search/query.py b/archivebox/search/query.py index 5761de83..2b16732a 100644 --- a/archivebox/search/query.py +++ b/archivebox/search/query.py @@ -8,7 +8,7 @@ from django.db.models import Case, IntegerField, Q, QuerySet, Value, When from archivebox.config.common import get_config from archivebox.misc.logging import stderr from archivebox.misc.util import enforce_types -from archivebox.search.backends import get_available_backends, get_backend, normalize_search_backend_name, search_backend_env +from archivebox.search.backends import get_available_backends, get_backend, normalize_search_backend_name, search_backend_command_env from archivebox.search.config import get_search_mode, get_search_mode_backend, get_search_mode_base @@ -278,18 +278,27 @@ def iter_query_search_ids( # --csv/--json. Keep daemon diagnostics on stderr. with redirect_stdout(sys.stderr): ensure_daemon_stack(reason="search query") - with search_backend_env(config=config): - if backend_name == "ripgrep": - ids = backend.iter_search(query, search_mode=search_mode_base) - else: - ids = backend.search(query) - for snapshot_id in ids: - if snapshot_id in seen: - continue - seen.add(snapshot_id) - yield snapshot_id - if max_results and len(seen) >= max_results: - return + from abx_dl.execution import iter_plugin_command + from archivebox.config.constants import CONSTANTS + from archivebox.plugins.discovery import get_plugin_catalog + + command = get_plugin_catalog().command(backend.name, "search") + if command is None: + raise RuntimeError(f'Plugin "{backend.name}" does not expose a search command') + ids = iter_plugin_command( + command, + arguments={"query": query, "search_mode": search_mode_base}, + env=search_backend_command_env(config=config), + cwd=CONSTANTS.DATA_DIR, + timeout=max(1, int(config.get("TIMEOUT", 60))) * 4, + ) + for snapshot_id in ids: + if snapshot_id in seen: + continue + seen.add(snapshot_id) + yield snapshot_id + if max_results and len(seen) >= max_results: + return successful_backends += 1 except Exception as err: errors.append(err) @@ -318,8 +327,22 @@ def flush_search_index(snapshots: QuerySet, config: dict[str, Any] | None = None snapshot_pks = [str(pk) for pk in snapshots.values_list("pk", flat=True)] try: - with search_backend_env(config=config): - backend.flush(snapshot_pks) + from abx_dl.execution import iter_plugin_command + from archivebox.config.constants import CONSTANTS + from archivebox.plugins.discovery import get_plugin_catalog + + command = get_plugin_catalog().command(backend.name, "flush") + if command is None: + raise RuntimeError(f'Plugin "{backend.name}" does not expose a flush command') + list( + iter_plugin_command( + command, + stdin=snapshot_pks, + env=search_backend_command_env(config=config), + cwd=CONSTANTS.DATA_DIR, + timeout=max(1, int(config.get("TIMEOUT", 60))) * 4, + ), + ) except Exception as err: stderr() stderr( diff --git a/archivebox/services/process_service.py b/archivebox/services/process_service.py index ab919149..870550ce 100644 --- a/archivebox/services/process_service.py +++ b/archivebox/services/process_service.py @@ -61,12 +61,8 @@ class ProcessService(BaseService): ] EMITS: ClassVar[list[type[BaseEvent]]] = [] - def __init__(self, bus, *, parent_process_id: str | None = None): + def __init__(self, bus): self._iface = None - # A direct run_hook() call owns a private bus, so its caller-supplied - # parent applies to every process projected from that bus. Crawl buses - # leave this unset and derive their hierarchy from lifecycle events. - self.parent_process_id = parent_process_id self._completed_queue: asyncio.Queue[ProcessCompletedEvent | None] = asyncio.Queue() self._completed_worker: asyncio.Task | None = None super().__init__(bus) @@ -104,7 +100,7 @@ class ProcessService(BaseService): process = await Process.objects.acreate( machine=iface.machine, iface=iface, - parent_id=self.parent_process_id, + parent_id=None, process_type=process_type, worker_type=worker_type, pwd=event.output_dir, @@ -138,7 +134,7 @@ class ProcessService(BaseService): hook_path=event.hook_path, ) await Process.objects.filter(id=process.id).aupdate( - parent_id=self.parent_process_id or process.parent_id, + parent_id=process.parent_id, pwd=process.pwd, cmd=process.cmd, env=process.env, @@ -225,7 +221,7 @@ class ProcessService(BaseService): await Process.objects.acreate( machine=iface.machine, iface=iface, - parent_id=self.parent_process_id, + parent_id=None, process_type=process_type, worker_type=worker_type, pwd=event.output_dir, @@ -246,7 +242,7 @@ class ProcessService(BaseService): updates = { "machine_id": iface.machine_id, "iface_id": iface.id, - "parent_id": self.parent_process_id or process.parent_id, + "parent_id": process.parent_id, "pwd": event.output_dir, "env": process_env, "pid": event.pid or process.pid, diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 89b4cbdc..1d61a43a 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -316,16 +316,9 @@ class CrawlRunner: @property def allow_maintenance_on_inactive_crawl(self) -> bool: - """Run targeted search indexing on an already-sealed snapshot. - - This is the singular hook-execution exception to the unified lifecycle. - All other work must queue the Snapshot and Crawl normally. - """ + """Run explicitly targeted plugin work on already-sealed snapshots.""" return bool( - self.initial_snapshot_ids - and self.selected_plugins - and self.crawl.status == self.crawl.StatusChoices.SEALED - and all(plugin.startswith("search_backend_") for plugin in self.selected_plugins), + self.initial_snapshot_ids and self.selected_plugins and self.crawl.status == self.crawl.StatusChoices.SEALED, ) async def run(self) -> None: @@ -603,7 +596,7 @@ class CrawlRunner: return [] if self.initial_snapshot_ids: # Explicit ids select normal runnable work, except for the one - # sealed-search backfill admitted by allow_maintenance_on_inactive_crawl. + # targeted maintenance admitted by allow_maintenance_on_inactive_crawl. return [str(snapshot_id) for snapshot_id in self.initial_snapshot_ids] pending_snapshots = list( self.crawl.snapshot_set.filter(status__in=Snapshot.RUNNABLE_STATES) @@ -1411,9 +1404,6 @@ def config_overrides_for_queued_plugins(selected_plugins: list[str], **overrides ) for plugin_name, enabled_key in _plugin_enabled_config_keys().items(): config_overrides[enabled_key] = plugin_name in selected_plugin_names - for plugin_name in selected_plugins: - if plugin_name.startswith("search_backend_"): - config_overrides[f"{plugin_name.upper()}_ENABLED"] = True return config_overrides @@ -1707,48 +1697,30 @@ def _run_due_snapshot_locked(snapshot, *, lock_seconds: int, interactive_interru snapshot.refresh_from_db() selected_plugins = queued_plugins_for_snapshot(str(snapshot.id)) if selected_plugins: - search_only_plugins = all(plugin.startswith("search_backend_") for plugin in selected_plugins) - if not search_only_plugins: - snapshot.update_and_requeue( - status=Snapshot.StatusChoices.QUEUED, - retry_at=timezone.now(), - current_step=0, - ) - snapshot.refresh_from_db() - else: - _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot) - run_crawl( - str(snapshot.crawl_id), - snapshot_ids=[str(snapshot.id)], - selected_plugins=selected_plugins, - process_discovered_snapshots_inline=True, - interactive_interrupts=interactive_interrupts, - config_overrides=config_overrides_for_queued_plugins(selected_plugins), - selected_plugins_are_explicit=False, - ) - from archivebox.core.models import ArchiveResult + _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot) + run_crawl( + str(snapshot.crawl_id), + snapshot_ids=[str(snapshot.id)], + selected_plugins=selected_plugins, + process_discovered_snapshots_inline=True, + interactive_interrupts=interactive_interrupts, + config_overrides=config_overrides_for_queued_plugins(selected_plugins), + selected_plugins_are_explicit=False, + ) + from archivebox.core.models import ArchiveResult - has_queued_results = ArchiveResult.objects.filter( - snapshot_id=snapshot.id, - status=ArchiveResult.StatusChoices.QUEUED, - ).exists() - if not has_queued_results: - type(snapshot).objects.filter( - pk=snapshot.pk, - status=snapshot.StatusChoices.SEALED, - ).update( - retry_at=None, - modified_at=timezone.now(), - ) - else: - type(snapshot).objects.filter( - pk=snapshot.pk, - status=snapshot.StatusChoices.SEALED, - ).update( - retry_at=timezone.now(), - modified_at=timezone.now(), - ) - return True + has_queued_results = ArchiveResult.objects.filter( + snapshot_id=snapshot.id, + status=ArchiveResult.StatusChoices.QUEUED, + ).exists() + type(snapshot).objects.filter( + pk=snapshot.pk, + status=snapshot.StatusChoices.SEALED, + ).update( + retry_at=timezone.now() if has_queued_results else None, + modified_at=timezone.now(), + ) + return True if snapshot.status == Snapshot.StatusChoices.SEALED: if maintenance_ran: return True @@ -2102,23 +2074,22 @@ def _run_due_queued_plugin_result( config_overrides=config_overrides_for_queued_plugins(selected_plugins, CRAWL_MAX_CONCURRENT_SNAPSHOTS=batch_size), selected_plugins_are_explicit=False, ) - if all(plugin.startswith("search_backend_") for plugin in selected_plugins): - queued_results = ArchiveResult.objects.filter( - snapshot_id=OuterRef("pk"), - status=ArchiveResult.StatusChoices.QUEUED, - plugin__in=selected_plugins, - ) - Snapshot.objects.filter( - id__in=claimed_snapshot_ids, - status=Snapshot.StatusChoices.SEALED, - ).annotate( - has_queued_results=Exists(queued_results), - ).filter( - has_queued_results=False, - ).update( - retry_at=None, - modified_at=timezone.now(), - ) + queued_results = ArchiveResult.objects.filter( + snapshot_id=OuterRef("pk"), + status=ArchiveResult.StatusChoices.QUEUED, + plugin__in=selected_plugins, + ) + Snapshot.objects.filter( + id__in=claimed_snapshot_ids, + status=Snapshot.StatusChoices.SEALED, + ).annotate( + has_queued_results=Exists(queued_results), + ).filter( + has_queued_results=False, + ).update( + retry_at=None, + modified_at=timezone.now(), + ) return True @@ -2284,21 +2255,17 @@ def run_pending_crawls( ): continue - # Search backend selection is live crawl-execution config, not an - # installed-plugin list. Old queued rows for a backend that is disabled - # by the current Machine/Crawl/Snapshot config must remain queued so - # they can run if the user re-enables that backend, but they should not - # launch a standalone hook process just to skip after imports/config - # hydration. Refreshing here preserves mid-run config edits while using - # the same enabled-hook discovery path that created ArchiveResult rows. + # Plugin selection is live crawl-execution config, not an installed- + # plugin list. Old queued rows for a plugin that is disabled by the + # current Machine/Crawl/Snapshot config remain queued until the user + # re-enables it. Refresh here to preserve mid-run config edits while + # using the same enabled-hook discovery path that created the rows. runtime_config = get_config() catalog = get_plugin_catalog() enabled_plugins = get_enabled_plugins(config=runtime_config) - search_plugin_names = frozenset( - plugin.name for plugin, _hook in catalog.hooks("Snapshot", names=enabled_plugins) if plugin.name.startswith("search_backend_") - ) + queued_plugin_names = frozenset(plugin.name for plugin, _hook in catalog.hooks("Snapshot", names=enabled_plugins)) if _run_due_queued_plugin_result( - search_plugin_names, + queued_plugin_names, crawl_id=crawl_id, lock_seconds=60, interactive_interrupts=interactive_interrupts, @@ -2316,13 +2283,13 @@ def run_pending_crawls( retry_at__lte=timezone.now(), status=Snapshot.StatusChoices.SEALED, ) - if search_plugin_names: - queued_search_snapshot_ids = ArchiveResult.objects.filter( + if queued_plugin_names: + queued_plugin_snapshot_ids = ArchiveResult.objects.filter( status=ArchiveResult.StatusChoices.QUEUED, - plugin__in=search_plugin_names, + plugin__in=queued_plugin_names, ).values("snapshot_id") sealed_snapshots = sealed_snapshots.exclude( - id__in=queued_search_snapshot_ids, + id__in=queued_plugin_snapshot_ids, ) if crawl_id: sealed_snapshots = sealed_snapshots.filter(crawl_id=crawl_id) diff --git a/archivebox/tests/conftest.py b/archivebox/tests/conftest.py index 3250b71d..4432bc21 100644 --- a/archivebox/tests/conftest.py +++ b/archivebox/tests/conftest.py @@ -1660,6 +1660,79 @@ def resolve_abxpkg_binary_env( return {str(key): str(value) for key, value in payload.items()} +def run_test_hook( + script: Path, + output_dir: Path, + config: dict[str, Any] | None = None, + timeout: int = 60, + **arguments: Any, +): + """Execute a shipped finite hook through abx-dl and ArchiveBox's real DB projector.""" + import asyncio + + from abx_dl.execution import execute_hook + from abx_dl.models import discover_plugins + from abx_dl.orchestrator import create_bus + from archivebox.machine.models import Process + from archivebox.services.process_service import ProcessService, parse_event_datetime + + resolved_script = script.resolve() + hook = next( + ( + hook + for plugin in discover_plugins(runtime="archivebox").values() + for hook in plugin.hooks + if hook.path.resolve() == resolved_script + ), + None, + ) + assert hook is not None, f"shipped hook is not in the plugin catalog: {script}" + assert not hook.is_background, f"run_test_hook only supports finite hooks: {hook.full_name}" + + env = os.environ.copy() + for key, value in (config or {}).items(): + if value is None: + continue + if isinstance(value, bool): + env[key] = "true" if value else "false" + elif isinstance(value, (dict, list, tuple)): + env[key] = json.dumps(value) + else: + env[key] = str(value) + if env.get("NODE_MODULES_DIR"): + env.setdefault("NODE_MODULE_DIR", env["NODE_MODULES_DIR"]) + + bus = create_bus(name=f"test_hook_{hook.plugin_name}", total_timeout=float(timeout) + 30.0) + ProcessService(bus) + + async def execute_and_close(): + try: + return await execute_hook( + hook, + output_dir=output_dir, + env=env, + arguments=arguments, + timeout=timeout, + bus=bus, + process_type=Process.TypeChoices.HOOK, + ) + finally: + await bus.wait_until_idle() + await bus.destroy(clear=False) + + completed = asyncio.run(execute_and_close()) + process = ( + Process.objects.filter( + pid=completed.pid or None, + started_at=parse_event_datetime(completed.start_ts), + ) + .order_by("-modified_at") + .first() + ) + assert process is not None, f"hook completed without an ArchiveBox Process projection: {hook.full_name}" + return process + + def resolve_abxpkg_chrome_env(lib_dir: Path, env: dict[str, str] | None = None) -> dict[str, str]: from abx_plugins import get_plugins_dir diff --git a/archivebox/tests/test_cli_piping.py b/archivebox/tests/test_cli_piping.py index a274db3a..43648c96 100644 --- a/archivebox/tests/test_cli_piping.py +++ b/archivebox/tests/test_cli_piping.py @@ -38,7 +38,7 @@ PIPE_TEST_ENV = { def run_real_txt_parser(tmp_path, text): """Run the shipped text parser and return its real snapshot output directory.""" - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "parser-snapshot" staticfile_dir = snap_dir / "staticfile" @@ -47,7 +47,7 @@ def run_real_txt_parser(tmp_path, text): output_dir.mkdir(parents=True) (staticfile_dir / "input.txt").write_text(text, encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.parse_txt_urls").joinpath("on_Snapshot__71_parse_txt_urls.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, diff --git a/archivebox/tests/test_cli_run.py b/archivebox/tests/test_cli_run.py index 11db15be..0ba02e86 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -1695,13 +1695,11 @@ class TestRecoverOrchestratorState: Snapshot.objects.filter(pk=snapshot.pk).update(fs_version="0.9.0") snapshot.refresh_from_db() snapshot.output_dir.mkdir(parents=True, exist_ok=True) - title_dir = snapshot.output_dir / "title" - title_dir.mkdir(parents=True, exist_ok=True) - (title_dir / "title.txt").write_text("Example Domain\n", encoding="utf-8") + (snapshot.output_dir / "source.txt").write_text("real targeted maintenance input\n", encoding="utf-8") result = ArchiveResult.objects.create( snapshot=snapshot, - plugin="search_backend_sqlite", - hook_name="on_Snapshot__90_index_sqlite", + plugin="hashes", + hook_name="on_Snapshot__93_hashes.py", status=ArchiveResult.StatusChoices.QUEUED, ) @@ -2175,6 +2173,7 @@ class TestRunDueCrawlState: assert finished.output_str == "keep me" assert finished.output_files == {"favicon.ico": {"size": 1}} + @pytest.mark.django_db(transaction=True) def test_finished_parser_result_projects_children_before_resume_seals_snapshot(self): from importlib.resources import files from pathlib import Path @@ -2184,7 +2183,7 @@ class TestRunDueCrawlState: from archivebox.base_models.models import get_or_create_system_user_pk from archivebox.core.models import ArchiveResult, Snapshot from archivebox.crawls.models import Crawl - from archivebox.plugins.hooks import extract_records_from_process, run_hook + from archivebox.tests.conftest import run_test_hook from archivebox.services.runner import run_due_snapshot crawl = Crawl.objects.create( @@ -2210,7 +2209,7 @@ class TestRunDueCrawlState: encoding="utf-8", ) hook_path = Path(str(files("abx_plugins.plugins.parse_txt_urls").joinpath("on_Snapshot__71_parse_txt_urls.py"))) - process = run_hook( + process = run_test_hook( hook_path, parser_dir, config={"ABXPKG_LIB_DIR": str(root.output_dir.parent.parent / "lib"), "SNAP_DIR": str(root.output_dir)}, @@ -2221,12 +2220,12 @@ class TestRunDueCrawlState: ) process.refresh_from_db() assert process.exit_code == 0, process.stderr - result_record = next(record for record in extract_records_from_process(process) if record.get("type") == "ArchiveResult") + result_record = next(record for record in process.get_records() if record.get("type") == "ArchiveResult") ArchiveResult.objects.create( snapshot=root, process=process, - plugin=result_record["plugin"], - hook_name=result_record["hook_name"], + plugin="parse_txt_urls", + hook_name=hook_path.name, status=result_record["status"], output_str=result_record.get("output_str", ""), output_files={"urls.jsonl": {"size": (parser_dir / "urls.jsonl").stat().st_size}}, diff --git a/archivebox/tests/test_cli_update_reindex_snapshots.py b/archivebox/tests/test_cli_update_reindex_snapshots.py index e122bb58..b45db278 100644 --- a/archivebox/tests/test_cli_update_reindex_snapshots.py +++ b/archivebox/tests/test_cli_update_reindex_snapshots.py @@ -15,7 +15,7 @@ from archivebox.tests.test_orm_helpers import use_archivebox_db pytestmark = pytest.mark.django_db(transaction=True) -def test_only_sealed_search_backfill_bypasses_snapshot_lifecycle(): +def test_targeted_plugin_retries_preserve_sealed_snapshot_lifecycle(): from archivebox.base_models.models import get_or_create_system_user_pk from archivebox.cli.archivebox_extract import run_plugins from archivebox.core.models import ArchiveResult @@ -85,8 +85,8 @@ def test_only_sealed_search_backfill_bypasses_snapshot_lifecycle(): ) extract_crawl.refresh_from_db() extract_snapshot.refresh_from_db() - assert extract_crawl.status == Crawl.StatusChoices.QUEUED - assert extract_snapshot.status == Snapshot.StatusChoices.QUEUED + assert extract_crawl.status == Crawl.StatusChoices.SEALED + assert extract_snapshot.status == Snapshot.StatusChoices.SEALED assert extract_snapshot.archiveresult_set.filter( plugin="wget", status=ArchiveResult.StatusChoices.QUEUED, diff --git a/archivebox/tests/test_config_MAX_limits.py b/archivebox/tests/test_config_MAX_limits.py index 3fba65fc..b8e2cd33 100644 --- a/archivebox/tests/test_config_MAX_limits.py +++ b/archivebox/tests/test_config_MAX_limits.py @@ -67,7 +67,7 @@ def test_enqueue_discovered_snapshots_refreshes_crawl_limits(tmp_path): from archivebox.base_models.models import get_or_create_system_user_pk from archivebox.crawls.models import Crawl from archivebox.core.models import Snapshot - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook from archivebox.services.runner import CrawlRunner crawl = Crawl.objects.create( @@ -92,7 +92,7 @@ def test_enqueue_discovered_snapshots_refreshes_crawl_limits(tmp_path): encoding="utf-8", ) hook_path = Path(str(files("abx_plugins.plugins.parse_txt_urls").joinpath("on_Snapshot__71_parse_txt_urls.py"))) - process = run_hook( + process = run_test_hook( hook_path, parser_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, diff --git a/archivebox/tests/test_hooks.py b/archivebox/tests/test_hooks.py index 14fdd4f2..9e190c5c 100755 --- a/archivebox/tests/test_hooks.py +++ b/archivebox/tests/test_hooks.py @@ -80,14 +80,14 @@ class TestJSONLParsing: @staticmethod def run_hashes_hook(tmp_path): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "hash-snapshot" output_dir = snap_dir / "hashes" output_dir.mkdir(parents=True) (snap_dir / "source.txt").write_text("real parser input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, @@ -100,7 +100,7 @@ class TestJSONLParsing: @staticmethod def run_parser_hook(tmp_path): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "parser-snapshot" staticfile_dir = snap_dir / "staticfile" @@ -112,7 +112,7 @@ class TestJSONLParsing: encoding="utf-8", ) hook_path = Path(str(files("abx_plugins.plugins.parse_txt_urls").joinpath("on_Snapshot__71_parse_txt_urls.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, @@ -182,35 +182,6 @@ class TestJSONLParsing: assert len(records) == 1 assert records[0]["type"] == "ArchiveResult" - def test_direct_hook_preserves_explicit_parent_process(self, tmp_path): - """The compatibility adapter must retain the caller's process hierarchy.""" - from archivebox.machine.models import Machine, Process - from archivebox.plugins.hooks import run_hook - - parent = Process.objects.create( - machine=Machine.current(), - process_type=Process.TypeChoices.CLI, - status=Process.StatusChoices.RUNNING, - ) - snap_dir = tmp_path / "parented-snapshot" - output_dir = snap_dir / "hashes" - output_dir.mkdir(parents=True) - (snap_dir / "source.txt").write_text("parented hook input", encoding="utf-8") - hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - - process = run_hook( - hook_path, - output_dir, - config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, - timeout=30, - parent=parent, - url="https://example.com/parented-hook", - ) - - process.refresh_from_db() - assert process.exit_code == 0, process.stderr - assert process.parent_id == parent.id - class TestRequiredBinaryConfigHandling: """Test that required_binaries keep configured XYZ_BINARY values intact.""" @@ -446,7 +417,7 @@ class TestHookExecution: @pytest.mark.django_db(transaction=True) def test_real_js_hook_runs_through_abxpkg_shebang(self, tmp_path, hermetic_lib_dir): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook from archivebox.services.runner import run_install lib_dir = hermetic_lib_dir @@ -456,7 +427,7 @@ class TestHookExecution: snap_dir = crawl_dir / "snapshot" hook_path = Path(str(files("abx_plugins.plugins.chrome").joinpath("on_CrawlSetup__89_chrome_kill_zombies.js"))) - process = run_hook( + process = run_test_hook( hook_path, crawl_dir / "chrome", config={ @@ -551,7 +522,7 @@ class TestSnapshotHookOutput: [(True, "succeeded"), (False, "skipped")], ) def test_hashes_hook_emits_real_archive_result(self, tmp_path, enabled, expected_status): - from archivebox.plugins.hooks import extract_records_from_process, run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / f"snapshot-{expected_status}" output_dir = snap_dir / "hashes" @@ -559,7 +530,7 @@ class TestSnapshotHookOutput: (snap_dir / "source.txt").write_text("real hook protocol input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={ @@ -573,51 +544,21 @@ class TestSnapshotHookOutput: process.refresh_from_db() assert process.exit_code == 0, process.stderr - records = extract_records_from_process(process) + records = process.get_records() assert len(records) == 1 assert records[0]["type"] == "ArchiveResult" assert records[0]["status"] == expected_status - assert records[0]["plugin"] == "hashes" - assert records[0]["hook_name"] == hook_path.name - assert records[0]["plugin_hook"] == str(hook_path) + assert process.cmd[0] == str(hook_path) if enabled: assert (output_dir / "hashes.json").is_file() else: assert records[0]["output_str"] == "HASHES_ENABLED=False" -class TestPluginMetadata: - """Test that plugin metadata is added to JSONL records.""" - - @pytest.mark.django_db(transaction=True) - def test_python_hook_metadata_comes_from_executed_shipped_hook(self, tmp_path): - from archivebox.plugins.hooks import extract_records_from_process, run_hook - - snap_dir = tmp_path / "snapshot-metadata" - output_dir = snap_dir / "hashes" - output_dir.mkdir(parents=True) - (snap_dir / "source.txt").write_text("metadata", encoding="utf-8") - script = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( - script, - output_dir, - config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, - timeout=30, - url="https://example.com/metadata", - ) - process.refresh_from_db() - - assert process.exit_code == 0, process.stderr - records = extract_records_from_process(process) - assert records[0]["plugin"] == "hashes" - assert records[0]["hook_name"] == script.name - assert records[0]["plugin_hook"] == str(script) - - @pytest.mark.django_db(transaction=True) -def test_run_hook_exports_singular_node_modules_dir_with_colon_node_path(tmp_path, hermetic_lib_dir): +def test_abx_dl_hook_execution_exports_singular_node_modules_dir_with_colon_node_path(tmp_path, hermetic_lib_dir): """Hook subprocesses must get a real NODE_MODULES_DIR even when NODE_PATH has multiple entries.""" - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook from archivebox.services.runner import run_install lib_dir = hermetic_lib_dir @@ -632,7 +573,7 @@ def test_run_hook_exports_singular_node_modules_dir_with_colon_node_path(tmp_pat crawl_dir = tmp_path / "crawl" output_dir = crawl_dir / "chrome" hook_path = Path(str(files("abx_plugins.plugins.chrome").joinpath("on_CrawlSetup__89_chrome_kill_zombies.js"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={ @@ -654,16 +595,16 @@ def test_run_hook_exports_singular_node_modules_dir_with_colon_node_path(tmp_pat @pytest.mark.django_db(transaction=True) -def test_run_hook_executes_python_hooks_through_abxpkg_shebang(tmp_path): +def test_abx_dl_executes_python_hooks_through_abxpkg_shebang(tmp_path): """ArchiveBox treats Python hooks as opaque abxpkg-launched executables.""" - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "snapshot" output_dir = snap_dir / "hashes" output_dir.mkdir(parents=True) (snap_dir / "source.txt").write_text("real runtime hook input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={ diff --git a/archivebox/tests/test_process_runtime_paths.py b/archivebox/tests/test_process_runtime_paths.py index d3f3e22f..a8f8f533 100644 --- a/archivebox/tests/test_process_runtime_paths.py +++ b/archivebox/tests/test_process_runtime_paths.py @@ -11,14 +11,14 @@ pytestmark = pytest.mark.django_db(transaction=True) class TestProcessRuntimePaths: def test_hook_processes_use_isolated_runtime_dir(self, tmp_path): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "snapshot" output_dir = snap_dir / "hashes" output_dir.mkdir(parents=True) (snap_dir / "source.txt").write_text("real runtime path input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, diff --git a/archivebox/tests/test_search.py b/archivebox/tests/test_search.py index 651cb2b6..3c5ce1e2 100644 --- a/archivebox/tests/test_search.py +++ b/archivebox/tests/test_search.py @@ -118,8 +118,8 @@ def configure_ripgrep_search_backend(lib_dir: Path) -> None: ) -def test_search_backend_env_exposes_resolved_runtime_config(tmp_path): - from archivebox.search.backends import search_backend_env +def test_search_backend_command_env_serializes_config_without_mutating_process_env(tmp_path): + from archivebox.search.backends import search_backend_command_env old_env = os.environ.get("SEARCH_BACKEND_SONIC_HOST_NAME") os.environ["SEARCH_BACKEND_SONIC_HOST_NAME"] = "old-host" @@ -136,15 +136,14 @@ def test_search_backend_env_exposes_resolved_runtime_config(tmp_path): ) try: - with search_backend_env(config=config): - assert os.environ["SEARCH_BACKEND_ENGINE"] == "sonic" - assert os.environ["SEARCH_BACKEND_SONIC_HOST_NAME"] == "sonic" - assert os.environ["SEARCH_BACKEND_SONIC_PORT"] == "1491" - assert os.environ["SEARCH_BACKEND_SONIC_PASSWORD"] == "SecretPassword" - assert os.environ["RIPGREP_BINARY"] == str(tmp_path / "env" / "bin" / "rg") - assert "IGNORED_NONE_VALUE" not in os.environ - assert "UNRELATED_BINARY_NAME" not in os.environ - + env = search_backend_command_env(config=config) + assert env["SEARCH_BACKEND_ENGINE"] == "sonic" + assert env["SEARCH_BACKEND_SONIC_HOST_NAME"] == "sonic" + assert env["SEARCH_BACKEND_SONIC_PORT"] == "1491" + assert env["SEARCH_BACKEND_SONIC_PASSWORD"] == "SecretPassword" + assert env["RIPGREP_BINARY"] == str(tmp_path / "env" / "bin" / "rg") + assert "IGNORED_NONE_VALUE" not in env + assert env["UNRELATED_BINARY_NAME"] == "rg" assert os.environ["SEARCH_BACKEND_SONIC_HOST_NAME"] == "old-host" finally: if old_env is None: diff --git a/archivebox/tests/test_ui_admin_links.py b/archivebox/tests/test_ui_admin_links.py index 8da7d0b9..5498077f 100644 --- a/archivebox/tests/test_ui_admin_links.py +++ b/archivebox/tests/test_ui_admin_links.py @@ -15,13 +15,13 @@ from archivebox.tests.conftest import cli_env, run_archivebox_cmd from archivebox.tests.conftest import install_real_binary -pytestmark = pytest.mark.django_db +pytestmark = pytest.mark.django_db(transaction=True) @pytest.fixture def real_hook_result(tmp_path): from archivebox.core.models import ArchiveResult - from archivebox.plugins.hooks import extract_records_from_process, run_hook + from archivebox.tests.conftest import run_test_hook snapshot = _create_snapshot() snap_dir = Path(snapshot.output_dir) @@ -29,7 +29,7 @@ def real_hook_result(tmp_path): output_dir.mkdir(parents=True, exist_ok=True) (snap_dir / "source.txt").write_text("real admin link hook input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={ @@ -45,12 +45,12 @@ def real_hook_result(tmp_path): ) process.refresh_from_db() assert process.exit_code == 0, process.stderr - record = extract_records_from_process(process)[0] + record = process.get_records()[0] hashes_file = output_dir / "hashes.json" result = ArchiveResult.objects.create( snapshot=snapshot, - plugin=record["plugin"], - hook_name=record["hook_name"], + plugin="hashes", + hook_name=hook_path.name, process=process, status=record["status"], output_str=record["output_str"], diff --git a/archivebox/tests/test_ui_admin_machine.py b/archivebox/tests/test_ui_admin_machine.py index 45f8d80f..6a39ed01 100644 --- a/archivebox/tests/test_ui_admin_machine.py +++ b/archivebox/tests/test_ui_admin_machine.py @@ -19,14 +19,14 @@ pytestmark = pytest.mark.django_db(transaction=True) @pytest.fixture def real_exited_hook_process(tmp_path): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "snapshot" output_dir = snap_dir / "hashes" output_dir.mkdir(parents=True) (snap_dir / "source.txt").write_text("real admin hook input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, diff --git a/archivebox/tests/test_ui_live_progress.py b/archivebox/tests/test_ui_live_progress.py index faf886c3..3af12cd8 100644 --- a/archivebox/tests/test_ui_live_progress.py +++ b/archivebox/tests/test_ui_live_progress.py @@ -21,14 +21,14 @@ pytestmark = pytest.mark.django_db(transaction=True) @pytest.fixture def real_unscoped_hook_process(tmp_path): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = tmp_path / "snapshot" output_dir = snap_dir / "hashes" output_dir.mkdir(parents=True) (snap_dir / "source.txt").write_text("real live progress input", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.hashes").joinpath("on_Snapshot__93_hashes.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, @@ -55,7 +55,7 @@ def real_snapshot_hook_projection(snapshot, cached_abxpkg_lib_dir): @pytest.fixture def real_second_snapshot_hook_process(snapshot, tmp_path): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook snap_dir = Path(snapshot.output_dir) staticfile_dir = snap_dir / "staticfile" @@ -64,7 +64,7 @@ def real_second_snapshot_hook_process(snapshot, tmp_path): output_dir.mkdir(parents=True, exist_ok=True) (staticfile_dir / "input.txt").write_text("plain text without links", encoding="utf-8") hook_path = Path(str(files("abx_plugins.plugins.parse_txt_urls").joinpath("on_Snapshot__71_parse_txt_urls.py"))) - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={"ABXPKG_LIB_DIR": str(tmp_path / "lib"), "SNAP_DIR": str(snap_dir)}, @@ -78,7 +78,7 @@ def real_second_snapshot_hook_process(snapshot, tmp_path): @pytest.fixture def real_crawl_setup_process(snapshot, hermetic_lib_dir): - from archivebox.plugins.hooks import run_hook + from archivebox.tests.conftest import run_test_hook from archivebox.services.runner import run_install hook_path = Path(str(files("abx_plugins.plugins.chrome").joinpath("on_CrawlSetup__89_chrome_kill_zombies.js"))) @@ -86,7 +86,7 @@ def real_crawl_setup_process(snapshot, hermetic_lib_dir): run_install(plugin_names=["chrome"]) binary_env = resolve_abxpkg_binary_env(hermetic_lib_dir, deps_from=config_path) output_dir = Path(snapshot.crawl.output_dir) / "chrome" - process = run_hook( + process = run_test_hook( hook_path, output_dir, config={