diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 1abc61e74..b6a18c4ed 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -48,6 +48,14 @@ catalog extension. | Databend | Manual/live harness | RustFS can generate an S3 stage read probe for table data files. RustFS does not claim Databend Iceberg REST Catalog integration yet. | | Snowflake Open Catalog / Iceberg integrations | Generated harness | RustFS can generate an operator-adapted external volume/catalog SQL template. Live RustFS interoperability is not claimed. | +## Live Evidence And Operations Matrix + +| 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. | +| 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. | +| 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. | + ## Catalog API Matrix | Area | Status | Covered behavior | @@ -199,6 +207,11 @@ 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 \ + --warehouse rustfs-s3table-smoke \ + --namespace smoke \ + --table events \ + --print-operations-guide python3 scripts/table-catalog/failure_coverage.py \ --warehouse rustfs-s3table-smoke \ --namespace smoke \ @@ -222,7 +235,8 @@ Acceptable wording: > with PyIceberg smoke coverage, table-aware S3 data-plane policy checks, > controlled maintenance, catalog recovery diagnostics, manual conformance > input for Spark, Trino, DuckDB, Databend, and Snowflake, production-failure -> probe harnesses, and disaster-recovery rehearsal probes. +> probe harnesses, disaster-recovery rehearsal probes, and a machine-readable +> production operations evidence guide. Do not claim: diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index c55430961..784a73a66 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -191,6 +191,7 @@ python3 scripts/table-catalog/engine_compatibility.py \ --print-spark-config python3 scripts/table-catalog/engine_compatibility.py --print-spark-sql --cleanup python3 scripts/table-catalog/engine_compatibility.py --print-live-conformance --cleanup +python3 scripts/table-catalog/engine_compatibility.py --print-operations-guide ``` The production failure helper records the negative coverage required before @@ -320,6 +321,27 @@ manual-review diagnostics, stale rollback/import conflicts, and data-plane policy failures as fail-closed results that require operator investigation before cutover or release claims. +## Production Operations Guide + +The engine helper can print a machine-readable operations guide that ties client +conformance, durable backing cutover, maintenance, recovery, permissions, and +unsupported-claim governance to the exact evidence operators must record: + +```bash +python3 scripts/table-catalog/engine_compatibility.py \ + --endpoint http://127.0.0.1:9000 \ + --warehouse rustfs-s3table-smoke \ + --namespace smoke \ + --table events \ + --print-operations-guide +``` + +Use this output as the release checklist when expanding compatibility language. +Each section records commands, required evidence, pass criteria, and fail-closed +signals. A client or vendor claim should only be promoted when the corresponding +live evidence records the RustFS build, catalog backing mode, client version, +expected status, observed status, and metadata location. + ## Vendor Profile References | Profile | Catalog shape | Warehouse shape | Signing name | RustFS claim | @@ -410,6 +432,9 @@ The output includes: - generated Spark SQL - a `spark-sql` command using the generated properties - expected `row_count=2` before optional cleanup +- an evidence template for recording RustFS build, catalog backing mode, client + version, expected status, observed status, metadata location, and claim + promotion boundary Generate the configuration properties: diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py index 5ef5bdfc6..2f0016448 100644 --- a/scripts/table-catalog/engine_compatibility.py +++ b/scripts/table-catalog/engine_compatibility.py @@ -7,6 +7,7 @@ import argparse import json import re import shlex +import urllib.parse from collections import OrderedDict from io import StringIO from typing import Any @@ -488,6 +489,463 @@ def databend_command(*, dsn: str) -> str: return shell_join(["databend-sql", "--dsn", dsn, "-f", "/tmp/rustfs-s3tables-databend-stage.sql"]) +def rest_api_prefix(rest_path: str) -> str: + prefix = normalized_rest_path(rest_path) + if not prefix.endswith("/v1"): + prefix = f"{prefix}/v1" + return prefix + + +def rest_path_segment(value: str) -> str: + return urllib.parse.quote(value, safe="") + + +def warehouse_catalog_path(*, warehouse: str, rest_path: str, suffix: str = "") -> str: + return f"{rest_api_prefix(rest_path)}/{rest_path_segment(warehouse)}{suffix}" + + +def table_catalog_path(*, warehouse: str, namespace: str, table: str, rest_path: str, suffix: str = "") -> str: + return ( + f"{rest_api_prefix(rest_path)}/{rest_path_segment(warehouse)}" + f"/namespaces/{rest_path_segment(namespace)}/tables/{rest_path_segment(table)}{suffix}" + ) + + +def live_conformance_evidence( + *, + warehouse: str, + namespace: str, + table: str, + rest_path: str, + metadata_location: str, + cleanup: bool, +) -> OrderedDict[str, Any]: + return OrderedDict( + [ + ("result", "operator-recorded"), + ( + "required_run_metadata", + [ + "rustfs_build", + "git_sha", + "catalog_backing", + "endpoint", + "warehouse", + "rest_path", + "namespace", + "table", + "metadata_location", + "client_name", + "client_version", + "run_timestamp_utc", + "operator", + "expected_status", + "observed_status", + "row_count", + "cleanup_result", + ], + ), + ( + "result_table_template", + [ + OrderedDict( + [ + ("client", "PyIceberg"), + ("scenario", "create-append-reload-scan-direct-rest-probes"), + ("expected_status", "pass"), + ("expected_row_count", 2), + ("claim_after_pass", "automated-smoke"), + ] + ), + OrderedDict( + [ + ("client", "Spark Iceberg REST catalog"), + ("scenario", "create-namespace-create-table-append-refresh-count-cleanup"), + ("expected_status", "pass"), + ("expected_row_count", 2), + ("claim_after_pass", "manual-live-verified"), + ] + ), + OrderedDict( + [ + ("client", "Trino Iceberg REST catalog"), + ("scenario", "read-count-existing-table"), + ("expected_status", "pass"), + ("expected_row_count", 2), + ("claim_after_pass", "manual-live-read-verified"), + ("write_claim_after_pass", "not-claimed"), + ] + ), + OrderedDict( + [ + ("client", "DuckDB Iceberg"), + ("scenario", "iceberg-scan-current-metadata-location"), + ("expected_status", "pass"), + ("expected_row_count", 2), + ("claim_after_pass", "manual-live-read-verified"), + ("write_claim_after_pass", "not-claimed"), + ] + ), + OrderedDict( + [ + ("client", "Databend"), + ("scenario", "s3-stage-read-table-data-files"), + ("expected_status", "pass"), + ("claim_after_pass", "manual-live-s3-stage-verified"), + ("iceberg_rest_catalog_claim_after_pass", "not-claimed"), + ] + ), + OrderedDict( + [ + ("client", "Snowflake Open Catalog / Iceberg integrations"), + ("scenario", "operator-adapted-external-volume-reference"), + ("expected_status", "operator-recorded"), + ("claim_after_pass", "reference-only"), + ("live_rustfs_interoperability_after_pass", "not-claimed"), + ] + ), + ], + ), + ( + "promotion_rules", + [ + "Keep PyIceberg as the only automated claim unless the run is executed by CI or a repeatable operator job.", + "Promote Spark only to manual-live-verified when the exact RustFS build, Spark version, Iceberg version, SQL output, and row_count are recorded.", + "Do not promote Trino or DuckDB write compatibility from read probes; write compatibility remains not-claimed.", + "Do not promote Snowflake or vendor catalog interoperability from a generated template without a repeatable live run.", + "Treat manual-live failures as compatibility findings and keep the previous public claim boundary.", + ], + ), + ( + "target", + OrderedDict( + [ + ("warehouse", warehouse), + ("namespace", namespace), + ("table", table), + ("rest_path", rest_path), + ("metadata_location", metadata_location), + ("cleanup", cleanup), + ] + ), + ), + ] + ) + + +def production_operations_guide( + *, + endpoint: str, + warehouse: str, + namespace: str, + table: str, + rest_path: str, +) -> OrderedDict[str, Any]: + endpoint = normalized_endpoint(endpoint) + table_base = table_catalog_path(warehouse=warehouse, namespace=namespace, table=table, rest_path=rest_path) + pyiceberg_smoke_command = shell_join( + [ + "python3", + "scripts/table-catalog/pyiceberg_smoke.py", + "--profile", + "rustfs", + "--endpoint", + endpoint, + "--bucket", + warehouse, + "--namespace", + namespace, + "--table", + table, + "--rest-path", + rest_path, + "--replace", + "--cleanup", + ] + ) + live_conformance_command = shell_join( + [ + "python3", + "scripts/table-catalog/engine_compatibility.py", + "--endpoint", + endpoint, + "--warehouse", + warehouse, + "--namespace", + namespace, + "--table", + table, + "--rest-path", + rest_path, + "--print-live-conformance", + "--cleanup", + ] + ) + return OrderedDict( + [ + ("claim_boundary", "Iceberg REST Catalog S3 Tables implementation"), + ( + "target", + OrderedDict( + [ + ("endpoint", endpoint), + ("warehouse", warehouse), + ("namespace", namespace), + ("table", table), + ("rest_path", rest_path), + ] + ), + ), + ( + "sections", + [ + OrderedDict( + [ + ("name", "live-client-conformance"), + ("objective", "Record repeatable client evidence before expanding engine compatibility claims."), + ( + "commands", + [ + pyiceberg_smoke_command, + live_conformance_command, + ], + ), + ( + "required_evidence", + [ + "RustFS build and git SHA", + "catalog backing mode", + "client name and version", + "generated command or SQL", + "expected and observed status", + "row_count or response status", + "current metadata location", + ], + ), + ( + "pass_criteria", + [ + "PyIceberg append/reload/scan returns row_count=2", + "Spark manual/live run returns row_count=2 before cleanup", + "read-only engines do not claim write compatibility", + ], + ), + ( + "fail_closed_signals", + [ + "missing run metadata", + "manual-live run fails", + "client writes are attempted through a read-only probe", + ], + ), + ] + ), + OrderedDict( + [ + ("name", "catalog-backing-cutover"), + ("objective", "Verify durable backing readiness before selecting a non-object-backed catalog mode."), + ("commands", [f"GET {warehouse_catalog_path(warehouse=warehouse, rest_path=rest_path, suffix='/catalog/migration')}"]), + ( + "required_evidence", + [ + "migration dry-run response", + "commit recovery blockers", + "idempotency index readiness", + "warehouse prefix index readiness", + "rollback configuration", + ], + ), + ( + "pass_criteria", + [ + "blockers is empty", + "recommended actions are completed", + "object-backed catalog backup exists before cutover", + ], + ), + ( + "fail_closed_signals", + [ + "non-empty blockers", + "recoverable commit gaps require repair", + "warehouse prefix index is missing or stale", + ], + ), + ] + ), + OrderedDict( + [ + ("name", "maintenance-operations"), + ("objective", "Validate controlled maintenance without claiming a continuous in-process scheduler."), + ( + "commands", + [ + f"POST {table_base}/maintenance/metadata", + f"GET {table_base}/maintenance/jobs/{{job}}", + ], + ), + ( + "required_evidence", + [ + "dry-run report before delete or rewrite", + "current pointer re-read", + "audit-events for planning and worker transitions", + "quarantine status for failed jobs", + "safe-window configuration", + ], + ), + ( + "pass_criteria", + [ + "unreachable object deletion candidates pass safety-window checks", + "rewrite groups stay partition-local and sort-order-local", + "delete-file or row-level maintenance remains manual review", + ], + ), + ( + "fail_closed_signals", + [ + "stale maintenance plan", + "quarantine required", + "row-level delete file rewrite requested", + "missing referenced object", + ], + ), + ] + ), + OrderedDict( + [ + ("name", "recovery-and-disaster-rehearsal"), + ("objective", "Exercise operator recovery paths without moving the table pointer unexpectedly."), + ( + "commands", + [ + f"GET {table_base}/catalog/diagnostics", + f"POST {table_base}/catalog/recovery", + "python3 scripts/table-catalog/failure_coverage.py --print-disaster-recovery-rehearsal", + ], + ), + ( + "required_evidence", + [ + "catalog export baseline", + "diagnostics recovery status", + "safe repair result", + "rollback or import response when used", + "post-recovery loadTable result", + ], + ), + ( + "pass_criteria", + [ + "safe repair does not move the current pointer", + "rollback/import uses normal catalog validation", + "post-recovery loadTable returns the expected metadata location", + ], + ), + ( + "fail_closed_signals", + [ + "manual-review diagnostics", + "stale rollback or import conflict", + "post-recovery data-plane policy failure", + ], + ), + ] + ), + OrderedDict( + [ + ("name", "permissions-and-credentials"), + ("objective", "Verify table policy and credential vending cannot bypass table warehouse scope."), + ( + "commands", + [ + "python3 scripts/table-catalog/pyiceberg_smoke.py --profile rustfs-vended-credentials --replace --cleanup", + f"GET {table_base}/credentials", + ], + ), + ( + "required_evidence", + [ + "catalog principal permissions", + "vended credential expiration", + "warehouse prefix scope", + "inside-prefix S3 put/head/get/delete result", + "outside-prefix deny result", + ], + ), + ( + "pass_criteria", + [ + "default credentials response is empty unless vending is enabled", + "vended credentials work inside the exact table prefix", + "vended credentials are denied outside the exact table prefix", + ], + ), + ( + "fail_closed_signals", + [ + "long-lived storage secret returned by catalog", + "prefix mismatch", + "outside-prefix object access succeeds", + ], + ), + ] + ), + OrderedDict( + [ + ("name", "unsupported-claim-governance"), + ("objective", "Keep release language aligned with verified RustFS behavior."), + ( + "commands", + [ + "python3 scripts/table-catalog/pyiceberg_smoke.py --print-unsupported-inventory", + "python3 scripts/table-catalog/engine_compatibility.py --print-engine-matrix", + ], + ), + ( + "not_claimed", + [ + "full AWS S3 Tables control-plane API parity", + "full MinIO AIStor private extension parity", + "full Cloudflare R2 Data Catalog interoperability", + "active-active multi-region table writes", + "multi-table transactions", + "built-in SQL query execution", + "Delta Lake or Hudi table format support", + "end-to-end SQL row-level DML validation", + ], + ), + ( + "required_evidence", + [ + "support matrix status label", + "unsupported inventory entry", + "live evidence before any promotion", + ], + ), + ( + "pass_criteria", + [ + "public claim wording matches support matrix status", + "reference profiles stay reference-only without live evidence", + ], + ), + ( + "fail_closed_signals", + [ + "vendor parity wording without recorded live evidence", + "manual probe documented as automated support", + ], + ), + ] + ), + ], + ), + ] + ) + + def live_conformance_harness( *, endpoint: str, @@ -600,6 +1058,17 @@ def live_conformance_harness( ] ), ), + ( + "evidence", + live_conformance_evidence( + warehouse=warehouse, + namespace=namespace, + table=table, + rest_path=rest_path, + metadata_location=metadata_location, + cleanup=cleanup, + ), + ), ( "clients", [ @@ -731,6 +1200,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument("--cleanup", action="store_true") parser.add_argument("--print-engine-matrix", action="store_true") parser.add_argument("--print-live-conformance", 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") return parser.parse_args(argv) @@ -808,6 +1278,20 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None: output, ) printed = True + if args.print_operations_guide: + print_json( + { + "production_operations_guide": production_operations_guide( + endpoint=args.endpoint, + warehouse=args.warehouse, + namespace=args.namespace, + table=args.table, + rest_path=args.rest_path or "/iceberg", + ) + }, + output, + ) + printed = True if args.print_spark_sql: sql = spark_sql_smoke( catalog_name=args.catalog_name, diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py index 0285658e5..8e67f05f5 100644 --- a/scripts/table-catalog/test_engine_compatibility.py +++ b/scripts/table-catalog/test_engine_compatibility.py @@ -278,6 +278,115 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertIn("SELECT COUNT(*)", databend["sql"]) self.assertEqual(databend["iceberg_rest_catalog"], "not-claimed") + def test_live_conformance_harness_records_evidence_contract(self) -> None: + harness = engine_compatibility.live_conformance_harness( + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="us-east-1", + catalog_name="rustfs", + namespace="smoke", + table="events", + rest_path="/iceberg", + rest_signing_name="s3", + pyiceberg_version="0.10.0", + spark_version="3.5.4", + iceberg_version="1.7.1", + scala_version="2.12", + ) + + evidence = harness["evidence"] + self.assertEqual(evidence["result"], "operator-recorded") + self.assertIn("rustfs_build", evidence["required_run_metadata"]) + self.assertIn("catalog_backing", evidence["required_run_metadata"]) + self.assertIn("client_version", evidence["required_run_metadata"]) + self.assertIn("metadata_location", evidence["required_run_metadata"]) + self.assertIn("expected_status", evidence["required_run_metadata"]) + self.assertIn("observed_status", evidence["required_run_metadata"]) + + table_by_client = {row["client"]: row for row in evidence["result_table_template"]} + self.assertEqual(table_by_client["PyIceberg"]["claim_after_pass"], "automated-smoke") + self.assertEqual(table_by_client["Spark Iceberg REST catalog"]["claim_after_pass"], "manual-live-verified") + self.assertEqual(table_by_client["Trino Iceberg REST catalog"]["write_claim_after_pass"], "not-claimed") + self.assertEqual(table_by_client["DuckDB Iceberg"]["write_claim_after_pass"], "not-claimed") + self.assertIn("manual-live", " ".join(evidence["promotion_rules"])) + self.assertIn("not-claimed", " ".join(evidence["promotion_rules"])) + + def test_production_operations_guide_covers_release_boundaries(self) -> None: + guide = engine_compatibility.production_operations_guide( + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + namespace="smoke", + table="events", + rest_path="/iceberg", + ) + + self.assertEqual(guide["claim_boundary"], "Iceberg REST Catalog S3 Tables implementation") + section_by_name = {section["name"]: section for section in guide["sections"]} + + self.assertIn("live-client-conformance", section_by_name) + self.assertIn("catalog-backing-cutover", section_by_name) + self.assertIn("maintenance-operations", section_by_name) + self.assertIn("recovery-and-disaster-rehearsal", section_by_name) + self.assertIn("permissions-and-credentials", section_by_name) + self.assertIn("unsupported-claim-governance", section_by_name) + + cutover = section_by_name["catalog-backing-cutover"] + self.assertIn("GET /iceberg/v1/rustfs-s3table-smoke/catalog/migration", cutover["commands"]) + self.assertIn("blockers is empty", " ".join(cutover["pass_criteria"])) + + maintenance = section_by_name["maintenance-operations"] + self.assertIn("quarantine", " ".join(maintenance["fail_closed_signals"])) + self.assertIn("audit-events", " ".join(maintenance["required_evidence"])) + + unsupported = section_by_name["unsupported-claim-governance"] + self.assertIn("full AWS S3 Tables control-plane API parity", unsupported["not_claimed"]) + self.assertNotIn("fully compatible", json.dumps(guide).lower()) + + def test_production_operations_guide_threads_target_args_into_commands(self) -> None: + guide = engine_compatibility.production_operations_guide( + endpoint="https://rustfs.example.com", + warehouse="lakehouse-prod", + namespace="sales", + table="orders", + rest_path="/_iceberg", + ) + + live = {section["name"]: section for section in guide["sections"]}["live-client-conformance"] + commands = "\n".join(live["commands"]) + + self.assertIn("--endpoint https://rustfs.example.com", commands) + self.assertIn("--bucket lakehouse-prod", commands) + self.assertIn("--warehouse lakehouse-prod", commands) + self.assertIn("--namespace sales", commands) + self.assertIn("--table orders", commands) + self.assertIn("--rest-path /_iceberg", commands) + self.assertNotIn("rustfs-s3table-smoke", commands) + + def test_production_operations_guide_encodes_catalog_path_segments(self) -> None: + guide = engine_compatibility.production_operations_guide( + endpoint="http://127.0.0.1:9000", + warehouse="lake bucket", + namespace="sales/prod", + table="orders table", + rest_path="/_iceberg", + ) + + section_by_name = {section["name"]: section for section in guide["sections"]} + cutover = section_by_name["catalog-backing-cutover"] + maintenance = section_by_name["maintenance-operations"] + recovery = section_by_name["recovery-and-disaster-rehearsal"] + credentials = section_by_name["permissions-and-credentials"] + + self.assertIn("GET /_iceberg/v1/lake%20bucket/catalog/migration", cutover["commands"]) + encoded_table_path = "/_iceberg/v1/lake%20bucket/namespaces/sales%2Fprod/tables/orders%20table" + for commands in [maintenance["commands"], recovery["commands"], credentials["commands"]]: + rendered = "\n".join(commands) + self.assertIn(encoded_table_path, rendered) + self.assertNotIn("orders table", rendered) + self.assertNotIn("/namespaces/sales/prod/", rendered) + def test_cli_prints_live_conformance_harness(self) -> None: payload = engine_compatibility.cli_json( [ @@ -295,6 +404,29 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertEqual(document["live_conformance"]["mode"], "manual-or-ci-optional") self.assertEqual(document["live_conformance"]["expected_results"]["row_count"], 2) + self.assertIn("evidence", document["live_conformance"]) + + def test_cli_prints_production_operations_guide(self) -> None: + payload = engine_compatibility.cli_json( + [ + "--print-operations-guide", + "--warehouse", + "rustfs-s3table-smoke", + "--namespace", + "smoke", + "--table", + "events", + ] + ) + document = json.loads(payload) + + self.assertEqual( + document["production_operations_guide"]["claim_boundary"], + "Iceberg REST Catalog S3 Tables implementation", + ) + sections = {section["name"] for section in document["production_operations_guide"]["sections"]} + self.assertIn("live-client-conformance", sections) + self.assertIn("unsupported-claim-governance", sections) def test_live_conformance_harness_sanitizes_sql_file_path(self) -> None: harness = engine_compatibility.live_conformance_harness( diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index f530c945b..47693a406 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -296,6 +296,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): job_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1") quarantine_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1/quarantine") scheduler_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler") + scheduler_run_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler/run") worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run") def fake_signed_request( @@ -317,6 +318,11 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): return {"action": "INSPECT", "report": {"job": {"job-id": "job-1"}}} if (method, path) == ("GET", scheduler_path): return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1", "audit-events": [{"action": "PLANNED"}]}]} + if (method, path) == ("POST", scheduler_run_path): + return { + "report": {"job": {"job-id": "job-2", "status": "QUEUED", "scheduler-id": "pyiceberg-smoke-scheduler"}}, + "scheduler": {"status": "QUEUED"}, + } if (method, path) == ("POST", worker_path): return {"job": {"status": "UNKNOWN"}, "audit-events": [{"action": "WORKER_CONTROL"}]} raise AssertionError(f"unexpected REST request: {method} {path}") @@ -338,6 +344,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") quarantine_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1/quarantine") scheduler_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler") + scheduler_run_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler/run") worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run") diagnostics_path = pyiceberg_smoke.table_endpoint_path(args, "/catalog/diagnostics") @@ -381,6 +388,11 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): return {"action": "INSPECT", "report": {"job": {"job-id": "job-1"}}} if (method, path) == ("GET", scheduler_path): return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1", "audit-events": [{"action": "PLANNED"}]}]} + if (method, path) == ("POST", scheduler_run_path): + return { + "report": {"job": {"job-id": "job-2", "status": "QUEUED", "scheduler-id": "pyiceberg-smoke-scheduler"}}, + "scheduler": {"status": "QUEUED"}, + } if (method, path) == ("POST", worker_path): return {"job": {"status": "PAUSED"}, "audit-events": [{"action": "WORKER_CONTROL"}]} if (method, path) == ("GET", diagnostics_path): @@ -411,6 +423,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertIn(("GET", metadata_location_path, None), calls) self.assertIn(("GET", diagnostics_path, None), calls) self.assertIn(("GET", scheduler_path, None), calls) + self.assertIn(("POST", scheduler_run_path, {"scheduler-id": "pyiceberg-smoke-scheduler"}), calls) self.assertIn(("POST", quarantine_path, {"action": "INSPECT"}), calls) self.assertIn(("POST", worker_path, {}), calls)