From d5ad1b9aed874e1e3b423c161246136684fa60a4 Mon Sep 17 00:00:00 2001 From: Petr Date: Wed, 22 Jul 2026 13:20:22 +0200 Subject: [PATCH] test(e2e): fix 4 pre-existing nightly failures (swap/clone/stream/config-secret) These four failed on the nightly independently of the expired-token outage (#518) and of the v0.75.0 release. Each is a test-side defect or an environment/consistency quirk, not a product regression -- all verified live against project 5946 on 2026-07-22. No source behavior changes. - clone (test_clone_prod_table_into_dev_branch): STEP 1 created a table on the DEFAULT branch without first creating its bucket. create-table auto-creates the bucket only in a dev branch (to materialize the branch's isolated storage); on the production path the bucket must exist. Added the missing create-bucket step. - config-secret (test_config_create_and_update_encrypt_secret): _read_password indexed data["configuration"], but _run_ok returns the full {status, data} envelope, so the config body is at data["data"]["configuration"]. Corrected the path. - swap (test_swap_in_dev_branch_exchanges_schemas): the swap is correct, but the test read the raw /tables/{id} `definition.columns` block, which is read-after-DDL eventually consistent on this stack -- it kept reporting the pre-swap length for >30s. The CLI's normalized `column_details` reflects the swap immediately, so the test now reads lengths via `storage table-detail` (with a short safety poll), matching what a user/agent actually sees. - stream (test_stream_otlp_e2e): the fixed source name `kbagent-e2e-stream` had a wedged server-side delete task (unremovable via the API), and --if-not-exists reused it every run, so cleanup could never succeed. Use a unique per-run source name (fresh sources create+delete in ~3s), tolerate a retryable TIMEOUT on delete by confirming eventual removal via a polled list, and best-effort drop the auto-provisioned sink bucket that lingers after the source is gone. make check green (4648 passed). Test-only diff. --- tests/test_e2e.py | 135 ++++++++++++++++++++++++++++++++++++---------- 1 file changed, 108 insertions(+), 27 deletions(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 4657726f..89145454 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -7811,14 +7811,27 @@ def test_swap_in_dev_branch_exchanges_schemas(self) -> None: "id", ) - def _value_len(table_detail: dict[str, Any]) -> str: - cols = {c["name"]: c for c in table_detail["definition"]["columns"]} - return cols["value"]["definition"]["length"] + # Read the 'value' length via the CLI's normalized `column_details` + # (what a user/agent sees). The raw `/tables/{id}` `definition.columns` + # block is read-after-DDL eventually consistent on this stack -- after a + # swap it can keep reporting the pre-swap length for >30s -- whereas + # `column_details` reflects the swap at once (verified 2026-07-22). + def _value_len(table_id: str) -> str: + detail = self._run_ok( + "storage", + "table-detail", + "--project", + self.alias, + "--table-id", + table_id, + "--branch", + str(branch_id), + )["data"] + cols = {c["name"]: c for c in detail["column_details"]} + return str(cols["value"]["length"]) - before_original = self.client.get_table_detail(original_id, branch_id=branch_id) - before_typed = self.client.get_table_detail(typed_id, branch_id=branch_id) - assert _value_len(before_original) == "20" - assert _value_len(before_typed) == "80" + assert _value_len(original_id) == "20" + assert _value_len(typed_id) == "80" _step(3, "storage swap-tables", "POST /tables/.../swap with targetTableId") result = self._run_ok( @@ -7840,15 +7853,30 @@ def _value_len(table_detail: dict[str, Any]) -> str: assert result["dry_run"] is False _step(4, "table-detail", "verify VARCHAR lengths exchanged") - after_original = self.client.get_table_detail(original_id, branch_id=branch_id) - after_typed = self.client.get_table_detail(typed_id, branch_id=branch_id) - assert _value_len(after_original) == "80", ( + + def _poll_value_len(table_id: str, expected: str) -> str: + """Poll column_details until 'value' has the expected length (max ~30s). + + Small safety margin for any residual read lag after the swap job + completes; column_details is normally consistent immediately. + """ + length = "" + for _ in range(15): + length = _value_len(table_id) + if length == expected: + return length + time.sleep(2) + return length + + after_original_len = _poll_value_len(original_id, "80") + after_typed_len = _poll_value_len(typed_id, "20") + assert after_original_len == "80", ( f"After swap, '{original_id}' should adopt the schema of '{typed_id}' " - f"(VARCHAR(80)); got VARCHAR({_value_len(after_original)})." + f"(VARCHAR(80)); got VARCHAR({after_original_len})." ) - assert _value_len(after_typed) == "20", ( + assert after_typed_len == "20", ( f"After swap, '{typed_id}' should adopt the schema of '{original_id}' " - f"(VARCHAR(20)); got VARCHAR({_value_len(after_typed)})." + f"(VARCHAR(20)); got VARCHAR({after_typed_len})." ) def test_swap_dry_run_does_not_call_api(self) -> None: @@ -8012,10 +8040,28 @@ def _run_ok(self, *args: str) -> dict[str, Any]: def test_clone_prod_table_into_dev_branch(self) -> None: """Live pull: a production table becomes available in the dev branch.""" - bucket_id = f"in.c-{RUN_ID.replace('-', '_')}_clone" + bucket_stage = "in" + bucket_name = f"{RUN_ID.replace('-', '_')}_clone" + bucket_id = f"in.c-{bucket_name}" table_id = f"{bucket_id}.source" - _step(1, "create a production table (default branch)") + # create-table on the DEFAULT branch does NOT auto-create the bucket + # (that convenience is dev-branch-only, to materialize the branch's + # isolated storage). The production path needs the bucket first. + _step(1, "create the destination bucket (default branch)") + self._run_ok( + "storage", + "create-bucket", + "--project", + self.alias, + "--stage", + bucket_stage, + "--name", + bucket_name, + ) + self._created_buckets.append(bucket_id) + + _step(2, "create a production table (default branch)") self._run_ok( "storage", "create-table", @@ -8032,16 +8078,15 @@ def test_clone_prod_table_into_dev_branch(self) -> None: "--primary-key", "id", ) - self._created_buckets.append(bucket_id) - _step(2, "branch create", "target dev branch for the pull") + _step(3, "branch create", "target dev branch for the pull") branch = self._run_ok( "branch", "create", "--project", self.alias, "--name", f"{RUN_ID}-clone-branch" )["data"] branch_id = int(branch["branch_id"]) self._created_branch_ids.append(branch_id) - _step(3, "storage clone-table", "POST /tables/.../pull (default -> branch)") + _step(4, "storage clone-table", "POST /tables/.../pull (default -> branch)") result = self._run_ok( "storage", "clone-table", @@ -8057,7 +8102,7 @@ def test_clone_prod_table_into_dev_branch(self) -> None: assert result["dry_run"] is False assert result["response"]["status"] == "success" - _step(4, "table-detail in branch", "table is materialized/visible after pull") + _step(5, "table-detail in branch", "table is materialized/visible after pull") detail = self.client.get_table_detail(table_id, branch_id=branch_id) col_names = {c["name"] for c in detail["definition"]["columns"]} assert col_names == {"id", "value"} @@ -9375,7 +9420,12 @@ def test_stream_otlp_e2e(tmp_path: Path) -> None: config_dir = tmp_path / "kbagent-config" config_dir.mkdir() alias = "e2e-stream-target" - source_name = "kbagent-e2e-stream" + # Unique per run: a fixed name reuses whatever source is already there, and + # a source whose async delete once got wedged server-side (unremovable via + # the API) would then red every subsequent run. A fresh per-run source + # creates and deletes cleanly (~3s) and is isolated from any orphan. + source_name = f"kbagent-{RUN_ID}" + sink_bucket = f"in.c-otlp-{source_name}" def _run(*args: str) -> Any: return runner.invoke( @@ -9432,14 +9482,43 @@ def _run(*args: str) -> Any: assert "/***" not in rdata["endpoint"] finally: # 5. Clean up -- delete the source and confirm it is gone. + # + # Deleting an OTLP source tears down its auto-provisioned sinks, which + # runs as an async Stream task. It is usually fast (~3s) but under CI + # load has exceeded the client's task timeout, surfacing as a retryable + # TIMEOUT (exit 4) even though the delete keeps running server-side. + # Treat that as "delete in progress" and confirm eventual removal via + # the list below, so a slow-but-successful teardown does not red the run. deleted = _run("stream", "delete", source_name, "--project", alias, "--yes") - assert deleted.exit_code == 0, deleted.output - assert json.loads(deleted.output)["data"]["status"] == "deleted" - - final = _run("stream", "list", "--project", alias) - remaining = {s["source_id"] for s in json.loads(final.output)["data"]["sources"]} + if deleted.exit_code == 0: + assert json.loads(deleted.output)["data"]["status"] == "deleted" + else: + payload = json.loads(deleted.output) + assert payload["error"]["code"] == "TIMEOUT", deleted.output + + # Poll the source list until the (possibly still-processing) delete lands. + remaining: set[str] = set() + for _ in range(30): + final = _run("stream", "list", "--project", alias) + remaining = {s["source_id"] for s in json.loads(final.output)["data"]["sources"]} + if source_name not in remaining: + break + time.sleep(2) assert source_name not in remaining + # Best-effort: drop the auto-provisioned sink bucket, which lingers after + # the source is gone. Failure here is not a test failure (throwaway project). + _run( + "storage", + "delete-bucket", + "--project", + alias, + "--bucket-id", + sink_bucket, + "--force", + "--yes", + ) + # --------------------------------------------------------------------------- # MCP-parity commands (since v0.30.0) @@ -11930,7 +12009,7 @@ def _run_ok(self, *args: str) -> dict[str, Any]: return _json_ok(_invoke(self.config_dir, ["--json", *args])) def _read_password(self, config_id: str) -> str: - data = self._run_ok( + envelope = self._run_ok( "config", "detail", "--project", @@ -11940,7 +12019,9 @@ def _read_password(self, config_id: str) -> str: "--config-id", config_id, ) - return data["configuration"]["parameters"]["#password"] + # _run_ok returns the full {status, data} envelope; the config body + # lives under .data.configuration (config detail is single-project here). + return envelope["data"]["configuration"]["parameters"]["#password"] def test_config_create_and_update_encrypt_secret(self) -> None: _step(1, "config new --push with a #password", "must encrypt before write")