mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-12 19:50:57 +05:00
628 lines
23 KiB
Python
628 lines
23 KiB
Python
"""
|
|
Tests for archivebox list command.
|
|
Verify list emits snapshot JSONL and applies the documented filters.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from django.contrib.auth import get_user_model
|
|
from django.db import connection
|
|
from django.utils import timezone
|
|
|
|
from archivebox.core.models import Snapshot
|
|
from archivebox.cli.archivebox_snapshot import iter_snapshot_json
|
|
from archivebox.tests.conftest import create_test_url, parse_jsonl_output, run_archivebox_cmd, run_queued_crawls, cli_env
|
|
|
|
from archivebox.tests.test_orm_helpers import use_archivebox_db
|
|
|
|
pytestmark = pytest.mark.django_db(transaction=True)
|
|
|
|
|
|
def test_static_export_creates_detail_page_for_unarchived_snapshot(snapshot):
|
|
from archivebox.config import CONSTANTS
|
|
|
|
snapshot_dir = Path(snapshot.output_dir)
|
|
assert snapshot_dir.is_dir()
|
|
assert not any(snapshot_dir.iterdir())
|
|
snapshot_dir.rmdir()
|
|
assert not snapshot_dir.exists()
|
|
|
|
html = Snapshot.objects.filter(pk=snapshot.pk).to_html(with_headers=True)
|
|
|
|
static_path = snapshot_dir.relative_to(CONSTANTS.DATA_DIR).as_posix()
|
|
detail_path = snapshot_dir / "index.html"
|
|
assert f"./{static_path}/index.html" in html
|
|
# The portable export command emits JSONL records; index.json is a legacy
|
|
# filename that is not created and leaves a broken footer link offline.
|
|
assert 'href="./index.jsonl"' in html
|
|
assert 'href="./index.json"' not in html
|
|
root_manifest = CONSTANTS.DATA_DIR / "index.jsonl"
|
|
assert root_manifest.exists()
|
|
manifest_records = [json.loads(line) for line in root_manifest.read_text().splitlines() if line.strip()]
|
|
assert [record["id"] for record in manifest_records] == [str(snapshot.id)]
|
|
# JSON and JSONL are alternate containers for one static-export schema;
|
|
# consumers must not see TYPE/tags/archive paths change by file format.
|
|
assert manifest_records[0]["TYPE"] == "core.models.Snapshot"
|
|
assert "type" not in manifest_records[0]
|
|
assert isinstance(manifest_records[0]["tags"], list)
|
|
assert manifest_records[0]["archive_path"] == static_path
|
|
assert manifest_records[0]["archive_url"] == f"./{static_path}/index.html"
|
|
assert detail_path.exists()
|
|
detail_html = detail_path.read_text()
|
|
assert f"/snapshot/{snapshot.id.hex}" not in detail_html
|
|
assert "/admin/" not in detail_html
|
|
|
|
|
|
def test_static_exports_use_filesystem_paths_not_live_django_routes(snapshot):
|
|
from archivebox.config import CONSTANTS
|
|
from archivebox.core.models import ArchiveResult
|
|
|
|
snapshot_dir = Path(snapshot.output_dir)
|
|
screenshot_dir = snapshot_dir / "screenshot"
|
|
screenshot_dir.mkdir(parents=True, exist_ok=True)
|
|
screenshot_file = screenshot_dir / "screenshot.png"
|
|
screenshot_file.write_bytes(b"real screenshot")
|
|
ArchiveResult.objects.create(
|
|
snapshot=snapshot,
|
|
plugin="screenshot",
|
|
hook_name="on_Snapshot__50_screenshot.py",
|
|
status=ArchiveResult.StatusChoices.SUCCEEDED,
|
|
output_str="screenshot.png",
|
|
output_files={"screenshot.png": {"size": screenshot_file.stat().st_size}},
|
|
output_size=screenshot_file.stat().st_size,
|
|
)
|
|
ArchiveResult.objects.create(
|
|
snapshot=snapshot,
|
|
plugin="chrome_screencast",
|
|
hook_name="on_Snapshot__02_chrome_screencast.py",
|
|
status=ArchiveResult.StatusChoices.SUCCEEDED,
|
|
output_str="2 screencast frames (0 kept)",
|
|
output_files={"hook.stderr.log": {"size": 12}},
|
|
output_size=12,
|
|
)
|
|
wget_dir = snapshot_dir / "wget"
|
|
wget_dir.mkdir()
|
|
wget_file = wget_dir / "index%3A.html"
|
|
wget_file.write_text("archived page")
|
|
ArchiveResult.objects.create(
|
|
snapshot=snapshot,
|
|
plugin="wget",
|
|
hook_name="on_Snapshot__35_wget.py",
|
|
status=ArchiveResult.StatusChoices.SUCCEEDED,
|
|
output_str="index%3A.html",
|
|
output_files={"index%3A.html": {"size": wget_file.stat().st_size}},
|
|
output_size=wget_file.stat().st_size,
|
|
)
|
|
ArchiveResult.objects.create(
|
|
snapshot=snapshot,
|
|
plugin="staticfile",
|
|
hook_name="on_Snapshot__26_staticfile.py",
|
|
status=ArchiveResult.StatusChoices.SUCCEEDED,
|
|
output_str="prenav.json",
|
|
output_files={"prenav.json": {"size": 66}},
|
|
output_size=66,
|
|
)
|
|
ytdlp_dir = snapshot_dir / "ytdlp"
|
|
ytdlp_dir.mkdir()
|
|
media_file = ytdlp_dir / "saved.m4a"
|
|
media_file.write_bytes(b"audio")
|
|
ArchiveResult.objects.create(
|
|
snapshot=snapshot,
|
|
plugin="ytdlp",
|
|
hook_name="on_Snapshot__60_ytdlp.py",
|
|
status=ArchiveResult.StatusChoices.SUCCEEDED,
|
|
output_str="saved.m4a",
|
|
output_files={
|
|
"saved.m4a": {"size": media_file.stat().st_size},
|
|
"deleted.temp.m4a": {"size": 123},
|
|
},
|
|
output_size=media_file.stat().st_size,
|
|
)
|
|
hashes_dir = snapshot_dir / "hashes"
|
|
hashes_dir.mkdir()
|
|
(hashes_dir / "hashes.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"screenshot/screenshot.png": {"size": screenshot_file.stat().st_size},
|
|
"wget/index%3A.html": {"size": wget_file.stat().st_size},
|
|
"staticfile/prenav.json": {"size": 66},
|
|
"ytdlp/saved.m4a": {"size": media_file.stat().st_size},
|
|
"ytdlp/deleted.temp.m4a": {"size": 123},
|
|
},
|
|
),
|
|
)
|
|
static_path = snapshot_dir.relative_to(CONSTANTS.DATA_DIR).as_posix()
|
|
queryset = Snapshot.objects.filter(pk=snapshot.pk).prefetch_related("tags")
|
|
|
|
html = queryset.to_html(with_headers=True)
|
|
[record] = json.loads(queryset.to_json(with_headers=False))
|
|
detail_html = snapshot_dir / "index.html"
|
|
|
|
assert f"./{static_path}/index.html" in html
|
|
assert f"./{static_path}/screenshot/screenshot.png" in html
|
|
assert f"./{static_path}/wget/index%253A.html" in html
|
|
assert f"./{static_path}/index.jsonl" in html
|
|
assert f"/snapshot/{snapshot.id.hex}" not in html
|
|
assert "/web/" not in html
|
|
assert "/static/" not in html
|
|
assert "/None" not in html
|
|
assert "staticfile/prenav.json" not in html
|
|
assert record["archive_path"] == static_path
|
|
assert record["archive_url"] == f"./{static_path}/index.html"
|
|
assert detail_html.exists()
|
|
rendered_detail = detail_html.read_text()
|
|
assert "core/snapshot.html" not in rendered_detail
|
|
assert "screenshot/screenshot.png" in rendered_detail
|
|
assert "staticfile/prenav.json" not in rendered_detail
|
|
assert "ytdlp/saved.m4a" in rendered_detail
|
|
assert "ytdlp/deleted.temp.m4a" not in rendered_detail
|
|
assert f"/snapshot/{snapshot.id.hex}" not in rendered_detail
|
|
|
|
|
|
def test_streaming_json_matches_snapshot_serializer(initialized_archive):
|
|
from archivebox.crawls.models import Crawl
|
|
|
|
with use_archivebox_db(initialized_archive):
|
|
user = get_user_model().objects.create_user(username="streaming-json-parity")
|
|
crawl = Crawl.objects.create(
|
|
urls="https://example.com/a\nhttps://example.com/b",
|
|
created_by=user,
|
|
status=Crawl.StatusChoices.SEALED,
|
|
retry_at=None,
|
|
)
|
|
populated = Snapshot.objects.create(
|
|
crawl=crawl,
|
|
url="https://example.com/a",
|
|
timestamp="20260721220000000000000000000001",
|
|
title="Populated title",
|
|
status=Snapshot.StatusChoices.SEALED,
|
|
retry_at=None,
|
|
output_size=42,
|
|
)
|
|
populated.save_tags(["éclair", "Zulu", "alpha"])
|
|
empty = Snapshot.objects.create(
|
|
crawl=crawl,
|
|
url="https://example.com/b",
|
|
timestamp="20260721220000000000000000000002",
|
|
title=None,
|
|
status=Snapshot.StatusChoices.QUEUED,
|
|
output_size=0,
|
|
)
|
|
queryset = Snapshot.objects.filter(id__in=(populated.id, empty.id)).order_by("url")
|
|
|
|
expected = [snapshot.to_json() for snapshot in queryset.prefetch_related("tags")]
|
|
actual = list(iter_snapshot_json(queryset))
|
|
|
|
assert actual == expected
|
|
assert [record["url"] for record in actual] == ["https://example.com/a", "https://example.com/b"]
|
|
|
|
|
|
def test_list_limit_zero_streams_one_million_snapshots_without_materializing(initialized_archive, tmp_path):
|
|
"""Regression: archivebox list --limit=0 must stream unbounded result sets."""
|
|
from archivebox.crawls.models import Crawl
|
|
|
|
with use_archivebox_db(initialized_archive):
|
|
user = get_user_model().objects.create_user(username="million-snapshot-list")
|
|
crawl = Crawl.objects.create(
|
|
urls="https://example.com",
|
|
created_by=user,
|
|
status=Crawl.StatusChoices.SEALED,
|
|
retry_at=None,
|
|
)
|
|
now = timezone.now().isoformat()
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
WITH RECURSIVE seq(n) AS (
|
|
SELECT 1
|
|
UNION ALL
|
|
SELECT n + 1 FROM seq WHERE n < 1000000
|
|
)
|
|
INSERT INTO core_snapshot (
|
|
id,
|
|
url,
|
|
timestamp,
|
|
title,
|
|
bookmarked_at,
|
|
created_at,
|
|
modified_at,
|
|
downloaded_at,
|
|
fs_version,
|
|
crawl_id,
|
|
config,
|
|
depth,
|
|
notes,
|
|
num_uses_failed,
|
|
num_uses_succeeded,
|
|
retry_at,
|
|
status,
|
|
delete_at,
|
|
output_size,
|
|
parent_snapshot_id
|
|
)
|
|
SELECT
|
|
lower(hex(randomblob(16))),
|
|
'https://example.com/page-' || n,
|
|
printf('9%031d', n),
|
|
'',
|
|
%s,
|
|
%s,
|
|
%s,
|
|
NULL,
|
|
'0.9.0',
|
|
%s,
|
|
'{}',
|
|
0,
|
|
'',
|
|
0,
|
|
0,
|
|
NULL,
|
|
'sealed',
|
|
NULL,
|
|
0,
|
|
NULL
|
|
FROM seq
|
|
""",
|
|
[now, now, now, str(crawl.id).replace("-", "")],
|
|
)
|
|
|
|
output_path = tmp_path / "million-snapshots.jsonl"
|
|
with output_path.open("w") as stdout:
|
|
result = run_archivebox_cmd(
|
|
["list", "--limit=0"],
|
|
cwd=initialized_archive,
|
|
stdout=stdout,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
with output_path.open() as stdout:
|
|
assert sum(1 for line in stdout if line.startswith("{")) == 1000000
|
|
|
|
|
|
def test_list_outputs_existing_snapshots_as_jsonl(initialized_archive):
|
|
"""Test that list prints one JSON object per stored snapshot."""
|
|
env = cli_env(disable_extractors=True)
|
|
for url in ["https://example.com", "https://iana.org"]:
|
|
run_archivebox_cmd(
|
|
["add", "--index-only", "--depth=0", url],
|
|
env=env,
|
|
check=True,
|
|
)
|
|
run_queued_crawls(initialized_archive, env)
|
|
with use_archivebox_db(initialized_archive):
|
|
Snapshot.objects.get(url="https://example.com").save_tags(["z-tag", "a-tag"])
|
|
|
|
result = run_archivebox_cmd(
|
|
["list"],
|
|
timeout=30,
|
|
)
|
|
|
|
rows = parse_jsonl_output(result.stdout)
|
|
urls = {row["url"] for row in rows}
|
|
rows_by_url = {row["url"]: row for row in rows}
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert "https://example.com" in urls
|
|
assert "https://iana.org" in urls
|
|
assert rows_by_url["https://example.com"]["tags"] == "a-tag,z-tag"
|
|
|
|
|
|
def test_list_filters_by_url_icontains(initialized_archive):
|
|
"""Test that list --url__icontains returns only matching snapshots."""
|
|
env = cli_env(disable_extractors=True)
|
|
for url in ["https://example.com", "https://iana.org"]:
|
|
run_archivebox_cmd(
|
|
["add", "--index-only", "--depth=0", url],
|
|
env=env,
|
|
check=True,
|
|
)
|
|
run_queued_crawls(initialized_archive, env)
|
|
|
|
result = run_archivebox_cmd(
|
|
["list", "--url__icontains", "example.com"],
|
|
timeout=30,
|
|
)
|
|
|
|
rows = parse_jsonl_output(result.stdout)
|
|
assert result.returncode == 0, result.stderr
|
|
assert len(rows) == 1
|
|
assert rows[0]["url"] == "https://example.com"
|
|
|
|
|
|
def test_list_filters_by_crawl_id_and_limit(initialized_archive):
|
|
"""Test that crawl-id and limit filters constrain the result set."""
|
|
env = cli_env(disable_extractors=True)
|
|
for url in ["https://example.com", "https://iana.org"]:
|
|
run_archivebox_cmd(
|
|
["add", "--index-only", "--depth=0", url],
|
|
env=env,
|
|
check=True,
|
|
)
|
|
run_queued_crawls(initialized_archive, env)
|
|
|
|
with use_archivebox_db(initialized_archive):
|
|
crawl_id = str(Snapshot.objects.values_list("crawl_id", flat=True).get(url="https://example.com"))
|
|
|
|
result = run_archivebox_cmd(
|
|
["list", "--crawl-id", crawl_id, "--limit", "1"],
|
|
timeout=30,
|
|
)
|
|
|
|
rows = parse_jsonl_output(result.stdout)
|
|
assert result.returncode == 0, result.stderr
|
|
assert len(rows) == 1
|
|
assert rows[0]["crawl_id"].replace("-", "") == crawl_id.replace("-", "")
|
|
assert rows[0]["url"] == "https://example.com"
|
|
|
|
|
|
def test_list_filters_by_status(initialized_archive):
|
|
"""Test that list can filter using the current snapshot status."""
|
|
env = cli_env(disable_extractors=True)
|
|
run_archivebox_cmd(
|
|
["add", "--index-only", "--depth=0", "https://example.com"],
|
|
env=env,
|
|
check=True,
|
|
)
|
|
run_queued_crawls(initialized_archive, env)
|
|
|
|
with use_archivebox_db(initialized_archive):
|
|
status = Snapshot.objects.values_list("status", flat=True).get()
|
|
|
|
result = run_archivebox_cmd(
|
|
["list", "--status", status],
|
|
timeout=30,
|
|
)
|
|
|
|
rows = parse_jsonl_output(result.stdout)
|
|
assert result.returncode == 0, result.stderr
|
|
assert len(rows) == 1
|
|
assert rows[0]["status"] == status
|
|
|
|
|
|
def test_list_help_lists_filter_options(initialized_archive):
|
|
"""Test that list --help documents the supported filter flags."""
|
|
|
|
result = run_archivebox_cmd(
|
|
["list", "--help"],
|
|
timeout=30,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert "--url__icontains" in result.stdout
|
|
assert "--crawl-id" in result.stdout
|
|
assert "--limit" in result.stdout
|
|
assert "--search" in result.stdout
|
|
assert "--json" in result.stdout
|
|
assert "--html" in result.stdout
|
|
assert "--with-headers" in result.stdout
|
|
|
|
|
|
def test_list_allows_sort_with_limit(initialized_archive):
|
|
"""Test that list can sort and then apply limit without queryset slicing errors."""
|
|
env = cli_env(disable_extractors=True)
|
|
for url in ["https://example.com", "https://iana.org", "https://example.net"]:
|
|
run_archivebox_cmd(
|
|
["add", "--index-only", "--depth=0", url],
|
|
env=env,
|
|
check=True,
|
|
)
|
|
run_queued_crawls(initialized_archive, env)
|
|
|
|
result = run_archivebox_cmd(
|
|
["list", "--limit", "2", "--sort", "-created_at"],
|
|
timeout=30,
|
|
)
|
|
|
|
rows = parse_jsonl_output(result.stdout)
|
|
assert result.returncode == 0, result.stderr
|
|
assert len(rows) == 2
|
|
|
|
|
|
def test_snapshot_list_search_meta(initialized_archive):
|
|
"""snapshot list should support metadata search mode."""
|
|
url = create_test_url(domain="meta-search-example.com")
|
|
run_archivebox_cmd(["snapshot", "create", url], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
|
|
_cmd_result = run_archivebox_cmd(
|
|
["snapshot", "list", "--search=meta", "meta-search-example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0, f"Command failed: {stderr}"
|
|
records = parse_jsonl_output(stdout)
|
|
assert len(records) == 1
|
|
assert "meta-search-example.com" in records[0]["url"]
|
|
|
|
|
|
def test_list_search_meta_matches_metadata(initialized_archive):
|
|
"""top-level list --search=meta should apply metadata search to the queryset."""
|
|
url = create_test_url(domain="top-level-meta-search-example.com")
|
|
run_archivebox_cmd(["snapshot", "create", url], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
|
|
_cmd_result = run_archivebox_cmd(
|
|
["list", "--search=meta", "top-level-meta-search-example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0, f"Command failed: {stderr}"
|
|
records = parse_jsonl_output(stdout)
|
|
assert len(records) == 1
|
|
assert "top-level-meta-search-example.com" in records[0]["url"]
|
|
|
|
|
|
def test_search_command_finds_snapshots(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
_cmd_result = run_archivebox_cmd(["search", "example"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0, stderr
|
|
assert "example" in stdout
|
|
|
|
|
|
def test_search_command_returns_no_results_for_missing_term(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
_cmd_result = run_archivebox_cmd(
|
|
["search", "nonexistentterm12345"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
_stdout, _stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code in [0, 1]
|
|
|
|
|
|
def test_search_command_on_empty_archive(initialized_archive):
|
|
_cmd_result = run_archivebox_cmd(["search", "anything"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
_stdout, _stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code in [0, 1]
|
|
|
|
|
|
def test_search_command_outputs_matching_snapshots_as_jsonl(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
_cmd_result = run_archivebox_cmd(["search"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0, stderr
|
|
records = parse_jsonl_output(stdout)
|
|
assert any("example.com" in row.get("url", "") for row in records)
|
|
|
|
|
|
def test_search_command_json_outputs_matching_snapshots(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
result = run_archivebox_cmd(["search", "--json"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
payload = json.loads(result.stdout)
|
|
assert any("example.com" in row.get("url", "") for row in payload)
|
|
|
|
|
|
def test_search_command_json_with_headers_wraps_links_payload(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
result = run_archivebox_cmd(
|
|
["search", "--json", "--with-headers"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
payload = json.loads(result.stdout)
|
|
assert "links" in payload
|
|
assert any("example.com" in row.get("url", "") for row in payload["links"])
|
|
|
|
|
|
def test_search_command_html_outputs_markup(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
result = run_archivebox_cmd(["search", "--html"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert "<" in result.stdout
|
|
assert "example.com" in result.stdout
|
|
|
|
|
|
def test_search_command_csv_outputs_requested_column(initialized_archive):
|
|
run_archivebox_cmd(
|
|
["snapshot", "create", "https://example.com"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
|
|
_cmd_result = run_archivebox_cmd(
|
|
["search", "--csv", "url", "--with-headers"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0, stderr
|
|
assert "url" in stdout
|
|
assert "example.com" in stdout
|
|
|
|
|
|
def test_search_command_with_headers_requires_structured_output_format(initialized_archive):
|
|
_cmd_result = run_archivebox_cmd(["search", "--with-headers"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
_stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code != 0
|
|
assert "requires" in stderr.lower()
|
|
assert "json" in stderr.lower()
|
|
|
|
|
|
def test_search_command_sort_option_runs_successfully(initialized_archive):
|
|
for url in ["https://iana.org", "https://example.com"]:
|
|
run_archivebox_cmd(["snapshot", "create", url], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
|
|
_cmd_result = run_archivebox_cmd(
|
|
["search", "--csv", "url", "--sort=url"],
|
|
cwd=initialized_archive,
|
|
default_cli_env=True,
|
|
disable_extractors=True,
|
|
)
|
|
stdout, stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0, stderr
|
|
assert "example.com" in stdout or "iana.org" in stdout
|
|
|
|
|
|
def test_search_command_help_lists_supported_filters(initialized_archive):
|
|
_cmd_result = run_archivebox_cmd(["search", "--help"], cwd=initialized_archive, default_cli_env=True, disable_extractors=True)
|
|
stdout, _stderr, code = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
|
|
|
|
assert code == 0
|
|
assert "--url__icontains" in stdout
|
|
assert "--crawl-id" in stdout
|
|
assert "--status" in stdout
|
|
assert "--sort" in stdout
|
|
assert "--json" in stdout
|
|
assert "--html" in stdout
|