refactor quicksight test resources as fixtures (#1928)

Author: malachi-constant

tests/conftest.py

@@ -1,4 +1,5 @@
 import os
+import uuid
 from datetime import datetime
 from importlib import reload
 from types import ModuleType
@@ -242,6 +243,26 @@ def glue_table2(glue_database) -> str:
     wr.catalog.delete_table_if_exists(database=glue_database, table=name)
 
 
+@pytest.fixture(scope="function")
+def quicksight_datasource() -> str:
+    name = f"test{str(uuid.uuid4())[:8]}"
+    print(f"Quicksight Data Source: {name}")
+    wr.quicksight.delete_all_data_sources(regex_filter=name)
+    yield name
+    wr.quicksight.delete_all_data_sources(regex_filter=name)
+    print(f"Quicksight Data Source: {name} deleted")
+
+
+@pytest.fixture(scope="function")
+def quicksight_dataset() -> str:
+    name = f"test{str(uuid.uuid4())[:8]}"
+    print(f"Quicksight Dataset: {name}")
+    wr.quicksight.delete_all_datasets(regex_filter=name)
+    yield name
+    wr.quicksight.delete_all_datasets(regex_filter=name)
+    print(f"Quicksight Dataset: {name} deleted")
+
+
 @pytest.fixture(scope="function")
 def path(bucket):
     yield from path_generator(bucket)

tests/test_quicksight.py

@@ -14,34 +14,30 @@
 
 
 @pytest.mark.xfail(raises=client.exceptions.ConflictException)
-def test_quicksight(path, glue_database, glue_table):
+def test_quicksight(path, quicksight_datasource, quicksight_dataset, glue_database, glue_table):
     df = get_df_quicksight()
     wr.s3.to_parquet(
         df=df, path=path, dataset=True, database=glue_database, table=glue_table, partition_cols=["par0", "par1"]
     )
-    resource_name = f"test{str(uuid.uuid4())[:8]}"
-
-    wr.quicksight.delete_all_dashboards(regex_filter="test.*")
-    wr.quicksight.delete_all_datasets(regex_filter="test.*")
-    wr.quicksight.delete_all_data_sources(regex_filter="test.*")
-    wr.quicksight.delete_all_templates(regex_filter="test.*")
 
     wr.quicksight.create_athena_data_source(
-        name=resource_name, allowed_to_manage=[wr.sts.get_current_identity_name()], tags={"Env": "aws-sdk-pandas"}
+        name=quicksight_datasource,
+        allowed_to_manage=[wr.sts.get_current_identity_name()],
+        tags={"Env": "aws-sdk-pandas"},
     )
-    assert wr.quicksight.describe_data_source(resource_name)["Name"] == resource_name
+    assert wr.quicksight.describe_data_source(quicksight_datasource)["Name"] == quicksight_datasource
     assert (
-        wr.quicksight.describe_data_source_permissions(resource_name)[0]["Principal"].endswith(
+        wr.quicksight.describe_data_source_permissions(quicksight_datasource)[0]["Principal"].endswith(
             wr.sts.get_current_identity_name()
         )
         is True
     )
-    dataset_name = f"{resource_name}-table"
+    dataset_name = f"{quicksight_dataset}-table"
     wr.quicksight.create_athena_dataset(
         name=dataset_name,
         database=glue_database,
         table=glue_table,
-        data_source_name=resource_name,
+        data_source_name=quicksight_datasource,
         allowed_to_manage=[wr.sts.get_current_identity_name()],
         rename_columns={"iint16": "new_col"},
         cast_columns_types={"new_col": "STRING"},
@@ -50,9 +46,9 @@ def test_quicksight(path, glue_database, glue_table):
     assert wr.quicksight.describe_dataset(dataset_name)["Name"] == dataset_name
 
     wr.quicksight.create_athena_dataset(
-        name=f"{resource_name}-sql",
+        name=f"{quicksight_dataset}-sql",
         sql=f"SELECT * FROM {glue_database}.{glue_table}",
-        data_source_name=resource_name,
+        data_source_name=quicksight_datasource,
         import_mode="SPICE",
         allowed_to_use=[wr.sts.get_current_identity_name()],
         allowed_to_manage=[wr.sts.get_current_identity_name()],
@@ -62,26 +58,26 @@ def test_quicksight(path, glue_database, glue_table):
         tags={"foo": "boo"},
     )
 
-    ingestion_id = wr.quicksight.create_ingestion(f"{resource_name}-sql")
+    ingestion_id = wr.quicksight.create_ingestion(f"{quicksight_dataset}-sql")
     status = None
     while status not in ["FAILED", "COMPLETED", "CANCELLED"]:
-        status = wr.quicksight.describe_ingestion(ingestion_id, f"{resource_name}-sql")["IngestionStatus"]
+        status = wr.quicksight.describe_ingestion(ingestion_id, f"{quicksight_dataset}-sql")["IngestionStatus"]
     assert status == "COMPLETED"
 
-    ingestion_id = wr.quicksight.create_ingestion(f"{resource_name}-sql")
-    wr.quicksight.cancel_ingestion(ingestion_id, f"{resource_name}-sql")
-    assert len(wr.quicksight.list_ingestions(f"{resource_name}-sql")) == 3
+    ingestion_id = wr.quicksight.create_ingestion(f"{quicksight_dataset}-sql")
+    wr.quicksight.cancel_ingestion(ingestion_id, f"{quicksight_dataset}-sql")
+    assert len(wr.quicksight.list_ingestions(f"{quicksight_dataset}-sql")) == 3
 
     wr.quicksight.list_groups()
     wr.quicksight.list_iam_policy_assignments()
     wr.quicksight.list_iam_policy_assignments_for_user(wr.sts.get_current_identity_name())
     wr.quicksight.list_user_groups(wr.sts.get_current_identity_name())
     wr.quicksight.list_users()
-    wr.quicksight.get_dataset_ids(f"{resource_name}-sql")
+    wr.quicksight.get_dataset_ids(f"{quicksight_dataset}-sql")
     wr.quicksight.get_data_source_ids("test")
 
-    wr.quicksight.delete_all_datasets(regex_filter="test.*")
-    wr.quicksight.delete_all_data_sources(regex_filter="test.*")
+    wr.quicksight.delete_all_datasets(regex_filter=quicksight_dataset)
+    wr.quicksight.delete_all_data_sources(regex_filter=quicksight_datasource)
 
 
 def test_quicksight_delete_all_datasources_filter():