wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(signoz): add recoverable Agent99 metadata toolchain route

Browse Source
This commit is contained in:
Your Name
2026-07-22 21:02:02 +08:00
parent d097013a03
commit 2bff3edcff
23 changed files with 3510 additions and 1111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
scripts/backup/tests/test_signoz_metadata_export_contract.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import hashlib
import importlib.util
import json
import os
import socket
@@ -11,13 +12,16 @@ import urllib.parse
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Iterator
ROOT = Path(__file__).resolve().parents[3]
EXPORTER = ROOT / "scripts" / "backup" / "signoz-metadata-export.py"
VERIFIER = ROOT / "scripts" / "backup" / "verify-signoz-metadata-export.py"
RESTORE = ROOT / "scripts" / "backup" / "signoz-metadata-restore-drill.py"
ISOLATED_CONTROLLER = (
ROOT / "scripts" / "backup" / "signoz-metadata-isolated-restore.py"
)
POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json"
TRACE_ID = "trace-P0-OBS-002-metadata"
RUN_ID = "run-P0-OBS-002-metadata"
@@ -106,9 +110,7 @@ class MetadataHandler(BaseHTTPRequestHandler):
if source_page not in {"logs", "meter", "metrics", "traces"}:
self._send_json(400, {"error": "sourcePage required"})
return
asset_key = (
f"{path}?{urllib.parse.urlencode({'sourcePage': source_page})}"
)
asset_key = f"{path}?{urllib.parse.urlencode({'sourcePage': source_page})}"
if asset_key not in self.assets:
self._send_json(404, {"error": "not found"})
return
@@ -220,6 +222,21 @@ def create_bundle(tmp_path: Path) -> tuple[Path, Path]:
return bundle, credential
def load_isolated_controller() -> Any:
backup_dir = str(ROOT / "scripts" / "backup")
if backup_dir not in sys.path:
sys.path.insert(0, backup_dir)
spec = importlib.util.spec_from_file_location(
"signoz_isolated_restore_contract_integration",
ISOLATED_CONTROLLER,
)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path:
policy = json.loads(POLICY.read_text(encoding="utf-8"))
identity = hashlib.sha256(
@@ -373,9 +390,7 @@ def test_policy_rejects_unhashable_empty_collection_sentinel_as_contract_error(
) -> None:
policy_value = json.loads(POLICY.read_text(encoding="utf-8"))
explorer = next(
item
for item in policy_value["assets"]
if item["id"] == "explorer_views_logs"
item for item in policy_value["assets"] if item["id"] == "explorer_views_logs"
)
explorer["empty_collection_sentinels"] = [{}]
malformed_policy = tmp_path / "malformed-policy.json"
@@ -549,9 +564,7 @@ def test_explorer_views_nonempty_scalar_fails_closed(tmp_path: Path) -> None:
def test_explorer_view_item_must_match_scoped_source_page(tmp_path: Path) -> None:
assets = json.loads(json.dumps(SOURCE_ASSETS))
assets["/api/v1/explorer/views?sourcePage=logs"]["data"][0][
"sourcePage"
] = "traces"
assets["/api/v1/explorer/views?sourcePage=logs"]["data"][0]["sourcePage"] = "traces"
output = tmp_path / "bundle"
credential = credential_file(tmp_path)
with metadata_server(assets) as (base_url, _):
@@ -752,6 +765,68 @@ def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None:
assert handler.post_count == 3
def test_isolated_controller_invokes_real_restore_driver_contract(
tmp_path: Path,
) -> None:
controller = load_isolated_controller()
bundle, credential = create_bundle(tmp_path)
target_assets = {
"/api/v1/dashboards": {"data": []},
"/api/v1/rules": {"status": "success", "data": {"rules": []}},
"/api/v1/explorer/views?sourcePage=logs": {"data": []},
"/api/v1/explorer/views?sourcePage=meter": {"data": None},
"/api/v1/explorer/views?sourcePage=metrics": {"data": {}},
"/api/v1/explorer/views?sourcePage=traces": {"data": []},
}
args = SimpleNamespace(
bundle_dir=bundle,
policy=POLICY,
trace_id=TRACE_ID,
run_id=RUN_ID,
work_item_id=WORK_ITEM_ID,
)
def real_child_runner(argv: list[str], timeout: int) -> Any:
assert argv[0] == controller.PYTHON
completed = subprocess.run(
[sys.executable, *argv[1:]],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
timeout=timeout,
check=False,
)
return controller.CommandResult(completed.returncode, completed.stdout)
with metadata_server(target_assets) as (target_base_url, handler):
isolation = isolation_receipt(tmp_path, target_base_url)
check = controller.invoke_restore_driver(
args,
real_child_runner,
target_base_url,
isolation,
credential,
"--check",
)
apply = controller.invoke_restore_driver(
args,
real_child_runner,
target_base_url,
isolation,
credential,
"--apply",
)
assert check["terminal"] == "check_pass_no_write"
assert apply["terminal"] == "partial_degraded_cleanup_pending"
assert set(apply["restored_asset_counts"]) == set(controller.PORTABLE_ASSET_IDS)
assert handler.post_count == 3
child_receipt = tmp_path / "restore-driver-apply-receipt.json"
assert child_receipt.is_file()
assert child_receipt.read_bytes() == controller.canonical_json_bytes(apply)
def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> None:
bundle, credential = create_bundle(tmp_path)
target = "https://signoz.example.invalid"

367
scripts/backup/tests/test_signoz_metadata_isolated_restore.py
View File

@@ -11,7 +11,6 @@ from typing import Any
import pytest
ROOT = Path(__file__).resolve().parents[3]
CONTROLLER = ROOT / "scripts" / "backup" / "signoz-metadata-isolated-restore.py"
POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json"
@@ -91,6 +90,12 @@ class RecordingRunner:
if argv[1:2] == ["run"]:
name = argv[argv.index("--name") + 1]
ref = argv[-1]
role = next(
candidate
for candidate, (expected_ref, _image_id) in c.EXPECTED_IMAGES.items()
if expected_ref == ref
)
limits = c.RESOURCE_LIMITS[role]
expected_id = next(
image_id
for expected_ref, image_id in c.EXPECTED_IMAGES.values()
@@ -100,16 +105,27 @@ class RecordingRunner:
"Id": (name.encode().hex() + "0" * 64)[:64],
"Image": expected_id,
"Config": {"Image": ref, "Labels": labels_from_command(argv)},
"HostConfig": {"RestartPolicy": {"Name": "no"}},
"HostConfig": {
"RestartPolicy": {"Name": "no"},
"NanoCpus": limits["nano_cpus"],
"Memory": limits["memory_bytes"],
"MemorySwap": limits["memory_swap_bytes"],
"PidsLimit": limits["pids_limit"],
"LogConfig": {
"Type": c.RESOURCE_LOG_CONFIG["type"],
"Config": {
"max-file": c.RESOURCE_LOG_CONFIG["max_file"],
"max-size": c.RESOURCE_LOG_CONFIG["max_size"],
},
},
},
"State": {"Running": True},
"NetworkSettings": {
"Ports": {
"8080/tcp": [
{"HostIp": "127.0.0.1", "HostPort": "49173"}
]
}
if name.endswith("-server")
else {}
"Ports": (
{"8080/tcp": [{"HostIp": "127.0.0.1", "HostPort": "49173"}]}
if name.endswith("-server")
else {}
)
},
"RestartCount": 0,
}
@@ -161,9 +177,7 @@ def labels_from_command(argv: list[str]) -> dict[str, str]:
def test_identity_and_exact_resource_names(controller: Any) -> None:
plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID)
assert plan.identity == controller.derive_identity(
TRACE_ID, RUN_ID, WORK_ITEM_ID
)
assert plan.identity == controller.derive_identity(TRACE_ID, RUN_ID, WORK_ITEM_ID)
assert len(plan.identity) == 64
assert plan.prefix == f"awoooi-signoz-md-restore-{plan.identity[:16]}"
assert plan.canonical_id == (
@@ -180,9 +194,10 @@ def test_identity_and_exact_resource_names(controller: Any) -> None:
def test_policy_requires_reviewed_exact_startup_and_images(controller: Any) -> None:
policy = json.loads(POLICY.read_text(encoding="utf-8"))
assert controller.validate_restore_policy(policy)["startup_contract"][
"reviewed"
] is True
isolation = controller.validate_restore_policy(policy)
assert isolation["startup_contract"]["reviewed"] is True
assert isolation["resource_limits"] == controller.RESOURCE_LIMITS
assert isolation["resource_log_config"] == controller.RESOURCE_LOG_CONFIG
policy["restore_isolation"]["startup_contract"]["reviewed"] = False
with pytest.raises(
@@ -191,14 +206,19 @@ def test_policy_requires_reviewed_exact_startup_and_images(controller: Any) -> N
controller.validate_restore_policy(policy)
policy = json.loads(POLICY.read_text(encoding="utf-8"))
policy["restore_isolation"]["images"]["clickhouse"]["id"] = (
f"sha256:{'a' * 64}"
)
policy["restore_isolation"]["images"]["clickhouse"]["id"] = f"sha256:{'a' * 64}"
with pytest.raises(
controller.ContractError, match="restore_image_policy_not_exact"
):
controller.validate_restore_policy(policy)
policy = json.loads(POLICY.read_text(encoding="utf-8"))
policy["assets"][0]["classification"] = "export_only"
with pytest.raises(
controller.ContractError, match="restore_portable_asset_ids_not_exact"
):
controller.validate_restore_policy(policy)
def test_resource_creation_is_internal_no_pull_no_restart_and_loopback(
controller: Any,
@@ -210,7 +230,9 @@ def test_resource_creation_is_internal_no_pull_no_restart_and_loopback(
controller.create_resources(runner, plan, RUN_ID, cluster)
commands = runner.commands
network = next(command for command in commands if command[1:3] == ["network", "create"])
network = next(
command for command in commands if command[1:3] == ["network", "create"]
)
assert "--internal" in network
run_commands = [command for command in commands if command[1:2] == ["run"]]
assert len(run_commands) == 3
@@ -220,21 +242,50 @@ def test_resource_creation_is_internal_no_pull_no_restart_and_loopback(
assert f"{controller.PRIMARY_LABEL}={RUN_ID}" in command
assert f"{controller.IDENTITY_LABEL}={plan.identity}" in command
assert not any("signoz-clickhouse:" in value for value in command)
server = next(command for command in run_commands if plan.server_container in command)
ref = command[-1]
role = next(
candidate
for candidate, (
expected_ref,
_image_id,
) in controller.EXPECTED_IMAGES.items()
if expected_ref == ref
)
limits = controller.RESOURCE_LIMITS[role]
assert command[command.index("--cpus") + 1] == limits["cpus"]
assert command[command.index("--memory") + 1] == limits["memory"]
assert command[command.index("--memory-swap") + 1] == limits["memory_swap"]
assert command[command.index("--pids-limit") + 1] == str(limits["pids_limit"])
assert command[command.index("--log-driver") + 1] == "local"
assert "max-size=10m" in command
assert "max-file=2" in command
server = next(
command for command in run_commands if plan.server_container in command
)
assert "127.0.0.1::8080" in server
assert f"{plan.server_data_volume}:/var/lib/signoz" in server
assert "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN=tcp://clickhouse:9000" in server
clickhouse_run_index = commands.index(
next(command for command in run_commands if plan.clickhouse_container in command)
next(
command for command in run_commands if plan.clickhouse_container in command
)
)
readiness_index = next(
index
for index, command in enumerate(commands)
if command[1:2] == ["exec"]
and command[-2:] == ["--query", "SELECT 1"]
if command[1:2] == ["exec"] and command[-2:] == ["--query", "SELECT 1"]
)
server_run_index = commands.index(server)
assert clickhouse_run_index < readiness_index < server_run_index
controller.verify_runtime_container_images(runner, plan)
server_runtime = runner.resources[("container", plan.server_container)]
server_runtime["HostConfig"]["Memory"] += 1
with pytest.raises(
controller.ContractError,
match="restore_runtime_container_image_restart_or_resource_limit_drift",
):
controller.verify_runtime_container_images(runner, plan)
def test_exact_image_drift_fails_before_resource_creation(controller: Any) -> None:
@@ -320,8 +371,7 @@ def test_cleanup_reconciles_without_listener_state_after_interruption(
controller.create_workspace(plan)
if server_created:
cluster = (
ROOT
/ "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
)
controller.create_resources(runner, plan, RUN_ID, cluster)
state = {
@@ -390,8 +440,7 @@ def test_cleanup_reconciles_resource_materialized_after_first_inventory(
removals = [
command
for command in runner.commands
if command[1:3] == ["volume", "rm"]
and command[-1] == plan.server_data_volume
if command[1:3] == ["volume", "rm"] and command[-1] == plan.server_data_volume
]
assert len(removals) == 1
@@ -441,8 +490,15 @@ def test_distinct_verifier_requires_immutable_apply_receipt(
"restoredAssetCounts": {
"dashboards": 1,
"alert_rules": 2,
"explorer_views": 3,
"explorer_views_logs": 3,
"explorer_views_meter": 0,
"explorer_views_metrics": 0,
"explorer_views_traces": 0,
},
"listenerBinding": {"host": "127.0.0.1", "port": 49173},
"productionContinuitySha256": "b" * 64,
"imageInventorySha256": "c" * 64,
"exportBundleManifestSha256": "d" * 64,
},
)
path = tmp_path / "apply.json"
@@ -453,12 +509,152 @@ def test_distinct_verifier_requires_immutable_apply_receipt(
assert executor_receipt["completionClaim"] is False
assert executor_receipt["executorZeroResidueVerified"] is True
assert executor_receipt["portableAssetIds"] == [
"dashboards",
"alert_rules",
"explorer_views_logs",
"explorer_views_meter",
"explorer_views_metrics",
"explorer_views_traces",
]
assert verified["phase"] == "independent_restore_verifier"
assert verified["completionClaim"] is True
assert verified["independentZeroResidueVerified"] is True
assert verified["restoreSemanticDigest"] == "a" * 64
assert verified["applyReceiptValidated"] is True
assert len(verified["applyReceiptSha256"]) == 64
assert verified["listenerBinding"] == {
"host": "127.0.0.1",
"port": 49173,
}
assert verified["productionContinuityVerified"] is True
assert verified["imageInventoryContinuityVerified"] is True
assert verified["listenerBindingUnreachableVerified"] is True
stale_receipt = json.loads(json.dumps(executor_receipt))
del stale_receipt["restoredAssetCounts"]["explorer_views_meter"]
stale_path = tmp_path / "stale-apply.json"
controller.write_terminal(stale_path, stale_receipt)
with pytest.raises(
controller.ContractError,
match="apply_receipt_semantic_evidence_invalid",
):
controller.validate_apply_receipt(stale_path, args, plan)
def test_independent_cleanup_uses_apply_listener_and_continuity_after_workspace_gone(
controller: Any,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
runner = RecordingRunner(controller)
plan = controller.build_resource_plan(
TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
)
production_digest = controller.sha256_bytes(
controller.canonical_json_bytes(controller.production_snapshot(runner))
)
inventory_digest = controller.sha256_bytes(
controller.canonical_json_bytes(controller.image_inventory(runner))
)
continuity = {
"listenerBinding": {"host": "127.0.0.1", "port": 49173},
"productionContinuitySha256": production_digest,
"imageInventorySha256": inventory_digest,
}
probed: list[tuple[str, int]] = []
def probe(host: str, port: int) -> bool:
probed.append((host, port))
return False
monkeypatch.setattr(controller, "tcp_connectable", probe)
monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None)
errors = controller.cleanup_and_verify(
runner,
plan,
RUN_ID,
None,
continuity,
)
assert errors == []
assert probed == [("127.0.0.1", 49173)] * controller.CLEANUP_STABLE_PASSES
assert not plan.workspace.exists()
def test_restore_driver_apply_uses_and_validates_private_child_receipt(
controller: Any,
tmp_path: Path,
) -> None:
args = SimpleNamespace(
bundle_dir=tmp_path / "bundle",
policy=POLICY,
trace_id=TRACE_ID,
run_id=RUN_ID,
work_item_id=WORK_ITEM_ID,
)
isolation_path = tmp_path / "isolation-receipt.json"
isolation_path.write_text("{}", encoding="utf-8")
token_path = tmp_path / "session-access-token"
token_path.write_text("test-only-token\n", encoding="ascii")
commands: list[list[str]] = []
def child_runner(argv: list[str], _timeout: int) -> Any:
commands.append(argv[:])
receipt_path = Path(argv[argv.index("--receipt-file") + 1])
value = controller.receipt(
trace_id=TRACE_ID,
run_id=RUN_ID,
work_item_id=WORK_ITEM_ID,
phase="terminal",
terminal="partial_degraded_cleanup_pending",
detail="isolated_restore_actions_3_semantic_readback_pass_cleanup_pending",
)
value.update(
{
"restored_asset_counts": {
"dashboards": 1,
"alert_rules": 1,
"explorer_views_logs": 1,
"explorer_views_meter": 0,
"explorer_views_metrics": 0,
"explorer_views_traces": 0,
},
"restored_asset_semantic_sha256": {
asset_id: "a" * 64 for asset_id in controller.PORTABLE_ASSET_IDS
},
"portable_semantic_sha256": "b" * 64,
"completion_scope": "portable_assets_only",
"completion_claim": False,
"full_sqlite_completion_claim": False,
"roles_preferences_global_config_restore_claim": False,
"secret_bearing_assets_restore_claim": False,
}
)
receipt_path.write_bytes(controller.canonical_json_bytes(value))
receipt_path.chmod(0o600)
return controller.CommandResult(
0,
json.dumps(value, sort_keys=True, separators=(",", ":")),
)
result = controller.invoke_restore_driver(
args,
child_runner,
"http://127.0.0.1:49173",
isolation_path,
token_path,
"--apply",
)
assert result["portable_semantic_sha256"] == "b" * 64
assert len(commands) == 1
assert "--receipt-file" in commands[0]
child_path = Path(commands[0][commands[0].index("--receipt-file") + 1])
assert child_path == tmp_path / "restore-driver-apply-receipt.json"
assert stat.S_IMODE(child_path.stat().st_mode) == 0o600
def test_interrupted_verify_cleanup_reconciles_again_before_zero_residue_receipt(
@@ -477,6 +673,7 @@ def test_interrupted_verify_cleanup_reconciles_again_before_zero_residue_receipt
_plan: Any,
_run_id: str,
_state: dict[str, Any] | None,
_continuity: dict[str, Any] | None = None,
) -> list[str]:
nonlocal calls
calls += 1
@@ -514,6 +711,94 @@ def test_interrupted_verify_cleanup_reconciles_again_before_zero_residue_receipt
assert "signal_15_cleanup_required" in emitted[-1]["detail"]
def test_signal_at_cleanup_transition_still_reconciles_and_writes_terminal(
controller: Any,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
plan = controller.build_resource_plan(
TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path
)
receipt_path = tmp_path / "transition-terminal.json"
args = SimpleNamespace(
check=False,
apply=True,
verify_cleanup=False,
bundle_dir=str(tmp_path / "bundle"),
policy=str(POLICY),
cluster_config=str(
ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
),
trace_id=TRACE_ID,
run_id=RUN_ID,
work_item_id=WORK_ITEM_ID,
receipt_file=str(receipt_path),
apply_receipt_file=None,
)
state = {
"production_before": {},
"image_inventory_before": ["sha256:test|test:fixture"],
"listener": {"host": "127.0.0.1", "port": 49173},
}
cleanup_calls = 0
installed: dict[int, Any] = {}
transition_signal_sent = False
def fake_getsignal(signum: int) -> Any:
return installed.get(signum, "original-handler")
def fake_signal(signum: int, handler: Any) -> Any:
nonlocal transition_signal_sent
previous = installed.get(signum, "original-handler")
installed[signum] = handler
if (
handler is controller.signal.SIG_IGN
and not transition_signal_sent
and callable(previous)
):
transition_signal_sent = True
previous(controller.signal.SIGTERM, None)
return previous
def cleanup_once(
_runner: Any,
_plan: Any,
_run_id: str,
_state: dict[str, Any] | None,
_continuity: dict[str, Any] | None = None,
) -> list[str]:
nonlocal cleanup_calls
cleanup_calls += 1
return []
monkeypatch.setattr(controller, "parse_args", lambda: args)
monkeypatch.setattr(controller, "build_resource_plan", lambda *_args: plan)
monkeypatch.setattr(controller, "load_policy", lambda _path: {})
monkeypatch.setattr(
controller,
"run_apply",
lambda *_args: {
"restoreSemanticDigest": "a" * 64,
"restoredAssetCounts": {},
},
)
monkeypatch.setattr(controller, "load_state_if_safe", lambda _plan: state)
monkeypatch.setattr(controller, "cleanup_and_verify", cleanup_once)
monkeypatch.setattr(controller.signal, "getsignal", fake_getsignal)
monkeypatch.setattr(controller.signal, "signal", fake_signal)
exit_code = controller.main()
terminal = json.loads(receipt_path.read_text(encoding="utf-8"))
assert exit_code == 1
assert transition_signal_sent is True
assert cleanup_calls == 1
assert terminal["terminal"] == "failed_zero_residue_verified_no_completion"
assert terminal["zeroResidueVerified"] is True
assert terminal["cleanupErrorCodes"] == []
assert "signal_15_cleanup_required" in terminal["detail"]
def test_bootstrap_secrets_stay_in_mode_0400_files(
controller: Any,
tmp_path: Path,
@@ -548,9 +833,7 @@ def test_bootstrap_secrets_stay_in_mode_0400_files(
assert token == tmp_path / "session-access-token"
assert stat.S_IMODE(token.stat().st_mode) == 0o400
assert sorted(path.name for path in tmp_path.iterdir()) == [
"session-access-token"
]
assert sorted(path.name for path in tmp_path.iterdir()) == ["session-access-token"]
assert calls[0][0].endswith("/api/v1/register")
assert calls[1][0].endswith("/api/v2/sessions/email_password")
assert capsys.readouterr().out == ""
@@ -571,9 +854,7 @@ def test_bootstrap_accepts_live_root_response_envelope(
monkeypatch.setattr(controller, "request_json", fake_request)
token = controller.bootstrap_isolated_session(
"http://127.0.0.1:49173", tmp_path
)
token = controller.bootstrap_isolated_session("http://127.0.0.1:49173", tmp_path)
assert token.read_text(encoding="ascii").strip() == "a" * 64
assert stat.S_IMODE(token.stat().st_mode) == 0o400
@@ -589,8 +870,22 @@ def test_controller_source_has_signal_cleanup_and_no_secret_argv() -> None:
assert '"--pull=never"' in source
assert '"--restart=no"' in source
assert '"--auth-mode"' in source
assert "access_token" not in source[source.index("argv = [") : source.index("output = run_required", source.index("argv = ["))]
assert "password" not in source[source.index("argv = [") : source.index("output = run_required", source.index("argv = ["))]
assert (
"access_token"
not in source[
source.index("argv = [") : source.index(
"output = run_required", source.index("argv = [")
)
]
)
assert (
"password"
not in source[
source.index("argv = [") : source.index(
"output = run_required", source.index("argv = [")
)
]
)
durable_write = source.rindex("write_terminal(Path(args.receipt_file), result)")
final_handler_restore = source.rindex("restore_signal_handlers()")
assert durable_write < final_handler_restore