enables refresh mode tests for athena, fixes tests

Author: rudolfix

tests/load/pipeline/test_refresh_modes.py

@@ -99,7 +99,7 @@ def some_data_4():
     "destination_config",
     destinations_configs(
         default_sql_configs=True,
-        subset=["duckdb", "filesystem", "iceberg"],
+        subset=["duckdb", "filesystem", "iceberg", "athena"],
         local_filesystem_configs=True,
         table_format_local_configs=True,
     ),
@@ -110,7 +110,9 @@ def some_data_4():
 def test_refresh_drop_sources(
     destination_config: DestinationTestConfiguration, in_source: bool, with_wipe: bool
 ):
-    pipeline = destination_config.setup_pipeline("refresh_source")
+    pipeline_name = "refresh_source"
+    dataset_name = pipeline_name + uniq_id()
+    pipeline = destination_config.setup_pipeline(pipeline_name, dataset_name=dataset_name)
 
     data: Any = refresh_source(first_run=True, drop_sources=True)
     if not in_source:
@@ -123,7 +125,7 @@ def test_refresh_drop_sources(
     # Second run of pipeline with only selected resources
     if with_wipe:
         pipeline._wipe_working_folder()
-        pipeline = destination_config.setup_pipeline("refresh_source")
+        pipeline = destination_config.setup_pipeline(pipeline_name, dataset_name=dataset_name)
 
     data = refresh_source(first_run=False, drop_sources=True).with_resources(
         "some_data_1", "some_data_2"
@@ -173,7 +175,9 @@ def test_existing_schema_hash(destination_config: DestinationTestConfiguration):
     """Test when new schema is identical to a previously stored schema after dropping and re-creating tables.
     The change should be detected regardless and tables are created again in destination db
     """
-    pipeline = destination_config.setup_pipeline("refresh_full_test", refresh="drop_sources")
+    pipeline = destination_config.setup_pipeline(
+        "refresh_full_test", refresh="drop_sources", dev_mode=True
+    )
 
     info = pipeline.run(
         refresh_source(first_run=True, drop_sources=True), **destination_config.run_kwargs
@@ -210,15 +214,13 @@ def test_existing_schema_hash(destination_config: DestinationTestConfiguration):
     assert new_schema_hash == first_schema_hash
 
 
-pytest.mark.essential
-
-
+@pytest.mark.essential
 @pytest.mark.parametrize(
     "destination_config",
     destinations_configs(
         default_sql_configs=True,
         local_filesystem_configs=True,
-        subset=["duckdb", "filesystem", "iceberg"],
+        subset=["duckdb", "filesystem", "iceberg", "athena"],
         table_format_local_configs=True,
     ),
     ids=lambda x: x.name,
@@ -232,19 +234,21 @@ def test_refresh_drop_resources(
         pytest.skip("not needed")
 
     # First run pipeline with load to destination so tables are created
-    pipeline = destination_config.setup_pipeline("refresh_source")
+    pipeline_name = "refresh_source"
+    dataset_name = pipeline_name + uniq_id()
+    pipeline = destination_config.setup_pipeline(pipeline_name, dataset_name=dataset_name)
 
     data: Any = refresh_source(first_run=True)
     if not in_source:
         data = list(data.selected_resources.values())
 
-    info = pipeline.run(data, refresh="drop_resources", **destination_config.run_kwargs)
+    info = pipeline.run(data, **destination_config.run_kwargs)
     assert_load_info(info)
 
     # Second run of pipeline with only selected resources
     if with_wipe:
         pipeline._wipe_working_folder()
-        pipeline = destination_config.setup_pipeline("refresh_source")
+        pipeline = destination_config.setup_pipeline(pipeline_name, dataset_name=dataset_name)
 
     data = refresh_source(first_run=False).with_resources("some_data_1", "some_data_2")
     if not in_source:
@@ -298,7 +302,9 @@ def test_refresh_drop_resources(
 def test_refresh_drop_data_only(destination_config: DestinationTestConfiguration):
     """Refresh drop_data should truncate all selected tables before load"""
     # First run pipeline with load to destination so tables are created
-    pipeline = destination_config.setup_pipeline("refresh_full_test", refresh="drop_data")
+    pipeline = destination_config.setup_pipeline(
+        "refresh_full_test", refresh="drop_data", dev_mode=True
+    )
 
     info = pipeline.run(
         refresh_source(first_run=True), write_disposition="append", **destination_config.run_kwargs
@@ -408,7 +414,9 @@ def source_2_data_2():
         yield source_2_data_1
         yield source_2_data_2
 
-    pipeline = destination_config.setup_pipeline("refresh_full_test", refresh="drop_sources")
+    pipeline = destination_config.setup_pipeline(
+        "refresh_full_test", refresh="drop_sources", dev_mode=True
+    )
 
     # Run both sources
     info = pipeline.run(
@@ -459,7 +467,7 @@ def source_2_data_2():
     ids=lambda x: x.name,
 )
 def test_refresh_argument_to_run(destination_config: DestinationTestConfiguration):
-    pipeline = destination_config.setup_pipeline("refresh_full_test")
+    pipeline = destination_config.setup_pipeline("refresh_full_test", dev_mode=True)
 
     info = pipeline.run(refresh_source(first_run=True), **destination_config.run_kwargs)
     assert_load_info(info)
@@ -495,7 +503,7 @@ def test_refresh_argument_to_run(destination_config: DestinationTestConfiguratio
     ids=lambda x: x.name,
 )
 def test_refresh_argument_to_extract(destination_config: DestinationTestConfiguration):
-    pipeline = destination_config.setup_pipeline("refresh_full_test")
+    pipeline = destination_config.setup_pipeline("refresh_full_test", dev_mode=True)
 
     info = pipeline.run(refresh_source(first_run=True), **destination_config.run_kwargs)
     assert_load_info(info)