Merge branch 'main' into CXD-321

Author: Aryamanz29

pyatlan/model/core.py

@@ -84,7 +84,7 @@ def _convert_to_display_text(cls, data):
         if display_text := AtlanTagCache.get_name_for_id(data):
             return AtlanTagName(display_text)
         else:
-            raise ValueError(f"{data} is not a valid AtlanTag")
+            return cls.get_deleted_sentinel()
 
     @staticmethod
     def json_encode_atlan_tag(atlan_tag_name: "AtlanTagName"):
@@ -237,11 +237,7 @@ def source_tag_attachements(self) -> List[SourceTagAttachment]:
     def type_name_is_tag_name(cls, value):
         if isinstance(value, AtlanTagName):
             return value
-        try:
-            value = AtlanTagName._convert_to_display_text(value)
-        except ValueError:
-            value = AtlanTagName.get_deleted_sentinel()
-        return value
+        return AtlanTagName._convert_to_display_text(value)
 
     def __init__(self, *args, **kwargs):
         from pyatlan.cache.atlan_tag_cache import AtlanTagCache

tests/unit/test_atlan_tag_name.py

@@ -1,8 +1,11 @@
 # SPDX-License-Identifier: Apache-2.0
 # Copyright 2022 Atlan Pte. Ltd.
 import pytest
+from pydantic.v1 import parse_obj_as
 
 import pyatlan.cache.atlan_tag_cache
+from pyatlan.model.assets import Purpose
+from pyatlan.model.constants import DELETED_
 from pyatlan.model.core import AtlanTagName
 
 ATLAN_TAG_ID = "yiB7RLvdC2yeryLPjaDeHM"
@@ -62,7 +65,7 @@ def test_convert_to_display_text_when_atlan_tag_passed_returns_same_atlan_tag(
     assert good_atlan_tag is AtlanTagName._convert_to_display_text(good_atlan_tag)
 
 
-def test_convert_to_display_text_when_bad_string_raises_value_error(monkeypatch):
+def test_convert_to_display_text_when_bad_string(monkeypatch):
     def get_name_for_id(_):
         return None
 
@@ -72,8 +75,10 @@ def get_name_for_id(_):
         get_name_for_id,
     )
 
-    with pytest.raises(ValueError, match="bad is not a valid AtlanTag"):
-        AtlanTagName._convert_to_display_text("bad")
+    assert (
+        AtlanTagName._convert_to_display_text("bad").__repr__()
+        == f"AtlanTagName('{DELETED_}')"
+    )
 
 
 def test_convert_to_display_text_when_id(monkeypatch):
@@ -102,3 +107,68 @@ def get_id_for_name(value):
 
 def test_json_encode_atlan_tag(monkeypatch, good_atlan_tag):
     assert AtlanTagName.json_encode_atlan_tag(good_atlan_tag) == ATLAN_TAG_ID
+
+
+def test_asset_tag_name_field_deserialization(monkeypatch):
+    def get_name_for_id(_):
+        return None
+
+    def get_id_for_name(_):
+        return None
+
+    monkeypatch.setattr(
+        pyatlan.cache.atlan_tag_cache.AtlanTagCache,
+        "get_id_for_name",
+        get_id_for_name,
+    )
+
+    monkeypatch.setattr(
+        pyatlan.cache.atlan_tag_cache.AtlanTagCache,
+        "get_name_for_id",
+        get_name_for_id,
+    )
+    # Simulate a `Purpose` asset with `purpose_atlan_tags` of type `AtlanTagName`
+    purpose_asset = {
+        "typeName": "Purpose",
+        "attributes": {
+            # AtlanTagName
+            "purposeClassifications": [
+                "some-deleted-purpose-tag-1",
+                "some-deleted-purpose-tag-2",
+            ],
+        },
+        "guid": "9f7a35f4-8d37-4273-81ec-c497a83a2472",
+        "status": "ACTIVE",
+        "classifications": [
+            # AtlanTag
+            {
+                "typeName": "some-deleted-purpose-tag-1",
+                "entityGuid": "82683fb9-1501-4627-a5d0-0da9be64c0d5",
+                "entityStatus": "DELETED",
+                "propagate": False,
+                "removePropagationsOnEntityDelete": True,
+                "restrictPropagationThroughLineage": True,
+                "restrictPropagationThroughHierarchy": False,
+            },
+            {
+                "typeName": "some-deleted-purpose-tag-2",
+                "entityGuid": "82683fb9-1501-4627-a5d0-0da9be64c0d5",
+                "entityStatus": "DELETED",
+                "propagate": False,
+                "removePropagationsOnEntityDelete": True,
+                "restrictPropagationThroughLineage": True,
+                "restrictPropagationThroughHierarchy": False,
+            },
+        ],
+    }
+    purpose = parse_obj_as(Purpose, purpose_asset)
+    assert purpose and isinstance(purpose, Purpose)
+
+    # Verify that deleted tags are correctly set to `None`
+    # assert purpose.atlan_tags == [AtlanTagName('(DELETED)')]
+    assert purpose.atlan_tags and len(purpose.atlan_tags) == 2
+    assert purpose.atlan_tags[0].type_name.__repr__() == f"AtlanTagName('{DELETED_}')"
+    assert purpose.atlan_tags[1].type_name.__repr__() == f"AtlanTagName('{DELETED_}')"
+    assert purpose.purpose_atlan_tags and len(purpose.purpose_atlan_tags) == 2
+    assert purpose.purpose_atlan_tags[0].__repr__() == f"AtlanTagName('{DELETED_}')"
+    assert purpose.purpose_atlan_tags[1].__repr__() == f"AtlanTagName('{DELETED_}')"