@@ -48,8 +48,21 @@ def s3_list_objects(s3: t.Any, location: str, **list_objects_kwargs: t.Any) -> t
4848 return lst
4949
5050
51+ def s3_delete_objects (s3 : t .Any , location : str ) -> None :
52+ # The S3 location for a given test is stable across pytest-rerunfailures retries within the same
53+ # test session (it's derived from the session's testrun_uid + the test's name), so a prior failed
54+ # attempt can leave objects behind that a subsequent retry would otherwise trip over. Proactively
55+ # clearing the prefix makes these tests self-healing instead of just asserting it's already empty.
56+ bucket , prefix = parse_s3_uri (location )
57+ for page in s3 .get_paginator ("list_objects_v2" ).paginate (Bucket = bucket , Prefix = prefix ):
58+ objects = [{"Key" : o ["Key" ]} for o in page .get ("Contents" , [])]
59+ if objects :
60+ s3 .delete_objects (Bucket = bucket , Delete = {"Objects" : objects })
61+
62+
5163def test_clear_partition_data (ctx : TestContext , engine_adapter : AthenaEngineAdapter , s3 : t .Any ):
5264 base_uri = engine_adapter .s3_warehouse_location_or_raise
65+ s3_delete_objects (s3 , base_uri )
5366 assert len (s3_list_objects (s3 , base_uri )) == 0
5467
5568 src_table = ctx .table ("src_table" )
@@ -239,6 +252,7 @@ def test_hive_truncate_table(ctx: TestContext, engine_adapter: AthenaEngineAdapt
239252 ]
240253 )
241254
255+ s3_delete_objects (s3 , base_uri )
242256 assert len (s3_list_objects (s3 , base_uri )) == 0
243257
244258 engine_adapter .ctas (table_name = table_1 , query_or_df = base_data )
0 commit comments