ArchiveBox/archivebox/tests/test_recursive_crawl.py

#!/usr/bin/env python3
"""Integration tests for recursive crawling functionality."""

import json
import os
from pathlib import Path

import pytest

from archivebox.core.models import ArchiveResult, Snapshot
from archivebox.crawls.models import Crawl
from archivebox.machine.models import Binary, Process
from archivebox.tests.conftest import run_archivebox_cmd, cli_env
from archivebox.tests.test_orm_helpers import use_archivebox_db

pytestmark = pytest.mark.django_db(transaction=True)


def run_add_until(args, env, condition, timeout=120):
    assert args[0] == "archivebox"
    result = run_archivebox_cmd(
        args[1:],
        cwd=Path.cwd(),
        env=env,
        timeout=timeout,
    )
    assert result.returncode == 0, result.stderr or result.stdout
    with use_archivebox_db("."):
        assert condition(), f"Condition was false after command completed: {' '.join(args)}"
    return result.stdout, result.stderr


def test_background_hooks_dont_block_parser_extractors(tmp_path, initialized_archive, recursive_test_site):
    """Test that background hooks (.bg.) don't block other extractors from running."""

    # Verify the initialized_archive fixture prepared the expected data dir.
    assert initialized_archive == tmp_path
    assert (initialized_archive / "index.sqlite3").exists()

    # Enable only parser extractors and background hooks for this test
    env = os.environ.copy()
    env.update(
        {
            # Disable most extractors
            "SAVE_WGET": "false",
            "SAVE_SINGLEFILE": "false",
            "SAVE_READABILITY": "false",
            "SAVE_MERCURY": "false",
            "SAVE_HTMLTOTEXT": "false",
            "SAVE_PDF": "false",
            "SAVE_SCREENSHOT": "false",
            "SAVE_DOM": "false",
            "SAVE_HEADERS": "false",
            "SAVE_GIT": "false",
            "SAVE_YTDLP": "false",
            "SAVE_ARCHIVEDOTORG": "false",
            "SAVE_TITLE": "false",
            "SAVE_FAVICON": "true",
        },
    )

    stdout, stderr = run_add_until(
        ["archivebox", "add", "--depth=1", "--plugins=favicon,parse_html_urls", recursive_test_site["root_url"]],
        env=env,
        timeout=120,
        condition=lambda: ArchiveResult.objects.filter(
            plugin__startswith="parse_",
            plugin__endswith="_urls",
            status__in=("started", "succeeded", "failed"),
        ).exists(),
    )

    if stderr:
        print(f"\n=== STDERR ===\n{stderr}\n=== END STDERR ===\n")
    if stdout:
        print(f"\n=== STDOUT (last 2000 chars) ===\n{stdout[-2000:]}\n=== END STDOUT ===\n")

    with use_archivebox_db(tmp_path):
        snapshots = list(Snapshot.objects.values_list("url", "depth", "status"))
        bg_hooks = list(
            ArchiveResult.objects.filter(plugin__in=("favicon", "consolelog", "ssl", "responses", "redirects", "staticfile"))
            .order_by("plugin")
            .values_list("plugin", "status"),
        )
        parser_extractors = list(
            ArchiveResult.objects.filter(plugin__startswith="parse_", plugin__endswith="_urls")
            .order_by("plugin")
            .values_list("plugin", "status"),
        )
        all_extractors = list(ArchiveResult.objects.order_by("plugin").values_list("plugin", "status"))

    assert len(snapshots) > 0, (
        f"Should have created snapshot after Crawl hooks finished. "
        f"If this fails, Crawl hooks may be taking too long. "
        f"Snapshots: {snapshots}"
    )

    assert len(all_extractors) > 0, (
        f"Should have extractors created for snapshot. If this fails, Snapshot.run() may not have started. Got: {all_extractors}"
    )

    parser_statuses = [status for _, status in parser_extractors]
    assert "started" in parser_statuses or "succeeded" in parser_statuses or "failed" in parser_statuses, (
        f"Parser extractors should have run, got statuses: {parser_statuses}. Background hooks: {bg_hooks}"
    )


def test_parser_extractors_emit_snapshot_jsonl(tmp_path, initialized_archive, recursive_test_site):
    """Test that parser extractors emit Snapshot JSONL to stdout."""

    env = os.environ.copy()
    env.update(
        {
            "SAVE_WGET": "false",
            "SAVE_SINGLEFILE": "false",
            "SAVE_READABILITY": "false",
            "SAVE_MERCURY": "false",
            "SAVE_HTMLTOTEXT": "false",
            "SAVE_PDF": "false",
            "SAVE_SCREENSHOT": "false",
            "SAVE_DOM": "false",
            "SAVE_HEADERS": "false",
            "SAVE_GIT": "false",
            "SAVE_YTDLP": "false",
            "SAVE_ARCHIVEDOTORG": "false",
            "SAVE_TITLE": "false",
            "SAVE_FAVICON": "false",
            "USE_CHROME": "false",
        },
    )

    result = run_archivebox_cmd(
        ["add", "--depth=0", "--plugins=wget,parse_html_urls", recursive_test_site["root_url"]],
        env=env,
        timeout=60,
    )
    assert result.returncode == 0, result.stderr

    with use_archivebox_db(tmp_path):
        parse_html = (
            ArchiveResult.objects.filter(plugin__endswith="parse_html_urls")
            .order_by("id")
            .values_list("id", "status", "output_str")
            .first()
        )

    if parse_html:
        status = parse_html[1]
        output = parse_html[2] or ""

        assert status in ["started", "succeeded", "failed"], f"60_parse_html_urls should have run, got status: {status}"

        if status == "succeeded" and output:
            assert "parsed" in output.lower(), "Parser summary should report parsed URLs"

    urls_jsonl_files = list(Path("archive/users/system/snapshots").rglob("parse_html_urls/**/urls.jsonl"))
    assert urls_jsonl_files, "parse_html_urls should write urls.jsonl output"

    records = []
    for line in urls_jsonl_files[0].read_text().splitlines():
        if line.strip():
            records.append(json.loads(line))

    assert records, "urls.jsonl should contain parsed Snapshot records"
    assert all(record.get("type") == "Snapshot" for record in records), f"Expected Snapshot JSONL records, got: {records}"


def test_recursive_crawl_creates_child_snapshots(tmp_path, initialized_archive, recursive_test_site):
    """Test that recursive crawling creates child snapshots with proper depth and parent_snapshot_id."""

    env = os.environ.copy()
    env.update(
        {
            "URL_ALLOWLIST": r"127\.0\.0\.1[:/].*",
            "SAVE_READABILITY": "false",
            "SAVE_SINGLEFILE": "false",
            "SAVE_MERCURY": "false",
            "SAVE_SCREENSHOT": "false",
            "SAVE_PDF": "false",
            "SAVE_HEADERS": "false",
            "SAVE_ARCHIVEDOTORG": "false",
            "SAVE_GIT": "false",
            "SAVE_YTDLP": "false",
            "SAVE_TITLE": "false",
        },
    )

    stdout, stderr = run_add_until(
        ["archivebox", "add", "--depth=1", "--plugins=wget,parse_html_urls", recursive_test_site["root_url"]],
        env=env,
        timeout=120,
        condition=lambda: (
            Snapshot.objects.filter(depth=0).count() >= 1
            and Snapshot.objects.filter(depth=1).count() >= len(recursive_test_site["child_urls"])
        ),
    )

    if stderr:
        print(f"\n=== STDERR ===\n{stderr}\n=== END STDERR ===\n")
    if stdout:
        print(f"\n=== STDOUT (last 2000 chars) ===\n{stdout[-2000:]}\n=== END STDOUT ===\n")

    with use_archivebox_db(tmp_path):
        all_snapshots = list(Snapshot.objects.values_list("url", "depth"))
        root_snapshot = (
            Snapshot.objects.filter(depth=0).order_by("created_at").values_list("id", "url", "depth", "parent_snapshot_id").first()
        )
        child_snapshots = list(Snapshot.objects.filter(depth=1).values_list("id", "url", "depth", "parent_snapshot_id"))
        crawl = Crawl.objects.order_by("-created_at").values_list("id", "max_depth").first()
        parser_status = list(
            ArchiveResult.objects.filter(
                snapshot_id=root_snapshot[0] if root_snapshot else None,
                plugin__startswith="parse_",
                plugin__endswith="_urls",
            ).values_list("plugin", "status"),
        )
        started_extractors = list(
            ArchiveResult.objects.filter(
                snapshot_id=root_snapshot[0] if root_snapshot else None,
                status="started",
            ).values_list("plugin", "status"),
        )

    assert root_snapshot is not None, f"Root snapshot should exist at depth=0. All snapshots: {all_snapshots}"
    root_id = root_snapshot[0]

    assert crawl is not None, "Crawl should be created"
    assert crawl[1] == 1, f"Crawl max_depth should be 1, got {crawl[1]}"

    assert len(child_snapshots) > 0, (
        f"Child snapshots should be created from monadical.com links. Parser status: {parser_status}. Started extractors blocking: {started_extractors}"
    )

    for child_id, child_url, child_depth, parent_id in child_snapshots:
        assert child_depth == 1, f"Child snapshot should have depth=1, got {child_depth}"
        assert parent_id == root_id, f"Child snapshot {child_url} should have parent_snapshot_id={root_id}, got {parent_id}"


def test_recursive_crawl_respects_depth_limit(tmp_path, initialized_archive, recursive_test_site):
    """Test that recursive crawling stops at max_depth."""
    env = cli_env(disable_extractors=True)

    env = env.copy()
    env["URL_ALLOWLIST"] = r"127\.0\.0\.1[:/].*"

    stdout, stderr = run_add_until(
        ["archivebox", "add", "--depth=1", "--plugins=wget,parse_html_urls", recursive_test_site["root_url"]],
        env=env,
        timeout=120,
        condition=lambda: (
            Snapshot.objects.filter(depth=0).count() >= 1
            and Snapshot.objects.filter(depth=1).count() >= len(recursive_test_site["child_urls"])
            and ArchiveResult.objects.filter(
                snapshot__depth=1,
                plugin__startswith="parse_",
                plugin__endswith="_urls",
                status__in=("started", "succeeded", "failed"),
            )
            .values("snapshot_id")
            .distinct()
            .count()
            >= len(recursive_test_site["child_urls"])
        ),
    )

    with use_archivebox_db(tmp_path):
        depths = list(Snapshot.objects.values_list("depth", flat=True))
        max_depth_found = max(depths) if depths else None
        depth_counts = [(depth, Snapshot.objects.filter(depth=depth).count()) for depth in sorted(set(depths))]

    assert max_depth_found is not None, "Should have at least one snapshot"
    assert max_depth_found <= 1, f"Max depth should not exceed 1, got {max_depth_found}. Depth distribution: {depth_counts}"


def test_recursive_crawl_depth_two_writes_real_outputs_and_process_records(tmp_path, initialized_archive, recursive_test_site):
    """Run a real depth=2 crawl and verify DB, output files, and process side effects."""

    env = os.environ.copy()
    env.update(
        {
            "URL_ALLOWLIST": r"127\.0\.0\.1[:/].*",
            "SAVE_WGET": "true",
            "SAVE_READABILITY": "false",
            "SAVE_SINGLEFILE": "false",
            "SAVE_MERCURY": "false",
            "SAVE_SCREENSHOT": "false",
            "SAVE_PDF": "false",
            "SAVE_HEADERS": "false",
            "SAVE_ARCHIVEDOTORG": "false",
            "SAVE_GIT": "false",
            "SAVE_YTDLP": "false",
            "SAVE_TITLE": "false",
            "SAVE_FAVICON": "false",
            "USE_CHROME": "false",
            "USE_COLOR": "false",
            "SHOW_PROGRESS": "false",
        },
    )

    result = run_archivebox_cmd(
        ["add", "--depth=2", "--plugins=wget,parse_html_urls", recursive_test_site["root_url"]],
        cwd=initialized_archive,
        env=env,
        timeout=240,
    )
    stdout, stderr = result.stdout, result.stderr

    if stderr:
        print(f"\n=== STDERR ===\n{stderr}\n=== END STDERR ===\n")
    if stdout:
        print(f"\n=== STDOUT (last 2000 chars) ===\n{stdout[-2000:]}\n=== END STDOUT ===\n")
    assert result.returncode == 0, stderr or stdout

    with use_archivebox_db(tmp_path):
        depths = list(Snapshot.objects.values_list("depth", flat=True))
        depth_counts = {depth: Snapshot.objects.filter(depth=depth).count() for depth in sorted(set(depths))}
        crawl = Crawl.objects.order_by("-created_at").values_list("id", "max_depth").first()
        root_snapshot = (
            Snapshot.objects.filter(depth=0).order_by("created_at").values_list("id", "url", "depth", "parent_snapshot_id").first()
        )
        child_rows = list(Snapshot.objects.filter(depth=1).values_list("id", "url", "parent_snapshot_id"))
        deep_rows = list(Snapshot.objects.filter(depth=2).values_list("id", "url", "parent_snapshot_id"))
        parser_results = list(
            ArchiveResult.objects.filter(plugin__startswith="parse_", plugin__endswith="_urls")
            .order_by("snapshot__depth", "snapshot__url")
            .values_list("snapshot__url", "snapshot__depth", "plugin", "status", "output_files", "output_size"),
        )
        wget_results = list(
            ArchiveResult.objects.filter(plugin="wget")
            .order_by("snapshot__depth", "snapshot__url")
            .values_list("snapshot__url", "snapshot__depth", "status", "output_files", "output_size"),
        )
        process_rows = list(
            Process.objects.filter(process_type="hook")
            .order_by("created_at")
            .values_list("process_type", "worker_type", "status", "exit_code", "pwd", "cmd"),
        )

    assert crawl is not None
    assert crawl[1] == 2
    assert root_snapshot is not None
    assert root_snapshot[2] == 0
    assert root_snapshot[3] is None
    assert depth_counts.get(0, 0) >= 1
    assert depth_counts.get(1, 0) >= len(recursive_test_site["child_urls"])
    assert depth_counts.get(2, 0) >= len(recursive_test_site["deep_urls"])
    assert max(depth_counts) <= 2

    child_urls = {row[1] for row in child_rows}
    deep_urls = {row[1] for row in deep_rows}
    child_ids = {row[0] for row in child_rows}
    assert set(recursive_test_site["child_urls"]).issubset(child_urls)
    assert set(recursive_test_site["deep_urls"]).issubset(deep_urls)
    assert all(parent_id == root_snapshot[0] for _id, _url, parent_id in child_rows)
    assert all(parent_id in child_ids for _id, _url, parent_id in deep_rows)

    parser_statuses = {status for _url, _depth, _plugin, status, _files, _size in parser_results}
    wget_statuses = {status for _url, _depth, status, _files, _size in wget_results}
    assert parser_results
    assert wget_results
    assert "succeeded" in parser_statuses
    assert "succeeded" in wget_statuses
    assert len([row for row in parser_results if row[3] == "failed"]) <= 2
    assert len([row for row in wget_results if row[2] == "failed"]) <= 2

    urls_jsonl_files = list(Path("archive/users/system/snapshots").rglob("parse_html_urls/**/urls.jsonl"))
    assert urls_jsonl_files, "parse_html_urls should write urls.jsonl files"
    parsed_urls = set()
    for path in urls_jsonl_files:
        for line in path.read_text().splitlines():
            if line.strip():
                parsed_urls.add(json.loads(line)["url"])
    assert set(recursive_test_site["child_urls"]).issubset(parsed_urls)
    assert set(recursive_test_site["deep_urls"]).issubset(parsed_urls)

    snapshot_dirs = [path.parent for path in Path("archive/users/system/snapshots").rglob("index.jsonl")]
    assert snapshot_dirs
    for snapshot_dir in snapshot_dirs:
        assert (snapshot_dir / "index.jsonl").exists()

    assert process_rows
    assert any("parse_html_urls" in (pwd or "") or "parse_html_urls" in (cmd or "") for *_rest, pwd, cmd in process_rows)
    assert any("wget" in (pwd or "") or "wget" in (cmd or "") for *_rest, pwd, cmd in process_rows)


@pytest.mark.timeout(1200)
def test_add_archivewebpage_installs_required_chrome_dependency(initialized_archive):
    """archivebox add should resolve selected plugins' required plugins and persist binary projections."""

    env = os.environ.copy()
    env.pop("CHROME_BINARY", None)
    env.update(
        {
            "USE_COLOR": "false",
            "SHOW_PROGRESS": "false",
            "TIMEOUT": "120",
            "ABXPKG_LIB_DIR": str(initialized_archive / "lib"),
            "CHROME_HEADLESS": "true",
            "CHROME_SANDBOX": "false",
            "CHROME_ISOLATION": "snapshot",
            "CHROMEWEBSTORE_EXTENSIONS_DIR": str(initialized_archive / "lib/chromewebstore/extensions"),
        },
    )

    result = run_archivebox_cmd(
        [
            "add",
            "--depth=0",
            "--max-urls=1",
            "--tag=archivewebpage-required-plugin-preflight",
            "--parser=url_list",
            "--plugins=archivewebpage",
            "https://example.com/",
        ],
        cwd=initialized_archive,
        env=env,
        timeout=1200,
    )
    stdout, stderr = result.stdout, result.stderr

    if stderr:
        print(f"\n=== STDERR ===\n{stderr}\n=== END STDERR ===\n")
    if stdout:
        print(f"\n=== STDOUT (last 4000 chars) ===\n{stdout[-4000:]}\n=== END STDOUT ===\n")
    assert result.returncode == 0, stderr or stdout

    with use_archivebox_db(initialized_archive):
        binaries = {
            row["name"]: row for row in Binary.objects.order_by("name").values("name", "status", "binprovider", "abspath", "version")
        }
        archive_results = list(
            ArchiveResult.objects.order_by("plugin", "hook_name").values_list(
                "plugin",
                "hook_name",
                "status",
                "output_str",
                "output_files",
            ),
        )
        process_rows = list(
            Process.objects.order_by("process_type", "created_at").values_list("process_type", "status", "exit_code", "cmd", "env"),
        )
        snapshot_output_dirs = [snapshot.output_dir for snapshot in Snapshot.objects.order_by("created_at")]

    assert "chromium" in binaries
    assert binaries["chromium"]["status"] == Binary.StatusChoices.INSTALLED
    assert Path(binaries["chromium"]["abspath"]).exists()
    chromium_version_parts = [int(part) for part in binaries["chromium"]["version"].split(".")[:3]]
    assert chromium_version_parts >= [149, 0, 0]

    assert "archivewebpage" in binaries
    assert binaries["archivewebpage"]["status"] == Binary.StatusChoices.INSTALLED
    assert binaries["archivewebpage"]["binprovider"] == "chromewebstore"
    archivewebpage_metadata = Path(binaries["archivewebpage"]["abspath"])
    assert archivewebpage_metadata.exists()
    assert archivewebpage_metadata.name == "archivewebpage.extension.json"
    archivewebpage_extension = json.loads(archivewebpage_metadata.read_text(encoding="utf-8"))
    archivewebpage_manifest = Path(archivewebpage_extension["unpacked_path"]) / "manifest.json"
    assert archivewebpage_manifest.exists()
    assert json.loads(archivewebpage_manifest.read_text(encoding="utf-8"))["version"] == binaries["archivewebpage"]["version"]

    plugins_seen = {plugin for plugin, _hook_name, _status, _output_str, _output_files in archive_results}
    assert {"chrome", "archivewebpage"}.issubset(plugins_seen)
    assert all(
        status == ArchiveResult.StatusChoices.SUCCEEDED
        for plugin, _hook_name, status, _output_str, _output_files in archive_results
        if plugin in {"chrome", "archivewebpage"}
    ), archive_results
    assert snapshot_output_dirs
    archivewebpage_wacz = Path(snapshot_output_dirs[0]) / "archivewebpage" / "archivewebpage.wacz"
    assert archivewebpage_wacz.exists()
    assert archivewebpage_wacz.stat().st_size > 0
    chrome_hook_envs = [
        env
        for process_type, _status, _exit_code, cmd, env in process_rows
        if process_type == Process.TypeChoices.HOOK and "chrome_launch" in str(cmd)
    ]
    assert chrome_hook_envs
    assert all("{ABXPKG_LIB_DIR}" not in str(env) for env in chrome_hook_envs)
    assert any(process_type == Process.TypeChoices.BINARY for process_type, _status, _exit_code, _cmd, _env in process_rows)
    assert all(
        status == Process.StatusChoices.EXITED and exit_code == 0
        for process_type, status, exit_code, _cmd, _env in process_rows
        if process_type == Process.TypeChoices.BINARY
    )


@pytest.mark.timeout(1200)
def test_recursive_crawl_depth_two_all_plugins_runs_snapshots_in_parallel(
    initialized_archive,
    free_tcp_port_factory,
    recursive_test_site,
):
    """Run a bounded real depth=2 crawl with all plugins enabled and verify parallel snapshot execution."""

    from archivebox.plugins.discovery import get_plugin_catalog

    root_url = recursive_test_site["root_url"]
    plugin_selection = ",".join(sorted(plugin for plugin in get_plugin_catalog() if not plugin.startswith("claude")))
    env = os.environ.copy()
    for preinstalled_path_key in (
        "CHROME_BINARY",
        "LIB_DIR",
        "DATA_DIR",
        "NODE_MODULES_DIR",
        "NODE_PATH",
        "PNPM_BIN_DIR",
        "NPM_BIN_DIR",
        "CHROMEWEBSTORE_EXTENSIONS_DIR",
    ):
        env.pop(preinstalled_path_key, None)
    env.update(
        {
            "USE_COLOR": "false",
            "SHOW_PROGRESS": "false",
            "URL_ALLOWLIST": r"127\.0\.0\.1[:/].*",
            "ABXPKG_LIB_DIR": str(initialized_archive / "lib"),
            "CHROMEWEBSTORE_EXTENSIONS_DIR": str(initialized_archive / "lib/chromewebstore/extensions"),
            "TIMEOUT": "90",
            "CRAWL_MAX_CONCURRENT_SNAPSHOTS": "3",
            "SEARCH_BACKEND_SONIC_HOST_NAME": "127.0.0.1",
            "SEARCH_BACKEND_SONIC_PORT": str(free_tcp_port_factory()),
            "CHROME_HEADLESS": "true",
            "CHROME_SANDBOX": "false",
            "CHROME_ISOLATION": "snapshot",
            "LITEPARSE_OCR_ENABLED": "false",
            "LITEPARSE_MAX_SOURCES": "4",
        },
    )

    result = run_archivebox_cmd(
        [
            "add",
            "--depth=2",
            "--max-urls=8",
            "--crawl-max-size=100mb",
            "--tag=recursive-all-plugins",
            "--parser=url_list",
            f"--plugins={plugin_selection}",
            root_url,
        ],
        cwd=initialized_archive,
        env=env,
        timeout=1200,
    )
    stdout, stderr = result.stdout, result.stderr

    if stderr:
        print(f"\n=== STDERR ===\n{stderr}\n=== END STDERR ===\n")
    if stdout:
        print(f"\n=== STDOUT (last 4000 chars) ===\n{stdout[-4000:]}\n=== END STDOUT ===\n")
    assert result.returncode == 0, stderr or stdout

    with use_archivebox_db(initialized_archive):
        crawl = Crawl.objects.get(tags_str="recursive-all-plugins")
        snapshots = list(
            Snapshot.objects.filter(crawl=crawl)
            .order_by("depth", "url")
            .values_list("id", "url", "depth", "status", "parent_snapshot_id", "downloaded_at"),
        )
        archive_results = list(
            ArchiveResult.objects.filter(snapshot__crawl=crawl)
            .select_related("snapshot")
            .order_by("snapshot__depth", "snapshot__url", "plugin", "hook_name")
            .values_list(
                "snapshot_id",
                "snapshot__url",
                "snapshot__depth",
                "plugin",
                "hook_name",
                "status",
                "output_files",
                "output_size",
                "output_str",
            ),
        )
        process_snapshot_ids = {
            process_id: str(snapshot_id)
            for snapshot_id, process_id in ArchiveResult.objects.filter(
                snapshot__crawl=crawl,
                process_id__isnull=False,
            ).values_list("snapshot_id", "process_id")
        }
        processes = list(
            Process.objects.filter(process_type=Process.TypeChoices.HOOK, id__in=process_snapshot_ids)
            .order_by("started_at")
            .values_list("id", "pwd", "cmd", "status", "exit_code", "started_at", "ended_at"),
        )

    assert crawl.max_depth == 2
    assert crawl.config["CRAWL_MAX_URLS"] == 8
    assert crawl.config["CRAWL_MAX_SIZE"] == 100 * 1024 * 1024
    assert crawl.config["CRAWL_MAX_CONCURRENT_SNAPSHOTS"] == 3
    assert crawl.status == Crawl.StatusChoices.SEALED
    assert crawl.retry_at is None

    expected_urls = {
        root_url,
        *recursive_test_site["child_urls"],
        *recursive_test_site["deep_urls"],
    }
    snapshots_by_url = {url: (snapshot_id, depth, parent_id) for snapshot_id, url, depth, _status, parent_id, _downloaded_at in snapshots}
    assert set(snapshots_by_url) == expected_urls
    root_id, root_depth, root_parent_id = snapshots_by_url[root_url]
    assert root_depth == 0
    assert root_parent_id is None
    child_ids_by_url = {}
    for child_url in recursive_test_site["child_urls"]:
        child_id, child_depth, child_parent_id = snapshots_by_url[child_url]
        assert child_depth == 1
        assert child_parent_id == root_id
        child_ids_by_url[child_url] = child_id
    for child_url, deep_url in zip(recursive_test_site["child_urls"], recursive_test_site["deep_urls"], strict=True):
        _deep_id, deep_depth, deep_parent_id = snapshots_by_url[deep_url]
        assert deep_depth == 2
        assert deep_parent_id == child_ids_by_url[child_url]
    assert all(status == Snapshot.StatusChoices.SEALED for _id, _url, _depth, status, _parent, _downloaded_at in snapshots)
    assert all(downloaded_at is not None for _id, _url, _depth, _status, _parent, downloaded_at in snapshots)

    assert archive_results
    allowed_statuses = {
        ArchiveResult.StatusChoices.SUCCEEDED,
        ArchiveResult.StatusChoices.NORESULTS,
        ArchiveResult.StatusChoices.SKIPPED,
    }
    unexpected_results = [
        {
            "url": url,
            "depth": depth,
            "plugin": plugin,
            "hook_name": hook_name,
            "status": status,
            "output_str": output_str,
        }
        for _snapshot_id, url, depth, plugin, hook_name, status, _files, _size, output_str in archive_results
        if not (status in allowed_statuses or (plugin == "archivedotorg" and status == ArchiveResult.StatusChoices.FAILED))
    ]
    assert not unexpected_results
    ytdlp_results = [
        (url, status, output_str)
        for _snapshot_id, url, _depth, plugin, _hook_name, status, _files, _size, output_str in archive_results
        if plugin == "ytdlp"
    ]
    assert ytdlp_results
    assert all(status != ArchiveResult.StatusChoices.FAILED for _url, status, _output_str in ytdlp_results), ytdlp_results

    plugins_seen = {plugin for _snapshot_id, _url, _depth, plugin, _hook_name, _status, _files, _size, _output in archive_results}
    assert {
        "wget",
        "headers",
        "title",
        "pdf",
        "screenshot",
        "dom",
        "singlefile",
        "readability",
        "mercury",
        "htmltotext",
        "favicon",
        "parse_html_urls",
        "archivedotorg",
    }.issubset(plugins_seen)

    snapshot_root = initialized_archive / "archive/users/system/snapshots"
    assert list(snapshot_root.rglob("wget/**/*.html"))
    assert list(snapshot_root.rglob("headers/**/headers.json"))
    assert list(snapshot_root.rglob("title/title.txt"))
    assert list(snapshot_root.rglob("pdf/**/*.pdf"))
    assert list(snapshot_root.rglob("screenshot/**/*.png"))
    assert list(snapshot_root.rglob("dom/**/*.html"))
    assert list(snapshot_root.rglob("singlefile/**/*.html"))
    assert list(snapshot_root.rglob("readability/**/*.html"))
    assert list(snapshot_root.rglob("mercury/**/*.html"))
    assert list(snapshot_root.rglob("htmltotext/**/*.txt"))
    assert list(snapshot_root.rglob("favicon/**/*"))
    urls_jsonl_files = list(snapshot_root.rglob("parse_html_urls/urls.jsonl"))
    assert urls_jsonl_files
    parsed_urls = {
        json.loads(line)["url"] for path in urls_jsonl_files for line in path.read_text(errors="ignore").splitlines() if line.strip()
    }
    assert set(recursive_test_site["child_urls"]).issubset(parsed_urls)
    assert set(recursive_test_site["deep_urls"]).issubset(parsed_urls)

    assert processes
    failed_hook_results = [
        {
            "url": url,
            "depth": depth,
            "plugin": plugin,
            "hook_name": hook_name,
            "status": status,
            "output_str": output_str,
        }
        for _snapshot_id, url, depth, plugin, hook_name, status, _files, _size, output_str in archive_results
        if status == ArchiveResult.StatusChoices.FAILED and plugin != "archivedotorg"
    ]
    assert not failed_hook_results
    assert all(status == Process.StatusChoices.EXITED for _id, _pwd, _cmd, status, _exit_code, _started_at, _ended_at in processes)

    intervals = []
    for process_id, pwd, cmd, _status, _exit_code, started_at, ended_at in processes:
        if not started_at or not ended_at:
            continue
        process_snapshot_id = process_snapshot_ids.get(process_id)
        if process_snapshot_id is None:
            continue
        intervals.append((process_snapshot_id, started_at, ended_at, pwd, cmd))

    overlapping = [
        (left, right)
        for index, left in enumerate(intervals)
        for right in intervals[index + 1 :]
        if left[0] != right[0] and left[1] < right[2] and right[1] < left[2]
    ]
    assert overlapping, f"Expected hook processes from different snapshots to overlap, got intervals: {intervals}"


def test_crawl_snapshot_has_parent_snapshot_field(tmp_path, initialized_archive):
    """Test that Snapshot model has parent_snapshot field."""

    column_names = {field.column for field in Snapshot._meta.local_fields}

    assert "parent_snapshot_id" in column_names, f"Snapshot table should have parent_snapshot_id column. Columns: {column_names}"


def test_snapshot_depth_field_exists(tmp_path, initialized_archive):
    """Test that Snapshot model has depth field."""

    column_names = {field.column for field in Snapshot._meta.local_fields}

    assert "depth" in column_names, f"Snapshot table should have depth column. Columns: {column_names}"


def test_root_snapshot_has_depth_zero(tmp_path, initialized_archive, recursive_test_site):
    """Test that root snapshots are created with depth=0."""
    env = cli_env(disable_extractors=True)

    env = env.copy()
    env["URL_ALLOWLIST"] = r"127\.0\.0\.1[:/].*"

    stdout, stderr = run_add_until(
        ["archivebox", "add", "--depth=1", "--plugins=wget,parse_html_urls", recursive_test_site["root_url"]],
        env=env,
        timeout=120,
        condition=lambda: Snapshot.objects.filter(url=recursive_test_site["root_url"]).count() >= 1,
    )

    with use_archivebox_db(tmp_path):
        snapshot = Snapshot.objects.filter(url=recursive_test_site["root_url"]).order_by("created_at").values_list("id", "depth").first()

    assert snapshot is not None, "Root snapshot should be created"
    assert snapshot[1] == 0, f"Root snapshot should have depth=0, got {snapshot[1]}"


def test_archiveresult_worker_queue_filters_by_foreground_extractors(tmp_path, initialized_archive, recursive_test_site):
    """Test that background hooks don't block foreground extractors from running."""

    env = os.environ.copy()
    env.update(
        {
            "SAVE_WGET": "true",
            "SAVE_SINGLEFILE": "false",
            "SAVE_PDF": "false",
            "SAVE_SCREENSHOT": "false",
            "SAVE_FAVICON": "true",
        },
    )

    stdout, stderr = run_add_until(
        ["archivebox", "add", "--plugins=favicon,wget,parse_html_urls", recursive_test_site["root_url"]],
        env=env,
        timeout=120,
        condition=lambda: ArchiveResult.objects.filter(
            plugin__startswith="parse_",
            plugin__endswith="_urls",
            status__in=("started", "succeeded", "failed"),
        ).exists(),
    )

    with use_archivebox_db(tmp_path):
        bg_results = list(
            ArchiveResult.objects.filter(
                plugin__in=("favicon", "consolelog", "ssl", "responses", "redirects", "staticfile"),
                status__in=("started", "succeeded", "failed"),
            ).values_list("plugin", "status"),
        )
        parser_status = list(
            ArchiveResult.objects.filter(plugin__startswith="parse_", plugin__endswith="_urls").values_list("plugin", "status"),
        )

    if len(bg_results) > 0:
        parser_statuses = [status for _, status in parser_status]
        non_queued = [s for s in parser_statuses if s != "queued"]
        assert len(non_queued) > 0 or len(parser_status) == 0, (
            f"With {len(bg_results)} background hooks started, parser extractors should still run. Got statuses: {parser_statuses}"
        )


if __name__ == "__main__":
    pytest.main([__file__, "-v"])