From 742b82354ca573df12c4d4b7fb41a189aeae3623 Mon Sep 17 00:00:00 2001 From: Chaitanya Panchal Date: Mon, 24 Aug 2026 21:02:12 -0400 Subject: [PATCH] fix(tests): make Athena S3 prefix tests self-cleaning to avoid flaky CI reruns test_clear_partition_data and test_hive_truncate_table asserted their S3 test prefix started empty. That prefix is derived from the pytest session's testrun_uid plus the test's name, so it stays stable across pytest-rerunfailures retries within the same session. If a prior attempt got far enough to CTAS into that prefix before failing for an unrelated reason, the retry's very first assertion would fail deterministically since nothing ever cleared the objects left behind. Add an s3_delete_objects helper (mirroring the existing s3_list_objects helper) and call it before the emptiness assertion in both tests so they clean up after any previous attempt instead of assuming a clean slate. Fixes #5971 Signed-off-by: Chaitanya Panchal --- .../integration/test_integration_athena.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/core/engine_adapter/integration/test_integration_athena.py b/tests/core/engine_adapter/integration/test_integration_athena.py index 9d23af206e..a4b91f823d 100644 --- a/tests/core/engine_adapter/integration/test_integration_athena.py +++ b/tests/core/engine_adapter/integration/test_integration_athena.py @@ -48,8 +48,21 @@ def s3_list_objects(s3: t.Any, location: str, **list_objects_kwargs: t.Any) -> t return lst +def s3_delete_objects(s3: t.Any, location: str) -> None: + # The S3 location for a given test is stable across pytest-rerunfailures retries within the same + # test session (it's derived from the session's testrun_uid + the test's name), so a prior failed + # attempt can leave objects behind that a subsequent retry would otherwise trip over. Proactively + # clearing the prefix makes these tests self-healing instead of just asserting it's already empty. + bucket, prefix = parse_s3_uri(location) + for page in s3.get_paginator("list_objects_v2").paginate(Bucket=bucket, Prefix=prefix): + objects = [{"Key": o["Key"]} for o in page.get("Contents", [])] + if objects: + s3.delete_objects(Bucket=bucket, Delete={"Objects": objects}) + + def test_clear_partition_data(ctx: TestContext, engine_adapter: AthenaEngineAdapter, s3: t.Any): base_uri = engine_adapter.s3_warehouse_location_or_raise + s3_delete_objects(s3, base_uri) assert len(s3_list_objects(s3, base_uri)) == 0 src_table = ctx.table("src_table") @@ -239,6 +252,7 @@ def test_hive_truncate_table(ctx: TestContext, engine_adapter: AthenaEngineAdapt ] ) + s3_delete_objects(s3, base_uri) assert len(s3_list_objects(s3, base_uri)) == 0 engine_adapter.ctas(table_name=table_1, query_or_df=base_data)