mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-12 19:50:57 +05:00
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
|
|
import pytest
|
|
|
|
from archivebox.core.shutdown_util import foreground_shutdown_signals
|
|
from archivebox.core.shutdown_util import raise_if_shutdown_requested
|
|
from archivebox.misc.checks import _migration_interrupt_message
|
|
from archivebox.misc.checks import is_archivebox_source_root
|
|
|
|
|
|
def test_migration_interrupt_message_prints_resume_command_and_atomic_safety():
|
|
message = _migration_interrupt_message()
|
|
|
|
assert "Migration interrupted." in message
|
|
assert "Database migrations are atomic" in message
|
|
assert "no data loss has occurred" in message
|
|
assert "archivebox init" in message
|
|
|
|
|
|
def test_migration_interrupt_message_before_apply_says_no_changes_applied():
|
|
message = _migration_interrupt_message(before_apply=True)
|
|
|
|
assert "cancelled before any changes were applied" in message
|
|
assert "archivebox init" in message
|
|
|
|
|
|
def test_source_root_check_treats_inaccessible_probe_paths_as_not_source_root(tmp_path):
|
|
(tmp_path / ".git").mkdir()
|
|
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'not-archivebox'\n")
|
|
archivebox_dir = tmp_path / "archivebox"
|
|
archivebox_dir.mkdir()
|
|
archivebox_dir.chmod(0)
|
|
try:
|
|
assert is_archivebox_source_root(tmp_path) is False
|
|
finally:
|
|
archivebox_dir.chmod(0o700)
|
|
|
|
|
|
@pytest.mark.parametrize("sig", [signal.SIGINT, signal.SIGTERM])
|
|
def test_migration_interrupt_handler_exits_for_sigint_and_sigterm(sig):
|
|
process = subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
textwrap.dedent(
|
|
"""
|
|
import signal
|
|
|
|
from archivebox.misc.checks import _exit_on_migration_interrupt
|
|
|
|
with _exit_on_migration_interrupt():
|
|
print("READY", flush=True)
|
|
signal.pause()
|
|
""",
|
|
),
|
|
],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
env=os.environ.copy(),
|
|
)
|
|
assert process.stdout is not None
|
|
assert process.stdout.readline() == "READY\n"
|
|
|
|
process.send_signal(sig)
|
|
stdout, stderr = process.communicate(timeout=30)
|
|
|
|
assert process.returncode == 130, (stdout, stderr)
|
|
|
|
|
|
def test_nested_foreground_signal_state_propagates_to_outer_context():
|
|
with foreground_shutdown_signals(first_signal_message=None) as outer_state:
|
|
try:
|
|
with foreground_shutdown_signals(first_signal_message=None):
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
assert outer_state.signal_name == "SIGTERM"
|
|
with pytest.raises(KeyboardInterrupt):
|
|
raise_if_shutdown_requested()
|