[temp] testing

Author: Aryamanz29

.github/workflows/pyatlan-pr.yaml

@@ -59,6 +59,7 @@ jobs:
           python -m pip install --no-cache-dir --upgrade pip setuptools
           if [ -f requirements.txt ]; then pip install --no-cache-dir -r requirements.txt; fi
           if [ -f requirements-dev.txt ]; then pip install --no-cache-dir -r requirements-dev.txt; fi
+          pip list
 
       - name: QA checks (ruff-format, ruff-lint, mypy)
         run: |
@@ -105,6 +106,7 @@ jobs:
           python -m pip install --no-cache-dir --upgrade pip setuptools
           if [ -f requirements.txt ]; then pip install --no-cache-dir -r requirements.txt; fi
           if [ -f requirements-dev.txt ]; then pip install --no-cache-dir -r requirements-dev.txt; fi
+          pip list
 
       - name: Run integration tests
         env: # Test tenant environment variables

pyatlan/test_utils/base_vcr.py

@@ -13,17 +13,17 @@
         "pytest-vcr plugin is not installed. Please install pytest-vcr."
     )
 
-# Check if vcrpy is installed and ensure the version is 6.0.x
-try:
-    vcr_version = pkg_resources.get_distribution("vcrpy").version
-    if not vcr_version.startswith("6.0"):
-        raise DependencyNotFoundError(
-            f"vcrpy version 6.0.x is required, but found {vcr_version}. Please install the correct version."
-        )
-except pkg_resources.DistributionNotFound:
-    raise DependencyNotFoundError(
-        "vcrpy version 6.0.x is not installed. Please install vcrpy version 6.0.x."
-    )
+# # Check if vcrpy is installed and ensure the version is 6.0.x
+# try:
+#     vcr_version = pkg_resources.get_distribution("vcrpy").version
+#     if not vcr_version.startswith("6.0"):
+#         raise DependencyNotFoundError(
+#             f"vcrpy version 6.0.x is required, but found {vcr_version}. Please install the correct version."
+#         )
+# except pkg_resources.DistributionNotFound:
+#     raise DependencyNotFoundError(
+#         "vcrpy version 6.0.x is not installed. Please install vcrpy version 6.0.x."
+#     )
 
 import json
 import os

requirements-dev.txt

@@ -3,11 +3,12 @@ ruff~=0.9.9
 # [PINNED] for Python 3.8 compatibility
 # higher versions require urllib3>=2.0
 types-requests~=2.31.0.6
+types-setuptools~=75.8.0.20250110
 pytest~=8.3.4
 pytest-vcr~=1.0.2
 # [PINNED] to v6.x since vcrpy>=7.0 requires urllib3>=2.0
 # which breaks compatibility with Python 3.8
-vcrpy~=6.0.2
+# vcrpy~=6.0.2
 pytest-order~=1.3.0
 pytest-timer[termcolor]~=1.0.0
 pytest-sugar~=1.0.0

test_vcr.py

@@ -0,0 +1,107 @@
+import logging
+from typing import Callable, List, Optional
+
+import pytest
+import requests
+
+from pyatlan.client.atlan import AtlanClient
+from pyatlan.model.assets import Asset, Connection
+from pyatlan.model.enums import AtlanConnectorType
+from pyatlan.model.response import AssetMutationResponse
+from pyatlan.test_utils import TestId
+from pyatlan.test_utils.base_vcr import BaseVCR
+
+LOGGER = logging.getLogger(__name__)
+
+
+class TestBaseVCR(BaseVCR):
+    @pytest.mark.vcr()
+    def test_sample_get(self):
+        response = requests.get("https://www.google.com/")
+        assert response.status_code == 201
+
+
+class TestConnection(BaseVCR):
+    connection: Optional[Connection] = None
+
+    @pytest.fixture(scope="module")
+    def client(self) -> AtlanClient:
+        return AtlanClient()
+
+    @pytest.fixture(scope="module")
+    def upsert(self, client: AtlanClient):
+        guids: List[str] = []
+
+        def _upsert(asset: Asset) -> AssetMutationResponse:
+            _response = client.asset.save(asset)
+            if (
+                _response
+                and _response.mutated_entities
+                and _response.mutated_entities.CREATE
+            ):
+                guids.append(_response.mutated_entities.CREATE[0].guid)
+            return _response
+
+        yield _upsert
+
+        # for guid in reversed(guids):
+        #     response = client.asset.purge_by_guid(guid)
+        #     if (
+        #         not response
+        #         or not response.mutated_entities
+        #         or not response.mutated_entities.DELETE
+        #     ):
+        #         LOGGER.error(f"Failed to remove asset with GUID {guid}.")
+
+    @pytest.mark.vcr(cassette_name="TestConnectionCreate.json")
+    def test_create(
+        self,
+        client: AtlanClient,
+        upsert: Callable[[Asset], AssetMutationResponse],
+    ):
+        role = client.role_cache.get_id_for_name("$admin")
+        assert role
+        connection_name = TestId.make_unique("INT")
+        c = Connection.create(
+            name=connection_name,
+            connector_type=AtlanConnectorType.SNOWFLAKE,
+            admin_roles=[role],
+        )
+        assert c.guid
+        response = upsert(c)
+        assert response.mutated_entities
+        assert response.mutated_entities.CREATE
+        assert len(response.mutated_entities.CREATE) == 1
+        assert isinstance(response.mutated_entities.CREATE[0], Connection)
+        assert response.guid_assignments
+        c = response.mutated_entities.CREATE[0]
+        c = client.asset.get_by_guid(c.guid, Connection, ignore_relationships=False)
+        assert isinstance(c, Connection)
+        TestConnection.connection = c
+
+    # @pytest.mark.order(after="test_create")
+    # @pytest.mark.vcr()
+    # def test_create_for_modification(
+    #     self, client: AtlanClient, upsert: Callable[[Asset], AssetMutationResponse]
+    # ):
+    #     assert TestConnection.connection
+    #     assert TestConnection.connection.name
+    #     connection = TestConnection.connection
+    #     description = f"{connection.description} more stuff"
+    #     connection = Connection.create_for_modification(
+    #         qualified_name=TestConnection.connection.qualified_name or "",
+    #         name=TestConnection.connection.name,
+    #     )
+    #     connection.description = description
+    #     response = upsert(connection)
+    #     verify_asset_updated(response, Connection)
+
+    # @pytest.mark.order(after="test_create")
+    # @pytest.mark.vcr()
+    # def test_trim_to_required(
+    #     self, client: AtlanClient, upsert: Callable[[Asset], AssetMutationResponse]
+    # ):
+    #     assert TestConnection.connection
+    #     connection = TestConnection.connection.trim_to_required()
+    #     response = upsert(connection)
+    #     assert response.mutated_entities is None