fix: distinguish deleted tag (keep sentinel) from never-existed (raise); consolidate tests

Refine BLDX-1530: get_id_for_name returns None for both a soft-deleted tag and a
name that never existed, but the cache tracks deleted names separately
(deleted_names). Use it so the retranslator only RAISES for a name that never
existed; a known-deleted tag (or the (DELETED) sentinel from a read round-trip)
keeps the sentinel — lossless, since the tag really existed. Applied to both the
sync and async retranslators.

Move the retranslator unit tests into tests/unit/test_atlan_tag_name.py (+ aio),
alongside the existing tag-serde tests, and drop the standalone test_retranslators
files. Adds the deleted-tag-keeps-sentinel case so both behaviours are covered.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Aryamanz29

pyatlan/model/aio/retranslators.py

@@ -98,8 +98,18 @@ async def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
                     if not raw_name:
                         continue
                     tag_name = str(raw_name)
-                    tag_id = await self.client.atlan_tag_cache.get_id_for_name(tag_name)
-                    if not tag_id and tag_name != DELETED_:
+                    cache = self.client.atlan_tag_cache
+                    tag_id = await cache.get_id_for_name(tag_name)
+                    # get_id_for_name returns None for BOTH a soft-deleted tag and
+                    # a name that never existed. Distinguish them: a known-deleted
+                    # tag (in deleted_names) — or the (DELETED) sentinel from a read
+                    # round-trip — keeps the sentinel (lossless; the tag really
+                    # existed), whereas a name that never existed is a client error.
+                    if (
+                        not tag_id
+                        and tag_name != DELETED_
+                        and tag_name not in cache.deleted_names
+                    ):
                         raise ErrorCode.ATLAN_TAG_NOT_FOUND_BY_NAME.exception_with_parameters(
                             tag_name
                         )

pyatlan/model/retranslators.py

@@ -105,8 +105,18 @@ def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
                     if not raw_name:
                         continue
                     tag_name = str(raw_name)
-                    tag_id = self.client.atlan_tag_cache.get_id_for_name(tag_name)
-                    if not tag_id and tag_name != DELETED_:
+                    cache = self.client.atlan_tag_cache
+                    tag_id = cache.get_id_for_name(tag_name)
+                    # get_id_for_name returns None for BOTH a soft-deleted tag and
+                    # a name that never existed. Distinguish them: a known-deleted
+                    # tag (in deleted_names) — or the (DELETED) sentinel from a read
+                    # round-trip — keeps the sentinel (lossless; the tag really
+                    # existed), whereas a name that never existed is a client error.
+                    if (
+                        not tag_id
+                        and tag_name != DELETED_
+                        and tag_name not in cache.deleted_names
+                    ):
                         raise ErrorCode.ATLAN_TAG_NOT_FOUND_BY_NAME.exception_with_parameters(
                             tag_name
                         )

tests/unit/aio/test_atlan_tag_name.py

@@ -1,11 +1,15 @@
 # SPDX-License-Identifier: Apache-2.0
 # Copyright 2025 Atlan Pte. Ltd.
+from unittest.mock import AsyncMock, MagicMock
+
 import pytest
 from pydantic.v1 import parse_obj_as
 
 import pyatlan.cache.aio.atlan_tag_cache
 from pyatlan.client.aio.client import AsyncAtlanClient
+from pyatlan.errors import NotFoundError
 from pyatlan.model.aio.core import AsyncAtlanRequest, AsyncAtlanResponse
+from pyatlan.model.aio.retranslators import AsyncAtlanTagRetranslator
 from pyatlan.model.assets import Purpose
 from pyatlan.model.constants import DELETED_
 from pyatlan.model.core import AtlanTagName
@@ -262,3 +266,46 @@ async def get_source_tags_attr_id(_, tag_id):
     _assert_asset_tags(
         purpose_without_translation_and_retranslation, is_retranslated=True
     )
+
+
+# --- AsyncAtlanTagRetranslator: tag name → ID resolution on the write path (BLDX-1530) ---
+def _tag_retranslator(name_to_id, deleted_names=None):
+    client = MagicMock()
+    client.atlan_tag_cache.get_id_for_name = AsyncMock(
+        side_effect=lambda name: name_to_id.get(name)
+    )
+    client.atlan_tag_cache.get_source_tags_attr_id = AsyncMock(return_value=None)
+    client.atlan_tag_cache.deleted_names = set(deleted_names or ())
+    return AsyncAtlanTagRetranslator(client)
+
+
+async def test_retranslate_resolves_existing_tag_name_to_id():
+    retranslator = _tag_retranslator({"PII": "abc123"})
+    out = await retranslator.retranslate({"classifications": [{"typeName": "PII"}]})
+    assert out["classifications"][0]["typeName"] == "abc123"
+
+
+async def test_retranslate_deleted_tag_keeps_sentinel_not_raises():
+    retranslator = _tag_retranslator({}, deleted_names={"WasDeleted"})
+    out = await retranslator.retranslate(
+        {"classifications": [{"typeName": "WasDeleted"}]}
+    )
+    assert out["classifications"][0]["typeName"] == DELETED_
+
+
+async def test_retranslate_missing_tag_raises_named_not_found():
+    retranslator = _tag_retranslator({})
+    with pytest.raises(NotFoundError) as exc:
+        await retranslator.retranslate(
+            {"addOrUpdateClassifications": [{"typeName": "Alert: DQ"}]}
+        )
+    message = str(exc.value)
+    assert "Alert: DQ" in message
+    assert "ATLAN-PYTHON-404-006" in message
+    assert DELETED_ not in message
+
+
+async def test_retranslate_deleted_sentinel_round_trip_is_preserved():
+    retranslator = _tag_retranslator({})
+    out = await retranslator.retranslate({"classifications": [{"typeName": DELETED_}]})
+    assert out["classifications"][0]["typeName"] == DELETED_

tests/unit/aio/test_retranslators.py

@@ -1,13 +1,17 @@
 # SPDX-License-Identifier: Apache-2.0
 # Copyright 2022 Atlan Pte. Ltd.
+from unittest.mock import MagicMock
+
 import pytest
 from pydantic.v1 import parse_obj_as
 
 import pyatlan.cache.atlan_tag_cache
 from pyatlan.client.atlan import AtlanClient
+from pyatlan.errors import NotFoundError
 from pyatlan.model.assets import Purpose
 from pyatlan.model.constants import DELETED_
 from pyatlan.model.core import AtlanRequest, AtlanResponse, AtlanTagName
+from pyatlan.model.retranslators import AtlanTagRetranslator
 
 ATLAN_TAG_ID = "yiB7RLvdC2yeryLPjaDeHM"
 
@@ -237,3 +241,68 @@ def get_source_tags_attr_id(_, tag_id):
     _assert_asset_tags(
         purpose_without_translation_and_retranslation, is_retranslated=True
     )
+
+
+# --- AtlanTagRetranslator: tag name → ID resolution on the write path (BLDX-1530) ---
+# Covers BOTH behaviours: an existing tag still resolves + the deleted sentinel
+# still round-trips (no regression), AND a genuinely unresolvable name now raises
+# a clear NotFoundError instead of shipping the (DELETED) sentinel to Atlas.
+def _tag_retranslator(name_to_id, deleted_names=None):
+    client = MagicMock()
+    client.atlan_tag_cache.get_id_for_name.side_effect = lambda name: name_to_id.get(
+        name
+    )
+    client.atlan_tag_cache.get_source_tags_attr_id.return_value = None
+    client.atlan_tag_cache.deleted_names = set(deleted_names or ())
+    return AtlanTagRetranslator(client)
+
+
+def test_retranslate_resolves_existing_tag_name_to_id():
+    # No regression: a real tag still maps name → hashed id.
+    retranslator = _tag_retranslator({"PII": "abc123"})
+    out = retranslator.retranslate({"classifications": [{"typeName": "PII"}]})
+    assert out["classifications"][0]["typeName"] == "abc123"
+
+
+def test_retranslate_deleted_tag_keeps_sentinel_not_raises():
+    # A KNOWN-deleted tag (tracked in the cache's deleted_names) referenced by its
+    # original name keeps the (DELETED) sentinel — the tag really existed and was
+    # deleted, so this is not the "never existed" client error.
+    retranslator = _tag_retranslator({}, deleted_names={"WasDeleted"})
+    out = retranslator.retranslate({"classifications": [{"typeName": "WasDeleted"}]})
+    assert out["classifications"][0]["typeName"] == DELETED_
+
+
+def test_retranslate_missing_tag_raises_named_not_found():
+    # New behaviour: add_atlan_tags(["Alert: DQ"]) on a tenant without that tag
+    # now fails fast with a clear, named client-side error (no sentinel leak).
+    retranslator = _tag_retranslator({})
+    with pytest.raises(NotFoundError) as exc:
+        retranslator.retranslate(
+            {"addOrUpdateClassifications": [{"typeName": "Alert: DQ"}]}
+        )
+    message = str(exc.value)
+    assert "Alert: DQ" in message
+    assert "ATLAN-PYTHON-404-006" in message
+    assert DELETED_ not in message
+
+
+def test_retranslate_remove_missing_tag_also_raises():
+    retranslator = _tag_retranslator({})
+    with pytest.raises(NotFoundError):
+        retranslator.retranslate({"removeClassifications": [{"typeName": "GhostTag"}]})
+
+
+def test_retranslate_deleted_sentinel_round_trip_is_preserved():
+    # No regression: an asset READ with an already-deleted tag surfaces (DELETED);
+    # re-serializing it stays lossless (no raise).
+    retranslator = _tag_retranslator({})
+    out = retranslator.retranslate({"classifications": [{"typeName": DELETED_}]})
+    assert out["classifications"][0]["typeName"] == DELETED_
+
+
+def test_retranslate_names_path_policy_serde_stays_tolerant():
+    # No regression: purpose/persona policy names round-trip through the sentinel.
+    retranslator = _tag_retranslator({})
+    out = retranslator.retranslate({"purposeClassifications": ["ghost-policy-tag"]})
+    assert out["purposeClassifications"] == [DELETED_]