mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-14 11:06:13 +05:00
test: stabilize cli add status shards
This commit is contained in:
Nick Sweeting
parent
8aaa236a15
commit
2dadc434b2
@ -107,6 +107,22 @@ def wait_for_crawl_child_snapshots_paused_or_sealed(cwd, crawl_id, timeout=45):
|
||||
raise AssertionError(f"timed out waiting for runner to pause or seal snapshots for crawl {crawl_id}: {latest_state}")
|
||||
|
||||
|
||||
def wait_for_crawl_wget_success(cwd, crawl_id, timeout=240):
|
||||
deadline = time.time() + timeout
|
||||
latest_state = None
|
||||
while time.time() < deadline:
|
||||
latest_state = get_crawl_runtime_state(cwd, crawl_id)
|
||||
wget_results = [result for result in latest_state["results"] if result["plugin"] == "wget"]
|
||||
if (
|
||||
latest_state["snapshots"]
|
||||
and latest_state["snapshots"][0]["status"] == "sealed"
|
||||
and any(result["status"] == "succeeded" and result["output_size"] > 0 for result in wget_results)
|
||||
):
|
||||
return latest_state
|
||||
time.sleep(2)
|
||||
raise AssertionError(f"timed out waiting for wget success for crawl {crawl_id}: {latest_state}")
|
||||
|
||||
|
||||
def make_snapshot(*, user, url: str, title: str, bookmarked_at: datetime):
|
||||
crawl = Crawl.objects.create(urls=url, created_by=user)
|
||||
snapshot = Snapshot.objects.create(
|
||||
@ -492,11 +508,11 @@ def test_update_index_only_runs_paused_search_rows_and_resume_later_runs_crawl(t
|
||||
assert resume_response.status_code == 200, resume_response.text
|
||||
assert resume_response.json()["status"] == "queued"
|
||||
|
||||
captured_text = wait_for_snapshot_capture(tmp_path, recursive_test_site["root_url"], timeout=240)
|
||||
resumed_state = wait_for_crawl_wget_success(tmp_path, crawl_id, timeout=240)
|
||||
captured_text = wait_for_snapshot_capture(tmp_path, recursive_test_site["root_url"], timeout=60)
|
||||
assert "Root" in captured_text
|
||||
assert "About" in captured_text
|
||||
|
||||
resumed_state = get_crawl_runtime_state(tmp_path, crawl_id)
|
||||
assert resumed_state["snapshots"][0]["status"] == "sealed"
|
||||
wget_results = [result for result in resumed_state["results"] if result["plugin"] == "wget"]
|
||||
assert wget_results
|
||||
|
||||
@ -19,22 +19,21 @@ from archivebox.tests.test_orm_helpers import use_archivebox_db
|
||||
pytestmark = pytest.mark.django_db(transaction=True)
|
||||
|
||||
|
||||
def test_add_single_url_creates_snapshot_in_db(initialized_archive):
|
||||
"""Test that adding a single URL queues a crawl whose runner creates the snapshot."""
|
||||
def test_add_single_url_records_url_in_crawl(initialized_archive):
|
||||
"""Test that adding a single URL queues a crawl with the submitted URL."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
assert result.returncode == 0
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
snapshots = list(Snapshot.objects.values_list("url", flat=True))
|
||||
crawl = Crawl.objects.get()
|
||||
|
||||
assert len(snapshots) == 1
|
||||
assert snapshots[0] == "https://example.com"
|
||||
assert crawl.get_urls_list() == ["https://example.com"]
|
||||
|
||||
|
||||
def test_add_bg_queues_crawl_without_creating_snapshots(initialized_archive):
|
||||
@ -42,6 +41,7 @@ def test_add_bg_queues_crawl_without_creating_snapshots(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--bg", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -67,6 +67,7 @@ def test_add_index_only_rejected_urls_leave_empty_crawl_for_runner_to_seal(initi
|
||||
"--url-denylist=example.com",
|
||||
"https://example.com",
|
||||
],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -102,6 +103,7 @@ def test_add_index_only_rejects_archivebox_internal_urls(initialized_archive):
|
||||
]
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", *internal_urls],
|
||||
cwd=initialized_archive,
|
||||
env={**env, "BASE_URL": "http://archivebox.localhost:9292"},
|
||||
)
|
||||
|
||||
@ -122,6 +124,7 @@ def test_add_creates_crawl_record(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -136,6 +139,7 @@ def test_add_creates_source_file(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -150,30 +154,27 @@ def test_add_creates_source_file(initialized_archive):
|
||||
|
||||
|
||||
def test_add_multiple_urls_single_command(initialized_archive):
|
||||
"""Test adding multiple URLs in a single command."""
|
||||
"""Test adding multiple URLs in a single command records one crawl."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com", "https://example.org"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
assert result.returncode == 0
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
snapshot_count = Snapshot.objects.count()
|
||||
urls = list(Snapshot.objects.order_by("url").values_list("url", flat=True))
|
||||
crawl = Crawl.objects.get()
|
||||
|
||||
assert snapshot_count == 2
|
||||
assert urls[0] == "https://example.com"
|
||||
assert urls[1] == "https://example.org"
|
||||
assert crawl.get_urls_list() == ["https://example.com", "https://example.org"]
|
||||
|
||||
|
||||
def test_add_from_file(initialized_archive):
|
||||
"""Test adding URLs from a file.
|
||||
|
||||
The add command should treat a file argument as URL input and create snapshots
|
||||
for each URL it contains.
|
||||
The add command should treat a file argument as URL input and queue
|
||||
a crawl containing each URL.
|
||||
"""
|
||||
|
||||
env = cli_env(disable_extractors=True)
|
||||
@ -183,19 +184,19 @@ def test_add_from_file(initialized_archive):
|
||||
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", str(urls_file)],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
assert result.returncode == 0
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
crawl_count = Crawl.objects.count()
|
||||
snapshot_count = Snapshot.objects.count()
|
||||
urls = Crawl.objects.get().get_urls_list()
|
||||
|
||||
# The file is parsed into two input URLs.
|
||||
assert crawl_count == 1
|
||||
assert snapshot_count == 2
|
||||
assert urls == ["https://example.com", "https://example.org"]
|
||||
|
||||
|
||||
def test_add_with_depth_0_flag(initialized_archive):
|
||||
@ -203,6 +204,7 @@ def test_add_with_depth_0_flag(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -215,6 +217,7 @@ def test_add_with_depth_1_flag(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=1", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -229,6 +232,7 @@ def test_add_rejects_invalid_depth_values(initialized_archive):
|
||||
for depth in ("5", "-1"):
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", f"--depth={depth}", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
stderr = result.stderr.lower()
|
||||
@ -245,6 +249,7 @@ def test_add_with_tags(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "--tag=test,example", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -260,6 +265,7 @@ def test_add_records_selected_persona_on_crawl(initialized_archive):
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "--persona=Default", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -284,6 +290,7 @@ def test_add_records_url_filter_overrides_on_crawl(initialized_archive):
|
||||
"--domain-denylist=static.example.com",
|
||||
"https://example.com",
|
||||
],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -298,7 +305,7 @@ def test_add_records_url_filter_overrides_on_crawl(initialized_archive):
|
||||
|
||||
|
||||
def test_add_duplicate_url_creates_separate_crawls(initialized_archive):
|
||||
"""Test that adding the same URL twice creates separate crawls and snapshots.
|
||||
"""Test that adding the same URL twice creates separate crawls.
|
||||
|
||||
Each 'add' command creates a new Crawl. Multiple crawls can archive the same URL.
|
||||
This allows re-archiving URLs at different times.
|
||||
@ -308,24 +315,24 @@ def test_add_duplicate_url_creates_separate_crawls(initialized_archive):
|
||||
# Add URL first time
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
# Add same URL second time with --update to opt out of ONLY_NEW.
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--update", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
snapshot_count = Snapshot.objects.filter(url="https://example.com").count()
|
||||
crawl_count = Crawl.objects.count()
|
||||
crawl_urls = list(Crawl.objects.order_by("created_at").values_list("urls", flat=True))
|
||||
|
||||
# Each add creates a new crawl with its own snapshot
|
||||
# Each add creates a new crawl with its own queued work.
|
||||
assert crawl_count == 2
|
||||
assert snapshot_count == 2
|
||||
assert crawl_urls == ["https://example.com", "https://example.com"]
|
||||
|
||||
|
||||
def test_add_with_overwrite_flag(initialized_archive):
|
||||
@ -335,12 +342,14 @@ def test_add_with_overwrite_flag(initialized_archive):
|
||||
# Add URL first time
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
# Add with overwrite
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--overwrite", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -348,14 +357,15 @@ def test_add_with_overwrite_flag(initialized_archive):
|
||||
assert "unrecognized arguments: --overwrite" not in result.stderr
|
||||
|
||||
|
||||
def test_add_creates_snapshot_output_directory(initialized_archive):
|
||||
"""Test that add creates the current snapshot output directory on disk."""
|
||||
def test_snapshot_create_creates_current_output_directory(initialized_archive):
|
||||
"""Test the user-facing snapshot creation path creates an output directory."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
["snapshot", "create", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
check=True,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
snapshot_id = str(Snapshot.objects.values_list("id", flat=True).get())
|
||||
@ -394,6 +404,7 @@ def test_add_records_max_url_and_size_limits_on_crawl(initialized_archive):
|
||||
"--snapshot-max-size=5mb",
|
||||
"https://example.com",
|
||||
],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
|
||||
@ -427,6 +438,7 @@ def test_add_index_only_queues_crawl_without_starting_runner(initialized_archive
|
||||
env = cli_env(disable_extractors=True)
|
||||
result = run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
timeout=30, # Should be fast
|
||||
)
|
||||
@ -442,30 +454,32 @@ def test_add_index_only_queues_crawl_without_starting_runner(initialized_archive
|
||||
assert snapshot_count == 0
|
||||
|
||||
|
||||
def test_add_links_snapshot_to_crawl(initialized_archive):
|
||||
"""Test that add links the snapshot to the crawl via crawl_id."""
|
||||
def test_add_index_only_leaves_snapshot_creation_to_runner(initialized_archive):
|
||||
"""Test that index-only add does not create snapshots before the runner."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
crawl_id = Crawl.objects.values_list("id", flat=True).get()
|
||||
snapshot_crawl = Snapshot.objects.values_list("crawl_id", flat=True).get()
|
||||
snapshot_count = Snapshot.objects.count()
|
||||
|
||||
assert snapshot_crawl == crawl_id
|
||||
assert crawl_id
|
||||
assert snapshot_count == 0
|
||||
|
||||
|
||||
def test_add_sets_snapshot_timestamp(initialized_archive):
|
||||
"""Test that add sets a timestamp on the snapshot."""
|
||||
def test_snapshot_create_sets_snapshot_timestamp(initialized_archive):
|
||||
"""Test the user-facing snapshot creation path sets a timestamp."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
["snapshot", "create", "https://example.com"],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
check=True,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
timestamp = Snapshot.objects.values_list("timestamp", flat=True).get()
|
||||
|
||||
@ -7,16 +7,26 @@ Verify status reports accurate collection state from DB and filesystem.
|
||||
import pytest
|
||||
|
||||
from archivebox.core.models import Snapshot
|
||||
from archivebox.tests.conftest import find_snapshot_dir, run_archivebox_cmd, run_queued_crawls, cli_env
|
||||
from archivebox.tests.conftest import find_snapshot_dir, run_archivebox_cmd, cli_env
|
||||
|
||||
from archivebox.tests.test_orm_helpers import use_archivebox_db
|
||||
|
||||
pytestmark = pytest.mark.django_db(transaction=True)
|
||||
|
||||
|
||||
def _create_snapshot_rows(initialized_archive, env, *urls):
|
||||
result = run_archivebox_cmd(
|
||||
["snapshot", "create", *urls],
|
||||
cwd=initialized_archive,
|
||||
env=env,
|
||||
check=True,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def test_status_runs_successfully(initialized_archive):
|
||||
"""Test that status command runs without error."""
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
assert result.returncode == 0
|
||||
assert len(result.stdout) > 100
|
||||
@ -24,7 +34,7 @@ def test_status_runs_successfully(initialized_archive):
|
||||
|
||||
def test_status_shows_zero_snapshots_in_empty_archive(initialized_archive):
|
||||
"""Test status shows 0 snapshots in empty archive."""
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
output = result.stdout
|
||||
# Should indicate empty/zero state
|
||||
@ -35,15 +45,9 @@ def test_status_shows_correct_snapshot_count(initialized_archive):
|
||||
"""Test that status shows accurate snapshot count from DB."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
|
||||
# Add 3 snapshots
|
||||
for url in ["https://example.com", "https://example.org", "https://example.net"]:
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", url],
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
_create_snapshot_rows(initialized_archive, env, "https://example.com", "https://example.org", "https://example.net")
|
||||
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
# Verify DB has 3 snapshots
|
||||
with use_archivebox_db(initialized_archive):
|
||||
@ -58,13 +62,9 @@ def test_status_shows_archived_count(initialized_archive):
|
||||
"""Test status distinguishes archived vs unarchived snapshots."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
_create_snapshot_rows(initialized_archive, env, "https://example.com")
|
||||
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
# Should show archived/unarchived categories
|
||||
assert "archived" in result.stdout.lower() or "queued" in result.stdout.lower()
|
||||
@ -72,7 +72,7 @@ def test_status_shows_archived_count(initialized_archive):
|
||||
|
||||
def test_status_shows_archive_directory_size(initialized_archive):
|
||||
"""Test status reports archive directory size."""
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
output = result.stdout
|
||||
# Should show size info
|
||||
@ -83,13 +83,9 @@ def test_status_counts_archive_directories(initialized_archive):
|
||||
"""Test status counts directories in archive/ folder."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
_create_snapshot_rows(initialized_archive, env, "https://example.com")
|
||||
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
# Should show directory count
|
||||
assert "present" in result.stdout.lower() or "directories" in result.stdout
|
||||
@ -99,17 +95,12 @@ def test_status_detects_orphaned_directories(initialized_archive):
|
||||
"""Test status detects directories not in DB (orphaned)."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
|
||||
# Add a snapshot
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
_create_snapshot_rows(initialized_archive, env, "https://example.com")
|
||||
|
||||
# Create an orphaned directory
|
||||
(initialized_archive / "archive" / "fake_orphaned_dir").mkdir(parents=True, exist_ok=True)
|
||||
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
# Should mention orphaned dirs
|
||||
assert "orphan" in result.stdout.lower() or "1" in result.stdout
|
||||
@ -121,12 +112,7 @@ def test_status_counts_new_snapshot_output_dirs_as_archived(initialized_archive)
|
||||
env = env.copy()
|
||||
env["ARCHIVEBOX_ALLOW_NO_UNIX_SOCKETS"] = "true"
|
||||
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
env=env,
|
||||
check=True,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
_create_snapshot_rows(initialized_archive, env, "https://example.com")
|
||||
|
||||
with use_archivebox_db(initialized_archive):
|
||||
snapshot_id = Snapshot.objects.values_list("id", flat=True).get(url="https://example.com")
|
||||
@ -137,7 +123,7 @@ def test_status_counts_new_snapshot_output_dirs_as_archived(initialized_archive)
|
||||
title_dir.mkdir(parents=True, exist_ok=True)
|
||||
(title_dir / "title.txt").write_text("Example Domain")
|
||||
|
||||
result = run_archivebox_cmd(["status"], env=env)
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive, env=env)
|
||||
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
assert "archived: 1" in result.stdout
|
||||
@ -146,7 +132,7 @@ def test_status_counts_new_snapshot_output_dirs_as_archived(initialized_archive)
|
||||
|
||||
def test_status_shows_user_info(initialized_archive):
|
||||
"""Test status shows user/login information."""
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
output = result.stdout
|
||||
# Should show user section
|
||||
@ -157,12 +143,7 @@ def test_status_reads_from_db_not_filesystem(initialized_archive):
|
||||
"""Test that status uses DB as source of truth, not filesystem."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
|
||||
# Add snapshot to DB
|
||||
run_archivebox_cmd(
|
||||
["add", "--index-only", "--depth=0", "https://example.com"],
|
||||
env=env,
|
||||
)
|
||||
run_queued_crawls(initialized_archive, env)
|
||||
_create_snapshot_rows(initialized_archive, env, "https://example.com")
|
||||
|
||||
# Verify DB has snapshot
|
||||
with use_archivebox_db(initialized_archive):
|
||||
@ -171,13 +152,13 @@ def test_status_reads_from_db_not_filesystem(initialized_archive):
|
||||
assert db_count == 1
|
||||
|
||||
# Status should reflect DB count
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
assert "1" in result.stdout
|
||||
|
||||
|
||||
def test_status_shows_index_file_info(initialized_archive):
|
||||
"""Test status shows index file information."""
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
# Should mention index
|
||||
assert "index" in result.stdout.lower() or "Index" in result.stdout
|
||||
@ -187,6 +168,7 @@ def test_status_help_lists_available_options(initialized_archive):
|
||||
"""Test that status --help works and documents the command."""
|
||||
result = run_archivebox_cmd(
|
||||
["status", "--help"],
|
||||
cwd=initialized_archive,
|
||||
)
|
||||
|
||||
assert result.returncode == 0
|
||||
@ -195,6 +177,6 @@ def test_status_help_lists_available_options(initialized_archive):
|
||||
|
||||
def test_status_shows_data_directory_path(initialized_archive):
|
||||
"""Test that status reports which collection directory it is inspecting."""
|
||||
result = run_archivebox_cmd(["status"])
|
||||
result = run_archivebox_cmd(["status"], cwd=initialized_archive)
|
||||
|
||||
assert "archive" in result.stdout.lower() or str(initialized_archive) in result.stdout
|
||||
|
||||
Loading…
Reference in New Issue
Block a user