From 52529403a6f4cebd2ae507dc9f4c3dfbcc91095d Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 9 Jul 2026 20:15:11 +0800 Subject: [PATCH] test(table-catalog): record live conformance evidence (#4606) Co-authored-by: Henry Guo --- docs/architecture/s3-tables-support-matrix.md | 12 +- scripts/table-catalog/README.md | 29 +++ scripts/table-catalog/engine_compatibility.py | 199 ++++++++++++++++++ scripts/table-catalog/pyiceberg_smoke.py | 163 +++++++++++++- .../test_engine_compatibility.py | 86 ++++++++ scripts/table-catalog/test_pyiceberg_smoke.py | 60 ++++++ 6 files changed, 544 insertions(+), 5 deletions(-) diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 5be48bbea..5f112c748 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -52,7 +52,7 @@ catalog extension. | Area | Status | Current RustFS claim | |---|---|---| -| Live conformance evidence template | Generated harness | `engine_compatibility.py --print-live-conformance` records required run metadata, per-client result rows, and claim promotion rules before a manual/live result can expand compatibility wording. | +| Live conformance evidence | Manual/live harness | `engine_compatibility.py --print-live-evidence-schema` defines the required evidence schema and claim promotion boundaries. `pyiceberg_smoke.py --live-evidence-output` writes a validated PyIceberg evidence record after a successful live smoke run. | | Production operations guide | Generated harness | `engine_compatibility.py --print-operations-guide` records command, evidence, pass criteria, and fail-closed signals for live conformance, durable backing cutover, maintenance, recovery, permissions, credential vending, and unsupported-claim governance. | | Vendor compatibility gap audit | Generated harness | `engine_compatibility.py --print-vendor-audit` records provider source URLs, catalog path and warehouse shapes, signing/auth models, error/permission/maintenance validation categories, and not-claimed boundaries for AWS S3 Tables, MinIO AIStor Tables, Cloudflare R2 Data Catalog, and Alibaba OSS Tables. | | Client claim promotion | Documented, not automated | PyIceberg remains the automated claim. Spark can be promoted only with recorded manual/live evidence; Trino and DuckDB read probes do not promote write compatibility; Snowflake and vendor profiles remain reference-only without repeatable live evidence. | @@ -210,6 +210,16 @@ python3 scripts/table-catalog/engine_compatibility.py \ --metadata-location s3://rustfs-s3table-smoke/tables/table-id/metadata/v1.metadata.json \ --print-live-conformance \ --cleanup +python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-schema +python3 scripts/table-catalog/pyiceberg_smoke.py \ + --endpoint http://127.0.0.1:9000 \ + --bucket rustfs-s3table-smoke \ + --replace \ + --cleanup \ + --rustfs-build rustfs-v1.0.0-beta.8 \ + --git-sha "$(git rev-parse HEAD)" \ + --catalog-backing durable-strong \ + --live-evidence-output /tmp/rustfs-pyiceberg-live-evidence.json python3 scripts/table-catalog/engine_compatibility.py \ --warehouse rustfs-s3table-smoke \ --namespace smoke \ diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 425090a13..0756c2039 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -27,6 +27,34 @@ python3 scripts/table-catalog/pyiceberg_smoke.py \ --cleanup ``` +To persist the successful PyIceberg run as live conformance evidence, provide +an output path and the operator-recorded deployment metadata: + +```bash +python3 scripts/table-catalog/pyiceberg_smoke.py \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-s3table-smoke \ + --replace \ + --cleanup \ + --rustfs-build rustfs-v1.0.0-beta.8 \ + --git-sha "$(git rev-parse HEAD)" \ + --catalog-backing durable-strong \ + --live-evidence-output /tmp/rustfs-pyiceberg-live-evidence.json +``` + +The evidence file uses the schema printed by: + +```bash +python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-schema +``` + +The smoke helper validates the record before writing it. Missing deployment +metadata, mismatched expected and observed status, invalid row counts, or +claim promotion beyond the client boundary are treated as evidence failures +instead of support-matrix proof. + The smoke test covers: - create or reuse the S3 bucket @@ -131,6 +159,7 @@ python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-failure-cove python3 scripts/table-catalog/pyiceberg_smoke.py --print-vendor-profiles python3 scripts/table-catalog/pyiceberg_smoke.py --print-unsupported-inventory python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-readiness +python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-schema ``` Use these outputs when updating release notes, PR descriptions, or follow-up diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py index 33662c539..b048653c3 100644 --- a/scripts/table-catalog/engine_compatibility.py +++ b/scripts/table-catalog/engine_compatibility.py @@ -22,6 +22,48 @@ DEFAULT_SNOWFLAKE_CLIENT_VERSION = "operator-recorded" DEFAULT_DATABEND_VERSION = "operator-recorded" DEFAULT_TRINO_SERVER = "http://127.0.0.1:8080" DEFAULT_DATABEND_DSN = "databend://root:@127.0.0.1:8000/default" +LIVE_EVIDENCE_SCHEMA = "rustfs-s3tables-live-conformance-evidence" +LIVE_EVIDENCE_SCHEMA_VERSION = 1 +LIVE_EVIDENCE_REQUIRED_FIELDS = [ + "schema", + "schema_version", + "client_name", + "client_version", + "scenario", + "rustfs_build", + "git_sha", + "catalog_backing", + "endpoint", + "warehouse", + "rest_path", + "namespace", + "table", + "metadata_location", + "run_timestamp_utc", + "operator", + "expected_status", + "observed_status", + "row_count", + "cleanup_result", + "claim", + "command", +] +LIVE_EVIDENCE_ALLOWED_CLAIMS = OrderedDict( + [ + ("PyIceberg", ["automated-smoke"]), + ("Spark Iceberg REST catalog", ["manual-live-verified"]), + ("Trino Iceberg REST catalog", ["manual-live-read-verified"]), + ("DuckDB Iceberg", ["manual-live-read-verified"]), + ("Databend", ["manual-live-s3-stage-verified"]), + ("Snowflake Open Catalog / Iceberg integrations", ["reference-only"]), + ] +) +LIVE_EVIDENCE_ROW_COUNT_CLIENTS = { + "PyIceberg", + "Spark Iceberg REST catalog", + "Trino Iceberg REST catalog", + "DuckDB Iceberg", +} VENDOR_SPARK_PROFILES: dict[str, dict[str, str]] = { "rustfs": { @@ -634,6 +676,159 @@ def live_conformance_evidence( ) +def live_conformance_evidence_schema() -> OrderedDict[str, Any]: + return OrderedDict( + [ + ("schema", LIVE_EVIDENCE_SCHEMA), + ("schema_version", LIVE_EVIDENCE_SCHEMA_VERSION), + ("required_fields", LIVE_EVIDENCE_REQUIRED_FIELDS.copy()), + ("claim_promotion", OrderedDict((client, claims.copy()) for client, claims in LIVE_EVIDENCE_ALLOWED_CLAIMS.items())), + ( + "fail_closed_rules", + [ + "missing required field rejects the evidence record", + "rustfs_build, git_sha, catalog_backing, and client_version must be recorded before claim promotion", + "observed_status must match expected_status before any claim can be promoted", + "expected_status must be pass before any claim can be promoted", + "read-only probes cannot promote write compatibility", + "PyIceberg, Spark, Trino, and DuckDB pass evidence must record row_count=2", + "reference-only providers remain reference-only without a repeatable live run", + ], + ), + ] + ) + + +def live_conformance_evidence_record( + *, + client_name: str, + client_version: str, + scenario: str, + rustfs_build: str, + git_sha: str, + catalog_backing: str, + endpoint: str, + warehouse: str, + rest_path: str, + namespace: str, + table: str, + metadata_location: str, + run_timestamp_utc: str, + operator: str, + expected_status: str, + observed_status: str, + row_count: int, + cleanup_result: str, + claim: str, + command: str, +) -> OrderedDict[str, Any]: + return OrderedDict( + [ + ("schema", LIVE_EVIDENCE_SCHEMA), + ("schema_version", LIVE_EVIDENCE_SCHEMA_VERSION), + ("client_name", client_name), + ("client_version", client_version), + ("scenario", scenario), + ("rustfs_build", rustfs_build), + ("git_sha", git_sha), + ("catalog_backing", catalog_backing), + ("endpoint", normalized_endpoint(endpoint)), + ("warehouse", warehouse), + ("rest_path", normalized_rest_path(rest_path)), + ("namespace", namespace), + ("table", table), + ("metadata_location", metadata_location), + ("run_timestamp_utc", run_timestamp_utc), + ("operator", operator), + ("expected_status", expected_status), + ("observed_status", observed_status), + ("row_count", row_count), + ("cleanup_result", cleanup_result), + ("claim", claim), + ("command", command), + ] + ) + + +def require_non_empty_string(record: dict[str, Any], field: str) -> str: + value = record.get(field) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"live conformance evidence field {field} must be a non-empty string") + return value + + +def require_operator_recorded_value(record: dict[str, Any], field: str) -> str: + value = require_non_empty_string(record, field) + if value == "operator-recorded": + raise ValueError(f"live conformance evidence field {field} must be recorded for claim promotion") + return value + + +def validate_live_conformance_evidence(record: dict[str, Any]) -> OrderedDict[str, Any]: + for field in LIVE_EVIDENCE_REQUIRED_FIELDS: + if field not in record: + raise ValueError(f"live conformance evidence is missing required field {field}") + if record["schema"] != LIVE_EVIDENCE_SCHEMA: + raise ValueError("live conformance evidence schema does not match RustFS S3 Tables evidence schema") + if record["schema_version"] != LIVE_EVIDENCE_SCHEMA_VERSION: + raise ValueError("live conformance evidence schema_version is not supported") + + client_name = require_non_empty_string(record, "client_name") + claim = require_non_empty_string(record, "claim") + allowed_claims = LIVE_EVIDENCE_ALLOWED_CLAIMS.get(client_name) + if allowed_claims is None: + raise ValueError(f"live conformance evidence client_name is not recognized: {client_name}") + if claim not in allowed_claims: + raise ValueError(f"live conformance evidence claim {claim} is not allowed for {client_name}") + + expected_status = require_non_empty_string(record, "expected_status") + observed_status = require_non_empty_string(record, "observed_status") + if expected_status != "pass": + raise ValueError("live conformance evidence expected_status must be pass for claim promotion") + if observed_status != expected_status: + raise ValueError("live conformance evidence observed_status does not match expected_status") + + for field in [ + "client_version", + "rustfs_build", + "git_sha", + "catalog_backing", + ]: + require_operator_recorded_value(record, field) + + for field in [ + "scenario", + "endpoint", + "warehouse", + "rest_path", + "namespace", + "table", + "metadata_location", + "run_timestamp_utc", + "operator", + "cleanup_result", + "command", + ]: + require_non_empty_string(record, field) + + row_count = record["row_count"] + if not isinstance(row_count, int) or row_count < 0: + raise ValueError("live conformance evidence row_count must be a non-negative integer") + if expected_status == "pass" and client_name in LIVE_EVIDENCE_ROW_COUNT_CLIENTS and row_count != 2: + raise ValueError("live conformance evidence row_count must be 2 for successful table smoke/read probes") + + return OrderedDict( + [ + ("status", "accepted"), + ("client_name", client_name), + ("scenario", record["scenario"]), + ("claim", claim), + ("metadata_location", record["metadata_location"]), + ("row_count", row_count), + ] + ) + + def production_operations_guide( *, endpoint: str, @@ -1427,6 +1622,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument("--print-engine-matrix", action="store_true") parser.add_argument("--print-vendor-audit", action="store_true") parser.add_argument("--print-live-conformance", action="store_true") + parser.add_argument("--print-live-evidence-schema", action="store_true") parser.add_argument("--print-operations-guide", action="store_true") parser.add_argument("--print-spark-config", action="store_true") parser.add_argument("--print-spark-sql", action="store_true") @@ -1508,6 +1704,9 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None: output, ) printed = True + if args.print_live_evidence_schema: + print_json({"live_conformance_evidence_schema": live_conformance_evidence_schema()}, output) + printed = True if args.print_operations_guide: print_json( { diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index beb1be3a9..778b2c9c7 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -5,8 +5,10 @@ from __future__ import annotations import argparse import hashlib +import importlib.metadata import json import os +import shlex import ssl import sys import time @@ -14,6 +16,8 @@ import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path from typing import Any import engine_compatibility @@ -27,6 +31,11 @@ REQUIRED_STORAGE_CREDENTIAL_KEYS = ( "s3.secret-access-key", "s3.session-token", ) +SENSITIVE_COMMAND_FLAGS = { + "--access-key", + "--secret-key", + "--session-token", +} TABLE_MAINTENANCE_CONFIG_VERSION = 1 IDENTIFIER_SEGMENT_MAX_LEN = 64 @@ -264,6 +273,14 @@ class StorageCredential: config: dict[str, str] +@dataclass(frozen=True) +class SmokeResult: + metadata_location: str + row_count: int + cleanup_result: str + table_warehouse_location: str + + class RestRequestError(RuntimeError): def __init__(self, method: str, path: str, status_code: int, response_body: str) -> None: super().__init__(f"{method} {path} failed with HTTP {status_code}: {response_body}") @@ -404,6 +421,16 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "20"))) parser.add_argument("--cleanup", action="store_true", help="Drop the smoke table and namespace before exiting.") parser.add_argument("--replace", action="store_true", help="Drop an existing table with the same identifier first.") + parser.add_argument( + "--live-evidence-output", + help="Write a validated live conformance evidence JSON record after a successful PyIceberg smoke run.", + ) + parser.add_argument("--rustfs-build", default=os.getenv("RUSTFS_BUILD", "operator-recorded")) + parser.add_argument("--git-sha", default=os.getenv("RUSTFS_GIT_SHA", "operator-recorded")) + parser.add_argument("--catalog-backing", default=os.getenv("RUSTFS_TABLE_CATALOG_BACKING", "operator-recorded")) + parser.add_argument("--operator", default=os.getenv("USER", "operator-recorded")) + parser.add_argument("--run-timestamp-utc", default=env_or_none("RUSTFS_TABLE_LIVE_RUN_TIMESTAMP_UTC")) + parser.add_argument("--client-version", default=env_or_none("RUSTFS_TABLE_CLIENT_VERSION")) parser.add_argument( "--skip-catalog-api-probes", action="store_true", @@ -701,6 +728,21 @@ def table_warehouse_location(table: Any) -> str: raise RuntimeError("created table did not expose a warehouse location") +def table_metadata_location(table: Any) -> str | None: + for source in (table, getattr(table, "metadata", None)): + if callable(source): + source = source() + if source is None: + continue + if isinstance(source, dict): + location = string_value(source.get("metadata-location") or source.get("metadata_location")) + else: + location = string_value(getattr(source, "metadata_location", None) or getattr(source, "metadataLocation", None)) + if location is not None: + return location + return None + + def safe_probe_segment(namespace: str, table: str) -> str: source = f"{namespace}-{table}" return "".join(char if char.isalnum() or char in {"-", "_", "."} else "-" for char in source) @@ -864,6 +906,98 @@ def printed_metadata(args: argparse.Namespace) -> bool: return printed +def current_utc_timestamp() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def pyiceberg_client_version(args: argparse.Namespace) -> str: + if args.client_version: + return args.client_version + try: + return importlib.metadata.version("pyiceberg") + except importlib.metadata.PackageNotFoundError: + return "operator-recorded" + + +def redacted_command(argv: list[str]) -> str: + redacted: list[str] = [] + redact_next = False + for arg in argv: + if redact_next: + redacted.append("") + redact_next = False + continue + flag, separator, value = arg.partition("=") + if flag in SENSITIVE_COMMAND_FLAGS: + if separator: + redacted.append(f"{flag}=") + else: + redacted.append(flag) + redact_next = True + continue + redacted.append(arg) + return shlex.join(redacted) + + +def pyiceberg_live_evidence_record( + args: argparse.Namespace, + result: SmokeResult, + *, + client_version: str, + rustfs_build: str, + git_sha: str, + catalog_backing: str, + run_timestamp_utc: str, + operator: str, + command: str, +) -> dict[str, Any]: + return engine_compatibility.live_conformance_evidence_record( + client_name="PyIceberg", + client_version=client_version, + scenario="create-append-reload-scan-direct-rest-probes", + rustfs_build=rustfs_build, + git_sha=git_sha, + catalog_backing=catalog_backing, + endpoint=args.endpoint, + warehouse=profile_warehouse(args), + rest_path=args.rest_path, + namespace=args.namespace, + table=args.table, + metadata_location=result.metadata_location, + run_timestamp_utc=run_timestamp_utc, + operator=operator, + expected_status="pass", + observed_status="pass", + row_count=result.row_count, + cleanup_result=result.cleanup_result, + claim="automated-smoke", + command=command, + ) + + +def write_live_evidence(args: argparse.Namespace, result: SmokeResult) -> None: + if not args.live_evidence_output: + return + record = pyiceberg_live_evidence_record( + args, + result, + client_version=pyiceberg_client_version(args), + rustfs_build=args.rustfs_build, + git_sha=args.git_sha, + catalog_backing=args.catalog_backing, + run_timestamp_utc=args.run_timestamp_utc or current_utc_timestamp(), + operator=args.operator, + command=redacted_command(sys.argv), + ) + validation = engine_compatibility.validate_live_conformance_evidence(record) + document = { + "live_conformance_evidence": record, + "validation": validation, + } + output_path = Path(args.live_evidence_output) + output_path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + def install_rustfs_rest_sigv4_adapter(catalog: Any, args: argparse.Namespace, deps: RuntimeDeps) -> None: from urllib import parse @@ -1120,7 +1254,7 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: raise RuntimeError("maintenance worker endpoint did not return audit events") -def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> None: +def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> dict[str, Any]: table_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args)) snapshot_id = current_snapshot_id_from_table_response(table_response) run_metadata_location_probe(args, deps, table_response) @@ -1130,9 +1264,10 @@ def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> None: diagnostics = signed_rest_request(args, deps, "GET", table_endpoint_path(args, "/catalog/diagnostics")) if not isinstance(diagnostics, dict) or not diagnostics: raise RuntimeError("catalog diagnostics endpoint returned an empty response") + return table_response -def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: +def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> SmokeResult: endpoint = normalized_endpoint(args.endpoint) ensure_local_proxy_bypass(endpoint) ensure_aws_env(args.access_key, args.secret_key, args.region) @@ -1197,18 +1332,37 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: scanned = loaded.scan().to_arrow() if scanned.num_rows != 2: raise RuntimeError(f"expected 2 rows after append, got {scanned.num_rows}") + loaded_table_location = table_warehouse_location(loaded) + loaded_metadata_location = table_metadata_location(loaded) + table_response = None if args.skip_catalog_api_probes: print("[10/10] skipping direct REST catalog API probes") + if args.live_evidence_output and loaded_metadata_location is None: + table_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args)) else: print("[10/10] probing direct REST catalog APIs") - run_catalog_api_probes(args, deps) + table_response = run_catalog_api_probes(args, deps) + + metadata_location = loaded_metadata_location + if isinstance(table_response, dict): + response_metadata_location = table_response.get("metadata-location") + if isinstance(response_metadata_location, str) and response_metadata_location: + metadata_location = response_metadata_location + if args.live_evidence_output and metadata_location is None: + raise RuntimeError("live evidence output requires a metadata location from PyIceberg or loadTable") if args.cleanup: print(f"cleanup: dropping table and namespace {'.'.join(identifier)}") cleanup_catalog(catalog, identifier) print(f"PASS: PyIceberg smoke test completed for {'.'.join(identifier)}") + return SmokeResult( + metadata_location=metadata_location or "operator-recorded", + row_count=scanned.num_rows, + cleanup_result="dropped-table-and-namespace" if args.cleanup else "not-requested", + table_warehouse_location=loaded_table_location, + ) def main() -> int: @@ -1217,7 +1371,8 @@ def main() -> int: if printed_metadata(args): return 0 deps = load_runtime_deps() - run_smoke(args, deps) + result = run_smoke(args, deps) + write_live_evidence(args, result) return 0 except Exception as error: print(f"ERROR: {error}", file=sys.stderr) diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py index 1e02c15c8..2a8d665e6 100644 --- a/scripts/table-catalog/test_engine_compatibility.py +++ b/scripts/table-catalog/test_engine_compatibility.py @@ -410,6 +410,92 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertIn("manual-live", " ".join(evidence["promotion_rules"])) self.assertIn("not-claimed", " ".join(evidence["promotion_rules"])) + def test_live_conformance_evidence_record_validates_claim_promotion(self) -> None: + record = engine_compatibility.live_conformance_evidence_record( + client_name="PyIceberg", + client_version="0.10.0", + scenario="create-append-reload-scan-direct-rest-probes", + rustfs_build="rustfs-test", + git_sha="abc123", + catalog_backing="durable-strong", + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + rest_path="/iceberg", + namespace="smoke", + table="events", + metadata_location="s3://rustfs-s3table-smoke/tables/table-id/metadata/v2.metadata.json", + run_timestamp_utc="2026-07-09T00:00:00Z", + operator="ci", + expected_status="pass", + observed_status="pass", + row_count=2, + cleanup_result="dropped-table-and-namespace", + claim="automated-smoke", + command="python3 scripts/table-catalog/pyiceberg_smoke.py --cleanup", + ) + + validation = engine_compatibility.validate_live_conformance_evidence(record) + + self.assertEqual(record["schema"], "rustfs-s3tables-live-conformance-evidence") + self.assertEqual(record["schema_version"], 1) + self.assertEqual(validation["status"], "accepted") + self.assertEqual(validation["claim"], "automated-smoke") + self.assertEqual(validation["client_name"], "PyIceberg") + + def test_live_conformance_evidence_rejects_missing_or_mismatched_results(self) -> None: + record = engine_compatibility.live_conformance_evidence_record( + client_name="Spark Iceberg REST catalog", + client_version="3.5.4/iceberg-1.7.1", + scenario="create-namespace-create-table-append-refresh-count-cleanup", + rustfs_build="rustfs-test", + git_sha="abc123", + catalog_backing="durable-strong", + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + rest_path="/iceberg", + namespace="smoke", + table="events", + metadata_location="s3://rustfs-s3table-smoke/tables/table-id/metadata/v2.metadata.json", + run_timestamp_utc="2026-07-09T00:00:00Z", + operator="ci", + expected_status="pass", + observed_status="fail", + row_count=0, + cleanup_result="not-run", + claim="manual-live-verified", + command="spark-sql -f /tmp/rustfs-s3tables-smoke-events-spark.sql", + ) + + with self.assertRaisesRegex(ValueError, "observed_status"): + engine_compatibility.validate_live_conformance_evidence(record) + + passing = dict(record, observed_status="pass", row_count=2) + del passing["metadata_location"] + with self.assertRaisesRegex(ValueError, "metadata_location"): + engine_compatibility.validate_live_conformance_evidence(passing) + + overclaim = dict(record, observed_status="pass", row_count=2, client_name="Trino Iceberg REST catalog", claim="manual-live-write-verified") + with self.assertRaisesRegex(ValueError, "claim"): + engine_compatibility.validate_live_conformance_evidence(overclaim) + + unrecorded = dict(record, expected_status="pass", observed_status="pass", row_count=2, rustfs_build="operator-recorded") + with self.assertRaisesRegex(ValueError, "rustfs_build"): + engine_compatibility.validate_live_conformance_evidence(unrecorded) + + failed_claim = dict(record, expected_status="fail", observed_status="fail", row_count=0) + with self.assertRaisesRegex(ValueError, "expected_status"): + engine_compatibility.validate_live_conformance_evidence(failed_claim) + + def test_cli_prints_live_conformance_evidence_schema(self) -> None: + payload = engine_compatibility.cli_json(["--print-live-evidence-schema"]) + document = json.loads(payload) + + schema = document["live_conformance_evidence_schema"] + self.assertEqual(schema["schema"], "rustfs-s3tables-live-conformance-evidence") + self.assertIn("metadata_location", schema["required_fields"]) + self.assertIn("claim", schema["required_fields"]) + self.assertEqual(schema["claim_promotion"]["Trino Iceberg REST catalog"], ["manual-live-read-verified"]) + def test_production_operations_guide_covers_release_boundaries(self) -> None: guide = engine_compatibility.production_operations_guide( endpoint="http://127.0.0.1:9000", diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 2631be910..5153a6a14 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -914,6 +914,66 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertEqual(selected["catalog_uri"], "http://127.0.0.1:9000/iceberg") self.assertEqual(selected["warehouse"], "rustfs-s3table-smoke") + def test_pyiceberg_live_evidence_record_uses_runtime_smoke_result(self) -> None: + args = self.parse_with_args([ + "--endpoint", + "http://127.0.0.1:9000", + "--bucket", + "lake", + "--namespace", + "smoke", + "--table", + "events", + "--live-evidence-output", + "/tmp/rustfs-live-evidence.json", + ]) + result = pyiceberg_smoke.SmokeResult( + metadata_location="s3://lake/tables/table-id/metadata/v2.metadata.json", + row_count=2, + cleanup_result="not-requested", + table_warehouse_location="s3://lake/tables/table-id", + ) + + record = pyiceberg_smoke.pyiceberg_live_evidence_record( + args, + result, + client_version="0.10.0", + rustfs_build="rustfs-test", + git_sha="abc123", + catalog_backing="durable-strong", + run_timestamp_utc="2026-07-09T00:00:00Z", + operator="ci", + command="python3 scripts/table-catalog/pyiceberg_smoke.py --live-evidence-output /tmp/rustfs-live-evidence.json", + ) + + self.assertEqual(record["client_name"], "PyIceberg") + self.assertEqual(record["client_version"], "0.10.0") + self.assertEqual(record["warehouse"], "lake") + self.assertEqual(record["metadata_location"], "s3://lake/tables/table-id/metadata/v2.metadata.json") + self.assertEqual(record["row_count"], 2) + self.assertEqual(record["claim"], "automated-smoke") + + def test_live_evidence_command_redacts_cli_secrets(self) -> None: + command = pyiceberg_smoke.redacted_command( + [ + "pyiceberg_smoke.py", + "--endpoint", + "http://127.0.0.1:9000", + "--access-key", + "root-access", + "--secret-key=root-secret", + "--bucket", + "lake", + ] + ) + + self.assertIn("--access-key ''", command) + self.assertIn("--secret-key", command) + self.assertIn("", command) + self.assertNotIn("root-access", command) + self.assertNotIn("root-secret", command) + self.assertIn("--bucket lake", command) + def test_print_vendor_profiles_renders_selected_aws_profile(self) -> None: args = self.parse_with_args( [