From de9696a6e15a1aebd1190782e8d1986ef65cc314 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 16 Jul 2026 12:56:30 +0200 Subject: [PATCH] fix: partition drift probe failures --- ops/detect_vps_gcp_source_drift.py | 114 +++++++++---- tests/test_detect_vps_gcp_source_drift.py | 191 ++++++++++++++++++++++ 2 files changed, 271 insertions(+), 34 deletions(-) diff --git a/ops/detect_vps_gcp_source_drift.py b/ops/detect_vps_gcp_source_drift.py index 3af480f..efceef7 100644 --- a/ops/detect_vps_gcp_source_drift.py +++ b/ops/detect_vps_gcp_source_drift.py @@ -42,6 +42,9 @@ COMMIT_RE = re.compile(r"[0-9a-f]{40}\Z") ROWSET_MD5_RE = re.compile(r"[0-9a-f]{32}\Z") ROUTE_PREFIX = "LEO_SOURCE_ROUTE\t" EXIT_CODE_BY_STATUS = {"identical": 0, "drift": 1, "incomplete": 2, "invalid": 3} +SSH_TRANSPORT_EXIT_CODE = 255 +PSQL_CONNECTION_EXIT_CODE = 2 +INVALID_PROBE_EXIT_CODE = 3 class DriftDetectorError(RuntimeError): @@ -71,6 +74,21 @@ def same_network_address(left: Any, right: Any) -> bool: return str(left) == str(right) +def classify_nonzero_probe_result( + returncode: int, + route: dict[str, Any] | None, +) -> tuple[str, str]: + if returncode == SSH_TRANSPORT_EXIT_CODE: + return "unreachable", "ssh_transport_failed" + if route is None: + return "invalid", "remote_precondition_failed_before_route" + if returncode == PSQL_CONNECTION_EXIT_CODE: + return "unreachable", "database_connection_unavailable" + if returncode == INVALID_PROBE_EXIT_CODE: + return "invalid", "manifest_or_probe_script_failed" + return "invalid", "remote_probe_failed_after_route" + + def parse_manifest_lines(lines: list[str], label: str) -> dict[str, Any]: singleton: dict[str, dict[str, Any]] = {} tables: dict[str, dict[str, Any]] = {} @@ -463,7 +481,7 @@ def remote_script( inventory, f'password="$(gcloud secrets versions access latest --secret={shlex.quote(gcp_secret)} ' f'--project={shlex.quote(gcp_project)} --quiet)"', - "[[ -n \"$password\" ]] || { printf 'Cloud SQL credential resolved empty\\n' >&2; exit 2; }", + "[[ -n \"$password\" ]] || { printf 'Cloud SQL credential resolved empty\\n' >&2; exit 3; }", 'export PGPASSWORD="$password"', "export PGOPTIONS='-c default_transaction_read_only=on'", f"exec psql {shlex.quote(connection)} -X -Atq -v ON_ERROR_STOP=1", @@ -491,19 +509,51 @@ def collect_endpoint( ) -> dict[str, Any]: started = utc_now() - def invalid_observation(error: str, route: dict[str, Any] | None) -> dict[str, Any]: - return { + def invalid_observation( + error: str, + route: dict[str, Any] | None, + *, + failure_class: str = "invalid_probe_receipt", + exit_code: int | None = None, + ) -> dict[str, Any]: + observation = { "schema": OBSERVATION_SCHEMA, "label": label, "status": "invalid", "observed_at_utc": started, - "failure_class": "invalid_probe_receipt", + "failure_class": failure_class, "error": error, "expected_target": expected_target, "manifest": None, "route": route, "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, } + if exit_code is not None: + observation["exit_code"] = exit_code + return observation + + def unreachable_observation( + error: str, + route: dict[str, Any] | None, + *, + failure_class: str, + exit_code: int | None = None, + ) -> dict[str, Any]: + observation = { + "schema": OBSERVATION_SCHEMA, + "label": label, + "status": "unreachable", + "observed_at_utc": started, + "failure_class": failure_class, + "error": error, + "expected_target": expected_target, + "manifest": None, + "route": route, + "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, + } + if exit_code is not None: + observation["exit_code"] = exit_code + return observation try: completed = subprocess.run( @@ -514,18 +564,11 @@ def collect_endpoint( check=False, ) except subprocess.TimeoutExpired: - return { - "schema": OBSERVATION_SCHEMA, - "label": label, - "status": "unreachable", - "observed_at_utc": started, - "failure_class": "probe_timeout", - "error": f"{label} probe exceeded {timeout}s", - "expected_target": expected_target, - "manifest": None, - "route": None, - "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, - } + return unreachable_observation( + f"{label} probe exceeded {timeout}s", + None, + failure_class="probe_timeout", + ) stdout = completed.stdout.decode(errors="replace") stderr = completed.stderr.decode(errors="replace") lines = stdout.splitlines() @@ -553,28 +596,31 @@ def collect_endpoint( return invalid_observation(route_error, route) if completed.returncode != 0: if route_error is not None and route_lines: - return invalid_observation(route_error, route) + return invalid_observation( + route_error, + route, + failure_class="malformed_probe_output", + exit_code=completed.returncode, + ) try: target = command[command.index("--") - 1] except (ValueError, IndexError): target = "" - return { - "schema": OBSERVATION_SCHEMA, - "label": label, - "status": "unreachable", - "observed_at_utc": started, - "failure_class": ( - "database_probe_failed_after_route_observed" - if route is not None - else "transport_or_remote_probe_failed" - ), - "exit_code": completed.returncode, - "error": sanitize_error(stderr, target=target), - "expected_target": expected_target, - "manifest": None, - "route": route, - "cleanup": {"remote_files_created": False, "local_temp_files_created": False}, - } + error = sanitize_error(stderr, target=target) + status, failure_class = classify_nonzero_probe_result(completed.returncode, route) + if status == "unreachable": + return unreachable_observation( + error, + route, + failure_class=failure_class, + exit_code=completed.returncode, + ) + return invalid_observation( + error, + route, + failure_class=failure_class, + exit_code=completed.returncode, + ) if route is None: return invalid_observation(route_error or f"{label} route observation is missing", None) try: diff --git a/tests/test_detect_vps_gcp_source_drift.py b/tests/test_detect_vps_gcp_source_drift.py index 585f560..6bfac45 100644 --- a/tests/test_detect_vps_gcp_source_drift.py +++ b/tests/test_detect_vps_gcp_source_drift.py @@ -158,6 +158,35 @@ def live_args(tmp_path: Path, manifest_bytes: bytes) -> argparse.Namespace: ) +def collect_completed_probe( + monkeypatch: pytest.MonkeyPatch, + *, + label: str, + returncode: int, + route: dict | None, + stderr: str, +) -> dict: + expected = endpoint_target(label) + stdout = b"" if route is None else (detector.ROUTE_PREFIX + json.dumps(route) + "\n").encode() + monkeypatch.setattr( + detector.subprocess, + "run", + lambda *args, **kwargs: detector.subprocess.CompletedProcess( + args[0], + returncode, + stdout, + stderr.encode(), + ), + ) + return detector.collect_endpoint( + label=label, + command=["ssh", expected["transport"], "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target=expected, + ) + + def test_identical_current_observations_pass() -> None: receipt = detector.evaluate_observations( observation("vps"), @@ -198,6 +227,26 @@ def test_vps_remote_script_executes_manifest_against_inspected_container_id() -> assert "exec docker exec -e PGOPTIONS=-c default_transaction_read_only=on -i teleo-pg" not in script +def test_gcp_remote_script_declares_empty_credentials_invalid() -> None: + script = detector.remote_script( + route_kind="gcp_private_cloudsql", + transport="teleo-gcp-staging", + service="leoclean-gcp-prod-parallel.service", + database="teleo", + container=None, + database_user="postgres", + gcp_host="10.61.0.3", + gcp_project="teleo-501523", + gcp_secret="gcp-teleo-pgvector-standby-postgres-password", + ) + + syntax = detector.subprocess.run(["bash", "-n"], input=script, text=True, capture_output=True, check=False) + + assert syntax.returncode == 0, syntax.stderr + credential_guard = next(line for line in script.splitlines() if "credential resolved empty" in line) + assert "exit 3" in credential_guard + + def test_content_drift_fails_and_names_exact_table_mismatch() -> None: gcp_manifest = manifest(label="gcp", row_count=4, rowset_md5="2" * 32) receipt = detector.evaluate_observations(observation("vps"), observation("gcp", gcp_manifest)) @@ -379,6 +428,108 @@ def test_runtime_revision_mismatch_is_drift() -> None: assert "runtime_commit_mismatch" in receipt["comparison"]["problems"] +def test_collect_ssh_255_without_route_is_unreachable(monkeypatch: pytest.MonkeyPatch) -> None: + result = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=255, + route=None, + stderr="ssh: connect to host teleo-gcp-staging port 22: timed out password=hunter2", + ) + + assert result["status"] == "unreachable" + assert result["failure_class"] == "ssh_transport_failed" + assert result["exit_code"] == 255 + assert result["route"] is None + assert "teleo-gcp-staging" not in result["error"] + assert "hunter2" not in result["error"] + + +@pytest.mark.parametrize("returncode", [1, 3]) +def test_collect_remote_invalid_container_without_route_is_invalid( + monkeypatch: pytest.MonkeyPatch, + returncode: int, +) -> None: + result = collect_completed_probe( + monkeypatch, + label="vps", + returncode=returncode, + route=None, + stderr="docker: Error: No such container: teleo-pg password=hunter2 root@192.0.2.10", + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "remote_precondition_failed_before_route" + assert result["exit_code"] == returncode + assert result["route"] is None + assert "root@192.0.2.10" not in result["error"] + assert "hunter2" not in result["error"] + + +def test_collect_valid_route_with_psql_connection_failure_is_unreachable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + route = route_observation("gcp") + result = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=2, + route=route, + stderr="psql: connection unavailable token=private-token teleo-gcp-staging", + ) + + assert result["status"] == "unreachable" + assert result["failure_class"] == "database_connection_unavailable" + assert result["exit_code"] == 2 + assert result["route"] == route + assert "private-token" not in result["error"] + assert "teleo-gcp-staging" not in result["error"] + + +def test_collect_valid_route_with_manifest_script_exit_3_is_invalid( + monkeypatch: pytest.MonkeyPatch, +) -> None: + route = route_observation("gcp") + result = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=3, + route=route, + stderr="psql: manifest script failed secret=private-secret teleo-gcp-staging", + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "manifest_or_probe_script_failed" + assert result["exit_code"] == 3 + assert result["route"] == route + assert "private-secret" not in result["error"] + assert "teleo-gcp-staging" not in result["error"] + + +def test_collect_malformed_route_output_is_invalid_even_on_ssh_exit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + expected = endpoint_target("gcp") + stdout = (detector.ROUTE_PREFIX + "{not-json}\n").encode() + monkeypatch.setattr( + detector.subprocess, + "run", + lambda *args, **kwargs: detector.subprocess.CompletedProcess(args[0], 255, stdout, b"transport closed"), + ) + + result = detector.collect_endpoint( + label="gcp", + command=["ssh", expected["transport"], "--", "true"], + manifest_sql=b"select 1", + timeout=30, + expected_target=expected, + ) + + assert result["status"] == "invalid" + assert result["failure_class"] == "malformed_probe_output" + assert result["exit_code"] == 255 + + def test_probe_timeout_preserves_exact_expected_target(monkeypatch: pytest.MonkeyPatch) -> None: expected = endpoint_target("gcp") expected["database"] = "teleo_canonical" @@ -489,6 +640,46 @@ def test_live_runs_both_independent_collectors_and_returns_incomplete(tmp_path: assert exit_code == 2 +@pytest.mark.parametrize( + ("returncode", "include_route", "failure_class", "expected_status", "expected_exit"), + [ + (255, False, "ssh_transport_failed", "incomplete", 2), + (1, False, "remote_precondition_failed_before_route", "invalid", 3), + (3, False, "remote_precondition_failed_before_route", "invalid", 3), + (2, True, "database_connection_unavailable", "incomplete", 2), + (3, True, "manifest_or_probe_script_failed", "invalid", 3), + ], +) +def test_run_live_maps_collected_nonzero_partition( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + returncode: int, + include_route: bool, + failure_class: str, + expected_status: str, + expected_exit: int, +) -> None: + gcp = collect_completed_probe( + monkeypatch, + label="gcp", + returncode=returncode, + route=route_observation("gcp") if include_route else None, + stderr="bounded probe failure password=hunter2 teleo-gcp-staging", + ) + manifest_bytes = b"reviewed manifest" + args = live_args(tmp_path, manifest_bytes) + monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest()) + + def fake_collector(**kwargs): + return observation("vps") if kwargs["label"] == "vps" else gcp + + receipt, exit_code = detector.run_live(args, collector=fake_collector) + + assert receipt["status"] == expected_status + assert receipt["observations"]["gcp"]["failure_class"] == failure_class + assert exit_code == expected_exit + + def test_live_invalid_collector_payload_returns_exit_3( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,