[change] Removed redundant block() utils function

tests/integration/adls_asset_test.py

@@ -11,11 +11,9 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from pyatlan.utils import get_parent_qualified_name
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("ADLS")
 
@@ -33,9 +31,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/airflow_asset_test.py

@@ -13,10 +13,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("AIRFLOW")
 
@@ -31,9 +29,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/anaplan_asset_test.py

@@ -23,10 +23,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("ANAPLAN")
 
@@ -56,9 +54,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/api_asset_test.py

@@ -19,10 +19,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("API")
 
@@ -60,9 +58,6 @@
 API_FIELD_REFERENCE_OBJECT_NAME = f"{MODULE_NAME}-api-object-reference"
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/app_asset_test.py

@@ -11,10 +11,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("APP")
 
@@ -29,9 +27,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/azure_event_hub_asset_test.py

@@ -14,10 +14,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("AZURE_EVENT_HUB")
 
@@ -31,9 +29,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/custom_asset_test.py

@@ -11,10 +11,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("CUSTOM")
 
@@ -27,8 +25,6 @@
 CERTIFICATE_MESSAGE = "Automated testing of the Python SDK."
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:

tests/integration/data_mesh_test.py

@@ -25,11 +25,10 @@
     EntityStatus,
 )
 from pyatlan.model.fluent_search import FluentSearch
-from pyatlan.model.response import AssetMutationResponse
 from pyatlan.model.typedef import AttributeDef, CustomMetadataDef
 from tests.integration.client import TestId, delete_asset
 from tests.integration.custom_metadata_test import create_custom_metadata
-from tests.integration.utils import block, wait_for_successful_custometadatadef_purge
+from tests.integration.utils import wait_for_successful_custometadatadef_purge
 
 DATA_PRODUCT_ASSETS_PLAYBOOK_FILTER = (
     '{"condition":"AND","isGroupLocked":false,"rules":[]}'
@@ -59,8 +58,6 @@
 ANNOUNCEMENT_TITLE = "Python SDK testing."
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def domain(client: AtlanClient) -> Generator[DataDomain, None, None]:

tests/integration/data_studio_asset_test.py

@@ -12,10 +12,8 @@
     EntityStatus,
     GoogleDatastudioAssetType,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("datastudio")
 
@@ -28,8 +26,6 @@
 ANNOUNCEMENT_TITLE = "Python SDK testing."
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:

tests/integration/gcs_asset_test.py

@@ -11,10 +11,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("GCS")
 
@@ -28,8 +26,6 @@
 ANNOUNCEMENT_TITLE = "Python SDK testing."
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:

tests/integration/insights_test.py

@@ -8,9 +8,7 @@
 from pyatlan.model.assets import Collection, Folder, Query, Schema
 from pyatlan.model.enums import AtlanConnectorType, EntityStatus
 from pyatlan.model.fluent_search import FluentSearch
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
-from tests.integration.utils import block
 
 PREFIX = TestId.make_unique("INS")
 
@@ -25,8 +23,6 @@
 SCHEMA_NAME = "WIDE_WORLD_IMPORTERS"
 USER_DESCRIPTION = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def collection(client: AtlanClient) -> Generator[Collection, None, None]:

tests/integration/kafka_asset_test.py

@@ -14,10 +14,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("KAFKA")
 
@@ -31,9 +29,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/preset_asset_test.py

@@ -17,10 +17,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("PRESET")
 
@@ -38,8 +36,6 @@
 ANNOUNCEMENT_TITLE = "Python SDK testing."
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:

tests/integration/s3_asset_test.py

@@ -13,10 +13,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("S3")
 
@@ -34,9 +32,6 @@
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
 
-response = block(AtlanClient(), AssetMutationResponse())
-
-
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:
     result = create_connection(

tests/integration/superset_asset_test.py

@@ -16,10 +16,8 @@
     CertificateStatus,
     EntityStatus,
 )
-from pyatlan.model.response import AssetMutationResponse
 from tests.integration.client import TestId, delete_asset
 from tests.integration.connection_test import create_connection
-from tests.integration.utils import block
 
 MODULE_NAME = TestId.make_unique("SUPERSET")
 
@@ -35,8 +33,6 @@
 ANNOUNCEMENT_TITLE = "Python SDK testing."
 ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK."
 
-response = block(AtlanClient(), AssetMutationResponse())
-
 
 @pytest.fixture(scope="module")
 def connection(client: AtlanClient) -> Generator[Connection, None, None]:

tests/integration/utils.py

@@ -1,6 +1,3 @@
-import time
-from typing import List
-
 from tenacity import (
     retry,
     retry_if_exception_type,
@@ -9,48 +6,10 @@
 )
 
 from pyatlan.client.atlan import AtlanClient
-from pyatlan.errors import AtlanError, ErrorCode, NotFoundError
-from pyatlan.model.assets import Asset
-from pyatlan.model.enums import EntityStatus
-from pyatlan.model.response import AssetMutationResponse
+from pyatlan.errors import AtlanError
 from pyatlan.model.typedef import AtlanTagDef, CustomMetadataDef, EnumDef
 
 
-def block(
-    client: AtlanClient, response: AssetMutationResponse
-) -> AssetMutationResponse:
-    if response.mutated_entities and response.mutated_entities.DELETE:
-        retrieve_and_check_assets(client, response.mutated_entities.DELETE, 0)
-    return response
-
-
-def retrieve_and_check_assets(
-    client: AtlanClient, to_check: List[Asset], retry_count: int
-):
-    """
-    Retrieve and check the status of a list of assets and retry if needed.
-    """
-    leftovers = []
-    for one in to_check:
-        try:
-            candidate = client.asset.get_by_guid(
-                one.guid, asset_type=type(one), ignore_relationships=False
-            )
-            if candidate and candidate.status == EntityStatus.ACTIVE:
-                leftovers.append(candidate)
-        except NotFoundError:
-            # If it is not found, it was successfully deleted (purged), so we
-            # do not need to look for it any further
-            print("Asset no longer exists.")
-        except AtlanError:
-            leftovers.append(one)
-    if leftovers:
-        if retry_count == 20:
-            raise ErrorCode.RETRY_OVERRUN.exception_with_parameters()
-        time.sleep(2)
-        retrieve_and_check_assets(client, leftovers, retry_count + 1)
-
-
 @retry(
     retry=retry_if_exception_type(AtlanError),
     wait=wait_random_exponential(multiplier=1, max=5),