test(table-catalog): harden smoke catalog probes (#3535)

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]:

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)