86b2f14a43
- persist TU-owned graphs and guarded configuration evidence - add closed-world production reachability classification - calibrate detector precision and reviewed self-scan policy - terminate poisoned workers instead of longjmp recovery
619 lines
22 KiB
Python
Executable file
619 lines
22 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Reproducible guarded whole-program reachability evaluation.
|
|
|
|
Retains every report/profile/log and measures five cold/warm/profile-change
|
|
pairs. Optional Fold/YAMS compile databases extend the same paired protocol to
|
|
production source trees. No threshold is inferred from these measurements.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shutil
|
|
import statistics
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCHEMA_VERSION = 1
|
|
EXPECTED_CLASSES = {
|
|
"dead_helper": "unreachable-production",
|
|
"middle": "default-inactive",
|
|
"impossible": "configuration-infeasible",
|
|
"test_helper": "test-only",
|
|
"test_leaf": "test-only",
|
|
}
|
|
|
|
|
|
def write_compile_db(project: Path, sources: list[Path]) -> Path:
|
|
rows = [
|
|
{
|
|
"directory": str(project),
|
|
"arguments": ["clang++", "-std=c++20", "-O0", "-c", str(source)],
|
|
"file": str(source),
|
|
}
|
|
for source in sources
|
|
]
|
|
path = project / "compile_commands.json"
|
|
path.write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def load_json(path: Path, description: str) -> Any:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
raise ValueError(f"invalid {description} at {path}: {error}") from error
|
|
|
|
|
|
def decode_json(text: str, description: str) -> Any:
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError as error:
|
|
raise ValueError(f"invalid {description}: {error}") from error
|
|
|
|
|
|
def profile_from_stdout(text: str) -> dict[str, Any]:
|
|
offsets = [match.start() for match in re.finditer(r"(?m)^\{", text)]
|
|
if not offsets:
|
|
raise ValueError("profile JSON was not present in stdout")
|
|
value = decode_json(text[offsets[-1] :], "profile JSON")
|
|
if not isinstance(value, dict) or value.get("tool") != "fold":
|
|
raise ValueError("profile JSON has the wrong envelope")
|
|
return value
|
|
|
|
|
|
def incomplete_profile() -> dict[str, Any]:
|
|
return {
|
|
"selected_tus": 0,
|
|
"attempted_tus": 0,
|
|
"completed_tus": 0,
|
|
"timed_out_tus": 0,
|
|
"frontend_failed_tus": 0,
|
|
"reused_tus": 0,
|
|
"reparsed_tus": 0,
|
|
"removed_tus": 0,
|
|
"graph_fallbacks": 0,
|
|
"closed_world": False,
|
|
"guarded_callsites": 0,
|
|
"operational_complete": False,
|
|
"solver_queries": 0,
|
|
}
|
|
|
|
|
|
def run_scan(
|
|
fold: Path,
|
|
compile_db: Path,
|
|
sources: list[Path],
|
|
profile: Path | None,
|
|
cache: Path,
|
|
output: Path,
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
command = [
|
|
str(fold),
|
|
"scan",
|
|
"--compile-db",
|
|
str(compile_db),
|
|
"--detectors",
|
|
"production-reachability",
|
|
"--format",
|
|
"findings-json",
|
|
"--output",
|
|
str(output.with_suffix(".findings.json")),
|
|
"--cache-dir",
|
|
str(cache),
|
|
"--processes",
|
|
"4",
|
|
]
|
|
if profile:
|
|
command.extend(["--config-profile", str(profile)])
|
|
command.extend(str(source) for source in sources)
|
|
command.append("--profile")
|
|
started = time.monotonic()
|
|
completed = subprocess.run(
|
|
command, text=True, capture_output=True, timeout=timeout, check=False
|
|
)
|
|
wall_seconds = time.monotonic() - started
|
|
output.with_suffix(".stdout.log").write_text(completed.stdout, encoding="utf-8")
|
|
output.with_suffix(".stderr.log").write_text(completed.stderr, encoding="utf-8")
|
|
output.with_suffix(".command.json").write_text(
|
|
json.dumps({"command": command, "returncode": completed.returncode}, indent=2)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
if completed.returncode not in (0, 1):
|
|
return {
|
|
"returncode": completed.returncode,
|
|
"status": "failed",
|
|
"wall_seconds": wall_seconds,
|
|
"classifications": {},
|
|
"removal_candidates": 0,
|
|
"profile": incomplete_profile(),
|
|
}
|
|
scan_profile = profile_from_stdout(completed.stdout)
|
|
output.with_suffix(".profile.json").write_text(
|
|
json.dumps(scan_profile, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
report = load_json(output.with_suffix(".findings.json"), "findings report")
|
|
classifications: dict[str, str] = {}
|
|
removal_candidates = 0
|
|
for finding in report.get("findings", []):
|
|
evidence = finding.get("evidence", {}).get("reachability")
|
|
if not evidence:
|
|
continue
|
|
locations = finding.get("locations", [])
|
|
function = locations[0].get("function", "") if locations else ""
|
|
function = function.split("@", 1)[0]
|
|
classifications[function] = evidence.get("classification", "")
|
|
removal_candidates += bool(evidence.get("removal_candidate"))
|
|
scan = scan_profile["scan"]
|
|
return {
|
|
"returncode": completed.returncode,
|
|
"wall_seconds": wall_seconds,
|
|
"classifications": classifications,
|
|
"removal_candidates": removal_candidates,
|
|
"profile": {
|
|
"selected_tus": scan.get("source_tus_selected", 0),
|
|
"attempted_tus": scan.get("source_tus_attempted", 0),
|
|
"completed_tus": scan.get("source_tus_completed", 0),
|
|
"timed_out_tus": scan.get("source_tus_timed_out", 0),
|
|
"frontend_failed_tus": scan.get("source_tus_frontend_failed", 0),
|
|
"reused_tus": scan.get("project_index_reused_tus", 0),
|
|
"reparsed_tus": scan.get("project_index_reparsed_tus", 0),
|
|
"removed_tus": scan.get("project_index_removed_tus", 0),
|
|
"graph_fallbacks": scan.get("project_index_fallbacks", 0),
|
|
"closed_world": scan.get("root_inventory_closed_world", False),
|
|
"guarded_callsites": scan.get("guarded_callsites", 0),
|
|
"operational_complete": scan.get("source_tus_attempted", 0)
|
|
== scan.get("source_tus_completed", 0)
|
|
and scan.get("source_tus_timed_out", 0) == 0
|
|
and scan.get("source_tus_frontend_failed", 0) == 0,
|
|
"solver_queries": next(
|
|
(
|
|
row.get("z3_queries", 0)
|
|
for row in scan_profile.get("detectors", [])
|
|
if row.get("name") == "production-reachability"
|
|
),
|
|
0,
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def percentile(values: list[float], fraction: float) -> float:
|
|
ordered = sorted(values)
|
|
if len(ordered) == 1:
|
|
return ordered[0]
|
|
position = (len(ordered) - 1) * fraction
|
|
lower = 0
|
|
while lower + 1 <= position:
|
|
lower += 1
|
|
upper = min(lower + 1, len(ordered) - 1)
|
|
weight = position - lower
|
|
return ordered[lower] * (1.0 - weight) + ordered[upper] * weight
|
|
|
|
|
|
def distribution(values: list[float]) -> dict[str, float]:
|
|
return {
|
|
"median": statistics.median(values),
|
|
"q1": percentile(values, 0.25),
|
|
"q3": percentile(values, 0.75),
|
|
"iqr": percentile(values, 0.75) - percentile(values, 0.25),
|
|
"min": min(values),
|
|
"max": max(values),
|
|
}
|
|
|
|
|
|
def changed_profile(source: Path, destination: Path) -> None:
|
|
value = load_json(source, "configuration profile")
|
|
value["assumptions"].append(
|
|
{"kind": "global", "selector": "evaluation_toggle", "value": False}
|
|
)
|
|
destination.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def paired_runs(
|
|
name: str,
|
|
fold: Path,
|
|
compile_db: Path,
|
|
sources: list[Path],
|
|
profile: Path,
|
|
artifact: Path,
|
|
repeats: int,
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
changed = artifact / f"{name}-changed-profile.json"
|
|
changed_profile(profile, changed)
|
|
rows: list[dict[str, Any]] = []
|
|
for repetition in range(1, repeats + 1):
|
|
cache = artifact / "cache" / name / str(repetition)
|
|
try:
|
|
shutil.rmtree(cache)
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError as error:
|
|
raise RuntimeError(
|
|
f"cannot clear evaluation cache {cache}: {error}"
|
|
) from error
|
|
phases: dict[str, Any] = {}
|
|
for phase, selected_profile in (
|
|
("cold", profile),
|
|
("warm", profile),
|
|
("change", changed),
|
|
):
|
|
phases[phase] = run_scan(
|
|
fold,
|
|
compile_db,
|
|
sources,
|
|
selected_profile,
|
|
cache,
|
|
artifact / "runs" / name / f"pair-{repetition}-{phase}",
|
|
timeout,
|
|
)
|
|
rows.append({"pair": repetition, **phases})
|
|
return {
|
|
"pairs": rows,
|
|
"operational_complete": all(
|
|
row[phase]["profile"]["operational_complete"]
|
|
for row in rows
|
|
for phase in ("cold", "warm", "change")
|
|
),
|
|
"wall_seconds": {
|
|
phase: distribution([row[phase]["wall_seconds"] for row in rows])
|
|
for phase in ("cold", "warm", "change")
|
|
},
|
|
"reparse_counts": {
|
|
phase: [row[phase]["profile"]["reparsed_tus"] for row in rows]
|
|
for phase in ("cold", "warm", "change")
|
|
},
|
|
"reuse_counts": {
|
|
phase: [row[phase]["profile"]["reused_tus"] for row in rows]
|
|
for phase in ("cold", "warm", "change")
|
|
},
|
|
}
|
|
|
|
|
|
def validate_fixture_pairs(paired: dict[str, Any], source_count: int) -> None:
|
|
for row in paired["pairs"]:
|
|
for phase in ("cold", "warm", "change"):
|
|
run = row[phase]
|
|
if run["classifications"] != EXPECTED_CLASSES:
|
|
raise AssertionError(
|
|
f"fixture pair {row['pair']} {phase} classifications changed: "
|
|
f"{run['classifications']}"
|
|
)
|
|
if run["removal_candidates"] != 1:
|
|
raise AssertionError(
|
|
f"fixture pair {row['pair']} {phase} must have one removal candidate"
|
|
)
|
|
if run["profile"]["graph_fallbacks"] != 0:
|
|
raise AssertionError(
|
|
f"fixture pair {row['pair']} {phase} unexpectedly fell back"
|
|
)
|
|
cold = row["cold"]["profile"]
|
|
warm = row["warm"]["profile"]
|
|
change = row["change"]["profile"]
|
|
if (cold["reparsed_tus"], cold["reused_tus"]) != (source_count, 0):
|
|
raise AssertionError(f"fixture pair {row['pair']} cold cache was not empty")
|
|
if (warm["reparsed_tus"], warm["reused_tus"]) != (0, source_count):
|
|
raise AssertionError(f"fixture pair {row['pair']} did not reuse every TU")
|
|
if (change["reparsed_tus"], change["reused_tus"]) != (source_count, 0):
|
|
raise AssertionError(
|
|
f"fixture pair {row['pair']} ignored the profile fingerprint change"
|
|
)
|
|
|
|
|
|
def production_sources(compile_db: Path, repo: Path) -> list[Path]:
|
|
rows = load_json(compile_db, "compilation database")
|
|
result: set[Path] = set()
|
|
source_root = (repo / "src").resolve()
|
|
for row in rows:
|
|
file = Path(row["file"])
|
|
if not file.is_absolute():
|
|
file = Path(row.get("directory", repo)) / file
|
|
try:
|
|
resolved = file.resolve()
|
|
resolved.relative_to(source_root)
|
|
except (OSError, ValueError):
|
|
continue
|
|
if resolved.exists() and resolved.suffix in {".c", ".cc", ".cpp", ".cxx"}:
|
|
result.add(resolved)
|
|
return sorted(result)
|
|
|
|
|
|
def bounded_project_sources(name: str, sources: list[Path], maximum: int) -> list[Path]:
|
|
if maximum <= 0 or len(sources) <= maximum:
|
|
return sources
|
|
preferred_names = {
|
|
"fold": {"quality_runner.cpp", "sarif.cpp"},
|
|
"yams": {"connection_pool.cpp", "content_index_writer.cpp"},
|
|
}.get(name, set())
|
|
preferred = [source for source in sources if source.name in preferred_names]
|
|
roots = [
|
|
source
|
|
for source in sources
|
|
if source.name in {"main.c", "main.cc", "main.cpp"} and source not in preferred
|
|
]
|
|
others = [
|
|
source for source in sources if source not in preferred and source not in roots
|
|
]
|
|
return (preferred + roots + others)[:maximum]
|
|
|
|
|
|
def compile_binary(project: Path, artifact: Path) -> dict[str, Any]:
|
|
binary = artifact / "cross-tu-binary"
|
|
compiler = shutil.which("clang++") or shutil.which("c++")
|
|
symbol_tool = shutil.which("nm")
|
|
if not compiler or not symbol_tool:
|
|
return {
|
|
"status": "unavailable",
|
|
"reason": "C++ compiler or nm was not found",
|
|
"absence_is_proof": False,
|
|
}
|
|
completed = subprocess.run(
|
|
[
|
|
compiler,
|
|
"-std=c++20",
|
|
"-O0",
|
|
str(project / "app.cpp"),
|
|
str(project / "callback.cpp"),
|
|
"-o",
|
|
str(binary),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
(artifact / "binary-build.log").write_text(
|
|
completed.stdout + completed.stderr, encoding="utf-8"
|
|
)
|
|
if completed.returncode != 0:
|
|
return {
|
|
"status": "unavailable",
|
|
"returncode": completed.returncode,
|
|
"absence_is_proof": False,
|
|
}
|
|
symbols = subprocess.run(
|
|
[symbol_tool, str(binary)], text=True, capture_output=True, check=False
|
|
)
|
|
(artifact / "binary-symbols.txt").write_text(symbols.stdout, encoding="utf-8")
|
|
return {
|
|
"status": "ok" if symbols.returncode == 0 else "unavailable",
|
|
"callback_present": "callback" in symbols.stdout,
|
|
"dead_neighbor_present": "dead_callback_neighbor" in symbols.stdout,
|
|
"absence_is_proof": False,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--fold", type=Path, default=Path("build/release/src/cli/fold"))
|
|
parser.add_argument(
|
|
"--fixtures", type=Path, default=Path("tests/fixtures/reachability")
|
|
)
|
|
parser.add_argument("--artifacts", type=Path)
|
|
parser.add_argument("--repeats", type=int, default=5)
|
|
parser.add_argument("--timeout", type=int, default=600)
|
|
parser.add_argument("--fold-compile-db", type=Path)
|
|
parser.add_argument("--yams-compile-db", type=Path)
|
|
parser.add_argument("--project-max-tus", type=int, default=24)
|
|
args = parser.parse_args()
|
|
if args.repeats < 5:
|
|
parser.error("--repeats must be at least 5 for paired evidence")
|
|
timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
artifact = (
|
|
args.artifacts
|
|
or Path(tempfile.mkdtemp(prefix=f"fold-reachability-eval-{timestamp}-"))
|
|
).resolve()
|
|
if artifact.exists() and any(artifact.iterdir()):
|
|
raise ValueError(f"artifact directory must be new or empty: {artifact}")
|
|
artifact.mkdir(parents=True, exist_ok=True)
|
|
fold = args.fold.resolve()
|
|
fixtures = args.fixtures.resolve()
|
|
|
|
manifest = load_json(fixtures / "variant_manifest.json", "variant manifest")
|
|
if manifest.get("schema_version") != SCHEMA_VERSION or not manifest.get("variants"):
|
|
raise ValueError("variant manifest must be explicit, versioned, and non-empty")
|
|
|
|
workspace = artifact / "workspace"
|
|
shutil.copytree(fixtures, workspace)
|
|
closed = workspace / "closed"
|
|
closed_sources = sorted(closed.rglob("*.cpp"))
|
|
closed_db = write_compile_db(closed, closed_sources)
|
|
closed_profile = closed / "configuration.json"
|
|
summary: dict[str, Any] = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"artifact_root": str(artifact),
|
|
"variant_manifest": manifest,
|
|
"fixture_pairs": paired_runs(
|
|
"fixture",
|
|
fold,
|
|
closed_db,
|
|
closed_sources,
|
|
closed_profile,
|
|
artifact,
|
|
args.repeats,
|
|
args.timeout,
|
|
),
|
|
}
|
|
validate_fixture_pairs(summary["fixture_pairs"], len(closed_sources))
|
|
first_closed = summary["fixture_pairs"]["pairs"][0]["cold"]
|
|
for function, classification in EXPECTED_CLASSES.items():
|
|
if first_closed["classifications"].get(function) != classification:
|
|
raise AssertionError(
|
|
f"{function}: expected {classification}, got "
|
|
f"{first_closed['classifications'].get(function)}"
|
|
)
|
|
if "unknown_target" in first_closed["classifications"]:
|
|
raise AssertionError("unknown guard emitted a stronger dead classification")
|
|
|
|
fail_closed = workspace / "fail_closed"
|
|
fail_sources = sorted(fail_closed.rglob("*.cpp"))
|
|
fail_db = write_compile_db(fail_closed, fail_sources)
|
|
fail_result = run_scan(
|
|
fold,
|
|
fail_db,
|
|
fail_sources,
|
|
None,
|
|
artifact / "cache" / "fail-closed",
|
|
artifact / "runs" / "fail-closed",
|
|
args.timeout,
|
|
)
|
|
if fail_result["removal_candidates"] or fail_result["profile"]["closed_world"]:
|
|
raise AssertionError("unresolved dynamic call did not fail closed")
|
|
summary["fail_closed"] = fail_result
|
|
|
|
renamed = workspace / "renamed"
|
|
shutil.copytree(closed, renamed)
|
|
renamed_library = renamed / "library.cpp"
|
|
renamed_library.write_text(
|
|
renamed_library.read_text(encoding="utf-8").replace(
|
|
"dead_helper", "renamed_dead_helper"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
renamed_sources = sorted(renamed.rglob("*.cpp"))
|
|
renamed_db = write_compile_db(renamed, renamed_sources)
|
|
renamed_result = run_scan(
|
|
fold,
|
|
renamed_db,
|
|
renamed_sources,
|
|
renamed / "configuration.json",
|
|
artifact / "cache" / "renamed",
|
|
artifact / "runs" / "renamed",
|
|
args.timeout,
|
|
)
|
|
base_classes = sorted(first_closed["classifications"].values())
|
|
if sorted(renamed_result["classifications"].values()) != base_classes:
|
|
raise AssertionError(
|
|
"renaming changed the reachability classification multiset"
|
|
)
|
|
summary["metamorphic_rename"] = {
|
|
"base_classes": base_classes,
|
|
"renamed_classes": sorted(renamed_result["classifications"].values()),
|
|
"stable": True,
|
|
}
|
|
|
|
cross_tu = workspace / "cross_tu"
|
|
cross_sources = sorted(cross_tu.rglob("*.cpp"))
|
|
cross_db = write_compile_db(cross_tu, cross_sources)
|
|
cross_result = run_scan(
|
|
fold,
|
|
cross_db,
|
|
cross_sources,
|
|
None,
|
|
artifact / "cache" / "cross-tu",
|
|
artifact / "runs" / "cross-tu",
|
|
args.timeout,
|
|
)
|
|
if (
|
|
cross_result["classifications"].get("dead_callback_neighbor")
|
|
!= "unreachable-production"
|
|
):
|
|
raise AssertionError(
|
|
"cross-TU callback root did not preserve the neighboring dead result"
|
|
)
|
|
if "callback" in cross_result["classifications"]:
|
|
raise AssertionError("registered cross-TU callback was classified inactive")
|
|
summary["cross_tu"] = cross_result
|
|
summary["binary_provenance"] = compile_binary(cross_tu, artifact)
|
|
|
|
project_inputs = (
|
|
("fold", args.fold_compile_db, Path.cwd()),
|
|
("yams", args.yams_compile_db, Path.cwd().parent / "yams"),
|
|
)
|
|
for name, compile_db, repo in project_inputs:
|
|
if not compile_db:
|
|
summary[name] = {
|
|
"status": "skipped",
|
|
"reason": "compile database not supplied",
|
|
}
|
|
continue
|
|
compile_db = compile_db.resolve()
|
|
if not compile_db.exists():
|
|
summary[name] = {
|
|
"status": "skipped",
|
|
"reason": "compile database unavailable",
|
|
}
|
|
continue
|
|
available_sources = production_sources(compile_db, repo.resolve())
|
|
sources = bounded_project_sources(name, available_sources, args.project_max_tus)
|
|
profile = artifact / f"{name}-profile.json"
|
|
profile.write_text(
|
|
json.dumps(
|
|
{"version": 1, "entrypoints": ["main"], "assumptions": []}, indent=2
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
summary[name] = {
|
|
"status": "ok",
|
|
"available_sources": len(available_sources),
|
|
"selected_sources": len(sources),
|
|
"selection": "calibration translation units, then main, then canonical path order",
|
|
"paired": paired_runs(
|
|
name,
|
|
fold,
|
|
compile_db,
|
|
sources,
|
|
profile,
|
|
artifact,
|
|
args.repeats,
|
|
args.timeout,
|
|
),
|
|
}
|
|
|
|
operational_failures = []
|
|
if not summary["fixture_pairs"]["operational_complete"]:
|
|
operational_failures.append("fixture-pairs")
|
|
if not fail_result["profile"]["operational_complete"]:
|
|
operational_failures.append("fail-closed")
|
|
if not cross_result["profile"]["operational_complete"]:
|
|
operational_failures.append("cross-tu")
|
|
for name in ("fold", "yams"):
|
|
lane = summary[name]
|
|
if lane.get("status") == "ok" and not lane["paired"]["operational_complete"]:
|
|
operational_failures.append(name)
|
|
summary["operational_coverage"] = {
|
|
"complete": not operational_failures,
|
|
"failures": operational_failures,
|
|
}
|
|
|
|
summary_path = artifact / "summary.json"
|
|
summary_path.write_text(
|
|
json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
(artifact / "summary.md").write_text(
|
|
"# Fold reachability evaluation\n\n"
|
|
f"Artifact root: `{artifact}`\n\n"
|
|
f"Paired repetitions: {args.repeats}\n\n"
|
|
"Cold, warm, and profile-change measurements are descriptive; this run does not invent "
|
|
"performance thresholds. Binary symbol presence is corroboration and absence is never "
|
|
"treated as source-unreachability proof.\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(summary_path)
|
|
return 0 if not operational_failures else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (
|
|
AssertionError,
|
|
OSError,
|
|
RuntimeError,
|
|
ValueError,
|
|
subprocess.TimeoutExpired,
|
|
) as error:
|
|
print(f"reachability evaluation failed: {error}", file=sys.stderr)
|
|
raise SystemExit(2) from error
|