From ea70d5697ae62231ba9be7f7ee5d3bc05c84820b Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 18 Jun 2026 11:01:09 +0800 Subject: [PATCH] test(table-catalog): harden smoke catalog probes (#3535) --- scripts/table-catalog/pyiceberg_smoke.py | 50 +++- scripts/table-catalog/test_pyiceberg_smoke.py | 279 +++++++++++++++++- 2 files changed, 317 insertions(+), 12 deletions(-) diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index c835fd1ab..84ddd2654 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -25,6 +25,7 @@ REQUIRED_STORAGE_CREDENTIAL_KEYS = ( "s3.session-token", ) TABLE_MAINTENANCE_CONFIG_VERSION = 1 +IDENTIFIER_SEGMENT_MAX_LEN = 64 PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "rustfs": { @@ -559,6 +560,18 @@ def safe_probe_segment(namespace: str, table: str) -> str: return "".join(char if char.isalnum() or char in {"-", "_", "."} else "-" for char in source) +def safe_ref_segment(namespace: str, table: str) -> str: + source = f"smoke-{namespace}-{table}".lower() + segment = "".join( + char if char.isascii() and (char.isalnum() or char in {"-", "_"}) else "-" + for char in source + ) + segment = segment.strip("-_") + if not segment: + return "smoke" + return segment[:IDENTIFIER_SEGMENT_MAX_LEN].rstrip("-_") or "smoke" + + def scope_probe_keys(object_prefix: str, namespace: str, table: str) -> tuple[str, str]: segment = safe_probe_segment(namespace, table) suffix = f"{segment}-{int(time.time())}.txt" @@ -792,21 +805,40 @@ def run_metadata_location_probe(args: argparse.Namespace, deps: RuntimeDeps, tab def run_table_ref_probe(args: argparse.Namespace, deps: RuntimeDeps, snapshot_id: int) -> None: - ref_name = f"smoke-{safe_probe_segment(args.namespace, args.table)}" + ref_name = safe_ref_segment(args.namespace, args.table) ref_body = { "snapshot-id": snapshot_id, "type": "tag", "max-ref-age-ms": 86_400_000, } + force_deleted = False signed_rest_request(args, deps, "PUT", table_ref_endpoint_path(args, ref_name), ref_body) - refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args)) - if refs.get("refs", {}).get(ref_name, {}).get("snapshot-id") != snapshot_id: - raise RuntimeError("table refs endpoint did not return the smoke tag") - signed_rest_request_expect_error(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), expected_statuses={400}) - signed_rest_request(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), {"force": True}) - refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args)) - if ref_name in refs.get("refs", {}): - raise RuntimeError("table refs endpoint still returned the deleted smoke tag") + try: + refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args)) + if refs.get("refs", {}).get(ref_name, {}).get("snapshot-id") != snapshot_id: + raise RuntimeError("table refs endpoint did not return the smoke tag") + unforced_delete = signed_rest_request_expect_error( + args, + deps, + "DELETE", + table_ref_endpoint_path(args, ref_name), + {}, + expected_statuses={400}, + ) + if "force is required" not in unforced_delete.response_body.lower(): + raise RuntimeError("unforced table ref delete did not report the retention force requirement") + signed_rest_request(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), {"force": True}) + force_deleted = True + refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args)) + if ref_name in refs.get("refs", {}): + raise RuntimeError("table refs endpoint still returned the deleted smoke tag") + finally: + if not force_deleted: + try: + signed_rest_request(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), {"force": True}) + except RestRequestError as error: + if error.status_code != 404: + raise def smoke_view_request(args: argparse.Namespace, view_name: str, version_id: int, sql: str) -> dict[str, Any]: diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 1412b5ed3..2de09db71 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -7,6 +7,9 @@ import os import re import sys import unittest +from contextlib import redirect_stdout +from io import StringIO +from types import SimpleNamespace from pathlib import Path from unittest import mock @@ -136,6 +139,17 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertTrue(config["worker-paused"]) self.assertEqual(config["max-retry-attempts"], 0) + def test_safe_ref_segment_matches_rustfs_identifier_segment_rules(self) -> None: + ref_name = pyiceberg_smoke.safe_ref_segment( + "sales.analytics", + "Orders With Spaces And A Very Long Name That Needs To Be Cut Down Safely", + ) + + self.assertLessEqual(len(ref_name), 64) + self.assertNotIn(".", ref_name) + self.assertRegex(ref_name, r"^[a-z0-9][a-z0-9_-]*[a-z0-9]$") + self.assertTrue(ref_name.startswith("smoke-sales-analytics-orders")) + def test_expected_error_helper_returns_matching_rest_error(self) -> None: args = self.parse_with_args([]) expected = pyiceberg_smoke.RestRequestError("DELETE", "/path", 400, "bad request") @@ -180,6 +194,80 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): with self.assertRaisesRegex(RuntimeError, "current snapshot id"): pyiceberg_smoke.current_snapshot_id_from_table_response({"metadata": {}}) + def test_metadata_location_probe_rejects_pointer_mismatch(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + + with mock.patch.object( + pyiceberg_smoke, + "signed_rest_request", + return_value={"metadata-location": "s3://lake/tables/id/metadata/v1.metadata.json", "version-token": "token-1"}, + ): + with self.assertRaisesRegex(RuntimeError, "does not match"): + pyiceberg_smoke.run_metadata_location_probe( + args, + mock.Mock(), + {"metadata-location": "s3://lake/tables/id/metadata/v2.metadata.json"}, + ) + + def test_view_probe_drops_smoke_view_after_load_failure(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + view_path = pyiceberg_smoke.view_endpoint_path(args) + calls: list[tuple[str, str, object]] = [] + + def fake_signed_request( + _args: object, + _deps: object, + method: str, + path: str, + body: object = None, + ) -> dict[str, object]: + calls.append((method, path, body)) + if (method, path) == ("POST", view_path): + return {} + if (method, path) == ("GET", view_path): + return {"identifiers": [{"name": "orders_smoke_view"}]} + if (method, path) == ("GET", f"{view_path}/orders_smoke_view"): + return {} + if (method, path) == ("DELETE", f"{view_path}/orders_smoke_view"): + return {} + raise AssertionError(f"unexpected REST request: {method} {path}") + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with self.assertRaisesRegex(RuntimeError, "metadata-location"): + pyiceberg_smoke.run_view_probe(args, mock.Mock()) + + self.assertIn(("DELETE", f"{view_path}/orders_smoke_view", None), calls) + + def test_maintenance_probe_rejects_unknown_worker_status(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config") + maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") + job_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1") + worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run") + + def fake_signed_request( + _args: object, + _deps: object, + method: str, + path: str, + body: object = None, + ) -> dict[str, object]: + if (method, path) == ("PUT", config_path): + return {} + if (method, path) == ("GET", config_path): + return {"version": 1} + if (method, path) == ("POST", maintenance_path): + return {"job": {"job-id": "job-1"}} + if (method, path) == ("GET", job_path): + return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}} + if (method, path) == ("POST", worker_path): + return {"job": {"status": "UNKNOWN"}} + raise AssertionError(f"unexpected REST request: {method} {path}") + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with self.assertRaisesRegex(RuntimeError, "stable job status"): + pyiceberg_smoke.run_maintenance_probe(args, mock.Mock()) + def test_catalog_api_probe_exercises_extended_rest_surfaces(self) -> None: args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) deps = mock.Mock() @@ -187,6 +275,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): table_path = pyiceberg_smoke.table_endpoint_path(args) metadata_location_path = pyiceberg_smoke.table_endpoint_path(args, "/metadata-location") refs_path = pyiceberg_smoke.table_ref_endpoint_path(args) + ref_name = pyiceberg_smoke.safe_ref_segment(args.namespace, args.table) view_path = pyiceberg_smoke.view_endpoint_path(args) config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config") maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") @@ -209,7 +298,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): return {} if (method, path) == ("GET", refs_path): if sum(1 for call in calls if call[:2] == ("GET", refs_path)) == 1: - return {"refs": {"smoke-sales-orders": {"snapshot-id": 7}}} + return {"refs": {ref_name: {"snapshot-id": 7}}} return {"refs": {}} if method == "DELETE" and path.startswith(f"{refs_path}/"): return {} @@ -239,15 +328,197 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): with mock.patch.object( pyiceberg_smoke, "signed_rest_request_expect_error", - return_value=pyiceberg_smoke.RestRequestError("DELETE", f"{refs_path}/smoke-sales-orders", 400, "retained"), + return_value=pyiceberg_smoke.RestRequestError( + "DELETE", + f"{refs_path}/{ref_name}", + 400, + "snapshot ref has retention policy; force is required", + ), ) as expect_error: pyiceberg_smoke.run_catalog_api_probes(args, deps) - expect_error.assert_called_once() + expect_error.assert_called_once_with( + args, + deps, + "DELETE", + f"{refs_path}/{ref_name}", + {}, + expected_statuses={400}, + ) self.assertIn(("GET", metadata_location_path, None), calls) self.assertIn(("GET", diagnostics_path, None), calls) self.assertIn(("POST", worker_path, {}), calls) + def test_table_ref_probe_force_deletes_smoke_ref_after_validation_failure(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + refs_path = pyiceberg_smoke.table_ref_endpoint_path(args) + calls: list[tuple[str, str, object]] = [] + + def fake_signed_request( + _args: object, + _deps: object, + method: str, + path: str, + body: object = None, + ) -> dict[str, object]: + calls.append((method, path, body)) + if method == "PUT" and path.startswith(f"{refs_path}/"): + return {} + if (method, path) == ("GET", refs_path): + return {"refs": {}} + if method == "DELETE" and path.startswith(f"{refs_path}/"): + return {} + raise AssertionError(f"unexpected REST request: {method} {path}") + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with self.assertRaisesRegex(RuntimeError, "smoke tag"): + pyiceberg_smoke.run_table_ref_probe(args, mock.Mock(), 7) + + self.assertIn( + ("DELETE", f"{refs_path}/smoke-sales-orders", {"force": True}), + calls, + ) + + def run_smoke_with_fakes(self, args: object, probe_calls: list[str]) -> list[str]: + events: list[str] = [] + + class FakeArrowTable: + num_rows = 2 + + class FakePyArrowTableFactory: + @staticmethod + def from_pylist(rows: list[dict[str, object]], *, schema: object) -> object: + events.append(f"rows:{len(rows)}") + return object() + + class FakePyArrow: + Table = FakePyArrowTableFactory + + @staticmethod + def int64() -> str: + return "int64" + + @staticmethod + def string() -> str: + return "string" + + @staticmethod + def field(name: str, field_type: str, *, nullable: bool) -> tuple[str, str, bool]: + return (name, field_type, nullable) + + @staticmethod + def schema(fields: list[tuple[str, str, bool]]) -> tuple[tuple[str, str, bool], ...]: + return tuple(fields) + + class FakeScan: + def to_arrow(self) -> FakeArrowTable: + events.append("scan") + return FakeArrowTable() + + class FakeTable: + metadata = SimpleNamespace(location="s3://lake/tables/table-id") + + def append(self, rows: object) -> None: + events.append("append") + + def scan(self) -> FakeScan: + return FakeScan() + + class FakeCatalog: + def create_table(self, identifier: tuple[str, str], *, schema: object) -> FakeTable: + events.append(f"create:{'.'.join(identifier)}") + return FakeTable() + + def load_table(self, identifier: tuple[str, str]) -> FakeTable: + events.append(f"load:{'.'.join(identifier)}") + return FakeTable() + + deps = SimpleNamespace(pyarrow=FakePyArrow(), load_catalog=lambda *_args, **_kwargs: FakeCatalog()) + storage_credential = pyiceberg_smoke.StorageCredential( + prefix="s3://lake/tables/table-id/", + config={ + "s3.access-key-id": "temp-access", + "s3.secret-access-key": "temp-secret", + "s3.session-token": "temp-token", + }, + ) + + with mock.patch.object(pyiceberg_smoke, "ensure_local_proxy_bypass"): + with mock.patch.object(pyiceberg_smoke, "ensure_aws_env"): + with mock.patch.object(pyiceberg_smoke, "ensure_bucket"): + with mock.patch.object(pyiceberg_smoke, "enable_table_bucket"): + with mock.patch.object(pyiceberg_smoke, "install_rustfs_rest_sigv4_adapter"): + with mock.patch.object(pyiceberg_smoke, "ensure_namespace"): + with mock.patch.object( + pyiceberg_smoke, + "load_table_storage_credential", + side_effect=lambda *_args: (events.append("load-vended-credential"), storage_credential)[1], + ): + with mock.patch.object( + pyiceberg_smoke, + "verify_vended_credential_data_plane_scope", + side_effect=lambda *_args: events.append("verify-vended-scope"), + ): + with mock.patch.object( + pyiceberg_smoke, + "run_catalog_api_probes", + side_effect=lambda *_args: ( + events.append("catalog-probes"), + probe_calls.append("catalog-probes"), + ), + ): + with redirect_stdout(StringIO()): + pyiceberg_smoke.run_smoke(args, deps) + return events + + def test_run_smoke_probes_extended_catalog_apis_by_default(self) -> None: + args = self.parse_with_args(["--bucket", "lake", "--namespace", "sales", "--table", "orders"]) + probe_calls: list[str] = [] + + events = self.run_smoke_with_fakes(args, probe_calls) + + self.assertEqual(probe_calls, ["catalog-probes"]) + self.assertLess(events.index("scan"), events.index("catalog-probes")) + self.assertEqual(events[-1], "catalog-probes") + + def test_run_smoke_can_skip_extended_catalog_api_probes(self) -> None: + args = self.parse_with_args([ + "--bucket", + "lake", + "--namespace", + "sales", + "--table", + "orders", + "--skip-catalog-api-probes", + ]) + probe_calls: list[str] = [] + + events = self.run_smoke_with_fakes(args, probe_calls) + + self.assertEqual(probe_calls, []) + self.assertIn("append", events) + self.assertIn("scan", events) + + def test_run_smoke_vended_credential_flow_still_runs_catalog_api_probes(self) -> None: + args = self.parse_with_args([ + "--profile", + "rustfs-vended-credentials", + "--bucket", + "lake", + "--namespace", + "sales", + "--table", + "orders", + ]) + probe_calls: list[str] = [] + + events = self.run_smoke_with_fakes(args, probe_calls) + + self.assertEqual(probe_calls, ["catalog-probes"]) + self.assertLess(events.index("load-vended-credential"), events.index("verify-vended-scope")) + self.assertLess(events.index("verify-vended-scope"), events.index("append")) + self.assertLess(events.index("scan"), events.index("catalog-probes")) + def test_smoke_view_request_uses_stable_iceberg_view_shape(self) -> None: args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) @@ -474,6 +745,8 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertIn("credential-vending", capabilities) self.assertIn("row-level-delete-update-merge", capabilities) self.assertIn("background-maintenance-worker", capabilities) + self.assertIn("external-catalog-bridge", capabilities) + self.assertIn("multi-table-transactions", capabilities) for entry in inventory: self.assertIn("status", entry) self.assertIn("roadmap_area", entry)