From bf6230752a896c70a8d6ca3d29a56e67a3447f21 Mon Sep 17 00:00:00 2001 From: Ernest Hill Date: Mon, 5 Jun 2023 09:26:50 +0300 Subject: [PATCH] Refactor to use CustomMetdataDict --- pyatlan/cache/custom_metadata_cache.py | 66 ++----- pyatlan/client/atlan.py | 75 ++++---- pyatlan/model/assets.py | 69 ++----- pyatlan/model/core.py | 14 -- pyatlan/model/custom_metadata.py | 61 +++++-- tests/integration/custom_metadata_test.py | 203 +++++++++------------ tests/unit/test_custom_metadata.py | 41 +++-- tests/unit/test_model.py | 209 +--------------------- 8 files changed, 226 insertions(+), 512 deletions(-) diff --git a/pyatlan/cache/custom_metadata_cache.py b/pyatlan/cache/custom_metadata_cache.py index c5a104fb8..b31416d56 100644 --- a/pyatlan/cache/custom_metadata_cache.py +++ b/pyatlan/cache/custom_metadata_cache.py @@ -1,11 +1,8 @@ # SPDX-License-Identifier: Apache-2.0 # Copyright 2022 Atlan Pte. Ltd. -import json -from typing import Any, Optional +from typing import Optional -from pyatlan.client.atlan import AtlanClient from pyatlan.error import InvalidRequestError, LogicError, NotFoundError -from pyatlan.model.core import CustomMetadata from pyatlan.model.enums import AtlanTypeCategory from pyatlan.model.typedef import AttributeDef, CustomMetadataDef @@ -35,7 +32,7 @@ class CustomMetadataCache: @classmethod def refresh_cache(cls) -> None: - from pyatlan.model.core import CustomMetadata, to_snake_case + from pyatlan.client.atlan import AtlanClient client = AtlanClient.get_default_client() if client is None: @@ -56,23 +53,10 @@ def refresh_cache(cls) -> None: cls.map_name_to_id[type_name] = type_id cls.map_attr_id_to_name[type_id] = {} cls.map_attr_name_to_id[type_id] = {} - meta_name = cm.display_name.replace(" ", "") - attribute_class_name = f"Attributes_{meta_name}" - attrib_type = type(attribute_class_name, (CustomMetadata,), {}) - attrib_type._meta_data_type_id = type_id # type: ignore - attrib_type._meta_data_type_name = type_name # type: ignore - cls.map_id_to_type[type_id] = attrib_type - applicable_types: set[str] = set() if cm.attribute_defs: for attr in cm.attribute_defs: - if attr.options and attr.options.custom_applicable_entity_types: - applicable_types.update( - json.loads(attr.options.custom_applicable_entity_types) - ) attr_id = str(attr.name) attr_name = str(attr.display_name) - # Use a renamed attribute everywhere - attr_renamed = to_snake_case(attr_name) cls.map_attr_id_to_name[type_id][attr_id] = attr_name if attr.options and attr.options.is_archived: cls.archived_attr_ids[attr_id] = attr_name @@ -83,12 +67,7 @@ def refresh_cache(cls) -> None: code="ATLAN-PYTHON-500-100", ) else: - setattr(attrib_type, attr_renamed, Synonym(attr_id)) cls.map_attr_name_to_id[type_id][attr_name] = attr_id - for asset_type in applicable_types: - if asset_type not in cls.types_by_asset: - cls.types_by_asset[asset_type] = set() - cls.types_by_asset[asset_type].add(attrib_type) @classmethod def get_id_for_name(cls, name: str) -> str: @@ -202,18 +181,22 @@ def get_attr_id_for_name(cls, set_name: str, attr_name: str) -> str: ) @classmethod - def get_attr_name_for_id(cls, set_id: str, attr_id: str) -> Optional[str]: + def get_attr_name_for_id(cls, set_id: str, attr_id: str) -> str: """ - Translate the provided human-readable custom metadata set and attribute names to the Atlan-internal ID string - for the attribute. + Given the Atlan-internal ID stringfor the set and the Atlan-internal ID for the attribute return the + human-readable custom metadata name for the attribute. """ if sub_map := cls.map_attr_id_to_name.get(set_id): if attr_name := sub_map.get(attr_id): return attr_name cls.refresh_cache() if sub_map := cls.map_attr_id_to_name.get(set_id): - return sub_map.get(attr_id) - return None + if attr_name := sub_map.get(attr_id): + return attr_name + raise NotFoundError( + message=f"Custom metadata property with ID {attr_id} does not exist in the custom metadata {set_id}.", + code="ATLAN-PYTHON-404-009", + ) @classmethod def _get_attributes_for_search_results(cls, set_id: str) -> Optional[list[str]]: @@ -234,33 +217,6 @@ def get_attributes_for_search_results(cls, set_name: str) -> Optional[list[str]] return cls._get_attributes_for_search_results(set_id) return None - @classmethod - def get_custom_metadata( - cls, - name: str, - asset_type: type, - business_attributes: Optional[dict[str, Any]] = None, - ) -> CustomMetadata: - type_name = asset_type.__name__ - ba_id = cls.get_id_for_name(name) - if ba_id is None: - raise ValueError(f"No custom metadata with the name: {name} exist") - for a_type in CustomMetadataCache.types_by_asset[type_name]: - if ( - hasattr(a_type, "_meta_data_type_name") - and a_type._meta_data_type_name == name - ): - break - else: - raise ValueError(f"Custom metadata {name} is not applicable to {type_name}") - if ba_type := CustomMetadataCache.get_type_for_id(ba_id): - return ( - ba_type(business_attributes[ba_id]) - if business_attributes and ba_id in business_attributes - else ba_type() - ) - raise ValueError(f"Custom metadata {name} is not applicable to {type_name}") - @classmethod def get_custom_metadata_def(cls, name: str) -> CustomMetadataDef: """ diff --git a/pyatlan/client/atlan.py b/pyatlan/client/atlan.py index 0f9796097..b723b8216 100644 --- a/pyatlan/client/atlan.py +++ b/pyatlan/client/atlan.py @@ -76,9 +76,8 @@ Classification, ClassificationName, Classifications, - CustomMetadata, - CustomMetadataReqest, ) +from pyatlan.model.custom_metadata import CustomMetadataDict, CustomMetadataRequest from pyatlan.model.enums import ( AtlanConnectorType, AtlanDeleteType, @@ -1015,60 +1014,56 @@ def remove_announcement( return self._update_asset_by_attribute(asset, asset_type, qualified_name) def update_custom_metadata_attributes( - self, guid: str, custom_metadata: CustomMetadata + self, guid: str, custom_metadata: CustomMetadataDict ): - custom_metadata_request = CustomMetadataReqest(__root__=custom_metadata) + custom_metadata_request = CustomMetadataRequest.create( + custom_metadata_dict=custom_metadata + ) self._call_api( ADD_BUSINESS_ATTRIBUTE_BY_ID.format_path( - {"entity_guid": guid, "bm_id": custom_metadata._meta_data_type_id} + { + "entity_guid": guid, + "bm_id": custom_metadata_request.custom_metadata_set_id, + } ), None, custom_metadata_request, ) - def replace_custom_metadata(self, guid: str, custom_metadata: CustomMetadata): - from pyatlan.cache.custom_metadata_cache import CustomMetadataCache - - # Iterate through the custom metadata provided and explicitly set every - # single attribute, so that they are all serialized out (forcing removal - # of any empty ones) - for k, v in custom_metadata.items(): - # Need to translate the hashed-string key here back to an attribute name - attr_name = str( - CustomMetadataCache.get_attr_name_for_id( - set_id=custom_metadata._meta_data_type_id, attr_id=k - ) - ) - if not v: - setattr(custom_metadata, attr_name, None) - else: - setattr(custom_metadata, attr_name, v) - custom_metadata_request = CustomMetadataReqest(__root__=custom_metadata) + def replace_custom_metadata(self, guid: str, custom_metadata: CustomMetadataDict): + # clear unset attributes so that they are removed + custom_metadata.clear_unset() + custom_metadata_request = CustomMetadataRequest.create( + custom_metadata_dict=custom_metadata + ) self._call_api( ADD_BUSINESS_ATTRIBUTE_BY_ID.format_path( - {"entity_guid": guid, "bm_id": custom_metadata._meta_data_type_id} + { + "entity_guid": guid, + "bm_id": custom_metadata_request.custom_metadata_set_id, + } ), None, custom_metadata_request, ) def remove_custom_metadata(self, guid: str, cm_name: str): - from pyatlan.cache.custom_metadata_cache import CustomMetadataCache - - # Ensure the custom metadata exists first - let this throw an error if not - if cm_id := CustomMetadataCache.get_id_for_name(cm_name): - # Initialize a dict of empty attributes for the custom metadata, and then - # send that so that they are removed accordingly - if cm_type := CustomMetadataCache.get_type_for_id(cm_id): - custom_metadata = cm_type() - custom_metadata_request = CustomMetadataReqest(__root__=custom_metadata) - self._call_api( - ADD_BUSINESS_ATTRIBUTE_BY_ID.format_path( - {"entity_guid": guid, "bm_id": cm_id} - ), - None, - custom_metadata_request, - ) + custom_metadata = CustomMetadataDict(name=cm_name) + # invoke clear_all so all attributes are set to None and consequently removed + custom_metadata.clear_all() + custom_metadata_request = CustomMetadataRequest.create( + custom_metadata_dict=custom_metadata + ) + self._call_api( + ADD_BUSINESS_ATTRIBUTE_BY_ID.format_path( + { + "entity_guid": guid, + "bm_id": custom_metadata_request.custom_metadata_set_id, + } + ), + None, + custom_metadata_request, + ) @validate_arguments() def append_terms( diff --git a/pyatlan/model/assets.py b/pyatlan/model/assets.py index 94fcea059..a3972cf6f 100644 --- a/pyatlan/model/assets.py +++ b/pyatlan/model/assets.py @@ -10,15 +10,10 @@ from typing import Any, ClassVar, Dict, List, Optional, TypeVar from urllib.parse import quote, unquote -from pydantic import Field, StrictStr, root_validator, validator - -from pyatlan.model.core import ( - Announcement, - AtlanObject, - Classification, - CustomMetadata, - Meaning, -) +from pydantic import Field, PrivateAttr, StrictStr, root_validator, validator + +from pyatlan.model.core import Announcement, AtlanObject, Classification, Meaning +from pyatlan.model.custom_metadata import CustomMetadataDict, CustomMetadataProxy from pyatlan.model.enums import ( ADLSAccessTier, ADLSAccountStatus, @@ -91,12 +86,19 @@ class Referenceable(AtlanObject): def __init__(__pydantic_self__, **data: Any) -> None: super().__init__(**data) __pydantic_self__.__fields_set__.update(["attributes", "type_name"]) + __pydantic_self__._metadata_proxy = CustomMetadataProxy( + __pydantic_self__.business_attributes + ) def __setattr__(self, name, value): if name in Referenceable._convience_properties: return object.__setattr__(self, name, value) super().__setattr__(name, value) + def json(self, *args, **kwargs) -> str: + self.business_attributes = self._metadata_proxy.business_attributes + return super().json(**kwargs) + _convience_properties: ClassVar[list[str]] = [ "qualified_name", "replicated_from", @@ -159,6 +161,7 @@ class Attributes(AtlanObject): def validate_required(self): pass + _metadata_proxy: CustomMetadataProxy = PrivateAttr() attributes: "Referenceable.Attributes" = Field( default_factory=lambda: Referenceable.Attributes(), description="Map of attributes in the instance and their values. The specific keys of this map will vary " @@ -259,51 +262,11 @@ def validate_required(self): if not self.create_time or self.created_by: self.attributes.validate_required() - def get_custom_metadata(self, name: str) -> CustomMetadata: - from pyatlan.cache.custom_metadata_cache import CustomMetadataCache - - ba_id = CustomMetadataCache.get_id_for_name(name) - if ba_id is None: - raise ValueError(f"No custom metadata with the name: {name} exist") - for a_type in CustomMetadataCache.types_by_asset[self.type_name]: - if ( - hasattr(a_type, "_meta_data_type_name") - and a_type._meta_data_type_name == name - ): - break - else: - raise ValueError( - f"Custom metadata attributes {name} are not applicable to {self.type_name}" - ) - if ba_type := CustomMetadataCache.get_type_for_id(ba_id): - return ( - ba_type(self.business_attributes[ba_id]) - if self.business_attributes and ba_id in self.business_attributes - else ba_type() - ) - else: - raise ValueError( - f"Custom metadata attributes {name} are not applicable to {self.type_name}" - ) - - def set_custom_metadata(self, custom_metadata: CustomMetadata) -> None: - from pyatlan.cache.custom_metadata_cache import CustomMetadataCache + def get_custom_metadata(self, name: str) -> CustomMetadataDict: + return self._metadata_proxy.get_custom_metadata(name=name) - if not isinstance(custom_metadata, CustomMetadata): - raise ValueError( - "business_attributes must be an instance of CustomMetadata" - ) - if ( - type(custom_metadata) - not in CustomMetadataCache.types_by_asset[self.type_name] - ): - raise ValueError( - f"Business attributes {custom_metadata._meta_data_type_name} are not applicable to {self.type_name}" - ) - ba_dict = dict(custom_metadata) - if not self.business_attributes: - self.business_attributes = {} - self.business_attributes[custom_metadata._meta_data_type_id] = ba_dict + def set_custom_metadata(self, custom_metadata: CustomMetadataDict): + return self._metadata_proxy.set_custom_metadata(custom_metadata=custom_metadata) class Asset(Referenceable): diff --git a/pyatlan/model/core.py b/pyatlan/model/core.py index 91938765d..2528fe6ba 100644 --- a/pyatlan/model/core.py +++ b/pyatlan/model/core.py @@ -187,17 +187,3 @@ class AssetRequest(AtlanObject, GenericModel, Generic[T]): class BulkRequest(AtlanObject, GenericModel, Generic[T]): entities: list[T] - - -class CustomMetadata(dict): - _meta_data_type_name = "" - _meta_data_type_id = "" - - def __setattr__(self, key, value): - if not hasattr(self, key): - raise AttributeError(f"Attribute {key} does not exist") - super().__setattr__(key, value) - - -class CustomMetadataReqest(AtlanObject): - __root__: CustomMetadata diff --git a/pyatlan/model/custom_metadata.py b/pyatlan/model/custom_metadata.py index 06ee984f1..03486b618 100644 --- a/pyatlan/model/custom_metadata.py +++ b/pyatlan/model/custom_metadata.py @@ -1,52 +1,76 @@ from collections import UserDict from typing import Any, Optional +from pydantic import PrivateAttr + from pyatlan.cache.custom_metadata_cache import CustomMetadataCache from pyatlan.model.core import AtlanObject class CustomMetadataDict(UserDict): + """This class allows the manipulation of a set of custom metadata attributes using the human readable names.""" + @property def attribute_names(self) -> set[str]: return self._names def __init__(self, name: str): + """Inits CustomMetadataDict with a string containing the human readable name of a set of custom metadata""" super().__init__() self._name = name self._modified = False id = CustomMetadataCache.get_id_for_name(name) self._names = set(CustomMetadataCache.map_attr_id_to_name[id].values()) - pass @property def modified(self): + """Returns a boolean indicating whether the set has been modified from its initial values""" return self._modified - def __setitem__(self, key, value): + def __setitem__(self, key: str, value): + """Set the value of a property of the custom metadata set using the human readable name as the key. + The name will be validated to ensure that it's valid for this custom metadata set""" if key not in self._names: raise KeyError(f"'{key}' is not a valid property name for {self._name}") self._modified = True self.data[key] = value - def __getitem__(self, key): + def __getitem__(self, key: str): + """Retrieve the value of a property of the custom metadata set using the human readable name as the key. + The name will be validated to ensure that it's valid for this custom metadata set""" if key not in self._names: raise KeyError(f"'{key}' is not a valid property name for {self._name}") if key not in self.data: - raise KeyError(f"'{key}' must be set before trying to retrieve the value") + return None return self.data[key] - def clear(self): + def clear_all(self): + """This method will set all the properties available explicitly to None""" for attribute_name in self._names: self.data[attribute_name] = None self._modified = True + def clear_unset(self): + """This method will set all properties that haven't been set to None""" + for name in self.attribute_names: + if name not in self.data: + self.data[name] = None + + def is_set(self, key: str): + """Returns a boolean indicating whether the given property has been set in the metadata set. The key + will be validated to ensure that it's a valid property name for this metadata set""" + if key not in self._names: + raise KeyError(f"'{key}' is not a valid property name for {self._name}") + return key in self.data + @property - def business_attributes(self) -> dict[str, dict[str, Any]]: - values = { + def business_attributes(self) -> dict[str, Any]: + """Returns a dict containing the metadat set with the human readable set name and property names resolved + to their internal values""" + return { CustomMetadataCache.get_attr_id_for_name(self._name, key): value for (key, value) in self.data.items() } - return {CustomMetadataCache.get_id_for_name(self._name): values} class CustomMetadataProxy: @@ -94,16 +118,25 @@ def modified(self) -> bool: @property def business_attributes(self) -> Optional[dict[str, Any]]: if self.modified and self._metadata is not None: - new_metadata = {} - for dict in self._metadata.values(): - new_metadata.update(dict.business_attributes) - self._business_attributes = new_metadata + return { + CustomMetadataCache.get_id_for_name(key): value.business_attributes + for key, value in self._metadata.items() + } return self._business_attributes class CustomMetadataRequest(AtlanObject): - __root__: dict[str, dict[str, Any]] + __root__: dict[str, Any] + _set_id: str = PrivateAttr() @classmethod def create(cls, custom_metadata_dict: CustomMetadataDict): - return cls(__root__=custom_metadata_dict.business_attributes) + ret_val = cls(__root__=custom_metadata_dict.business_attributes) + ret_val._set_id = CustomMetadataCache.get_id_for_name( + custom_metadata_dict._name + ) + return ret_val + + @property + def custom_metadata_set_id(self): + return self._set_id diff --git a/tests/integration/custom_metadata_test.py b/tests/integration/custom_metadata_test.py index 912d7aac0..2049bc807 100644 --- a/tests/integration/custom_metadata_test.py +++ b/tests/integration/custom_metadata_test.py @@ -8,15 +8,8 @@ from pyatlan.cache.custom_metadata_cache import CustomMetadataCache from pyatlan.client.atlan import AtlanClient -from pyatlan.error import NotFoundError -from pyatlan.model.assets import ( - AtlasGlossary, - AtlasGlossaryTerm, - Badge, - BadgeCondition, - Table, -) -from pyatlan.model.core import CustomMetadata, to_snake_case +from pyatlan.model.assets import AtlasGlossary, AtlasGlossaryTerm, Badge, BadgeCondition +from pyatlan.model.custom_metadata import CustomMetadataDict from pyatlan.model.enums import ( AtlanCustomAttributePrimitiveType, AtlanTypeCategory, @@ -41,11 +34,6 @@ CM_ATTR_RACI_INFORMED = "Informed" CM_ATTR_RACI_EXTRA = "Extra" -CM_ATTR_RACI_RESPONSIBLE_RENAMED = to_snake_case(CM_ATTR_RACI_RESPONSIBLE) -CM_ATTR_RACI_ACCOUNTABLE_RENAMED = to_snake_case(CM_ATTR_RACI_ACCOUNTABLE) -CM_ATTR_RACI_CONSULTED_RENAMED = to_snake_case(CM_ATTR_RACI_CONSULTED) -CM_ATTR_RACI_INFORMED_RENAMED = to_snake_case(CM_ATTR_RACI_INFORMED) -CM_ATTR_RACI_EXTRA_RENAMED = to_snake_case(CM_ATTR_RACI_EXTRA) CM_ATTR_IPR_LICENSE = "License" CM_ATTR_IPR_VERSION = "Version" @@ -53,12 +41,6 @@ CM_ATTR_IPR_DATE = "Date" CM_ATTR_IPR_URL = "URL" -CM_ATTR_IPR_LICENSE_RENAMED = to_snake_case(CM_ATTR_IPR_LICENSE) -CM_ATTR_IPR_VERSION_RENAMED = to_snake_case(CM_ATTR_IPR_VERSION) -CM_ATTR_IPR_MANDATORY_RENAMED = to_snake_case(CM_ATTR_IPR_MANDATORY) -CM_ATTR_IPR_DATE_RENAMED = to_snake_case(CM_ATTR_IPR_DATE) -CM_ATTR_IPR_URL_RENAMED = to_snake_case(CM_ATTR_IPR_URL) - CM_ATTR_QUALITY_COUNT = "Count" CM_ATTR_QUALITY_SQL = "SQL" CM_ATTR_QUALITY_TYPE = "Type" @@ -71,10 +53,6 @@ "Uniqueness", ] -CM_ATTR_QUALITY_COUNT_RENAMED = to_snake_case(CM_ATTR_QUALITY_COUNT) -CM_ATTR_QUALITY_SQL_RENAMED = to_snake_case(CM_ATTR_QUALITY_SQL) -CM_ATTR_QUALITY_TYPE_RENAMED = to_snake_case(CM_ATTR_QUALITY_TYPE) - _removal_epoch: Optional[int] @@ -429,10 +407,10 @@ def test_add_term_cm_raci( raci_attrs = term.get_custom_metadata(cm_name) _validate_raci_empty(raci_attrs) group1, group2 = _get_groups(client, make_unique) - setattr(raci_attrs, CM_ATTR_RACI_RESPONSIBLE_RENAMED, [FIXED_USER]) - setattr(raci_attrs, CM_ATTR_RACI_ACCOUNTABLE_RENAMED, FIXED_USER) - setattr(raci_attrs, CM_ATTR_RACI_CONSULTED_RENAMED, [group1.name]) - setattr(raci_attrs, CM_ATTR_RACI_INFORMED_RENAMED, [group1.name, group2.name]) + raci_attrs[CM_ATTR_RACI_RESPONSIBLE] = [FIXED_USER] + raci_attrs[CM_ATTR_RACI_ACCOUNTABLE] = FIXED_USER + raci_attrs[CM_ATTR_RACI_CONSULTED] = [group1.name] + raci_attrs[CM_ATTR_RACI_INFORMED] = [group1.name, group2.name] client.update_custom_metadata_attributes(term.guid, raci_attrs) t = client.retrieve_minimal(guid=term.guid, asset_type=AtlasGlossaryTerm) assert t @@ -448,15 +426,12 @@ def test_add_term_cm_ipr( cm_name = make_unique("IPR") ipr_attrs = term.get_custom_metadata(cm_name) _validate_ipr_empty(ipr_attrs) - setattr(ipr_attrs, CM_ATTR_IPR_LICENSE_RENAMED, "CC BY") - setattr(ipr_attrs, CM_ATTR_IPR_VERSION_RENAMED, 2.0) - setattr(ipr_attrs, CM_ATTR_IPR_MANDATORY_RENAMED, True) - setattr(ipr_attrs, CM_ATTR_IPR_DATE_RENAMED, 1659308400000) - setattr( - ipr_attrs, - CM_ATTR_IPR_URL_RENAMED, - "https://creativecommons.org/licenses/by/2.0/", - ) + ipr_attrs[CM_ATTR_IPR_LICENSE] = "CC BY" + ipr_attrs[CM_ATTR_IPR_VERSION] = 2.0 + ipr_attrs[CM_ATTR_IPR_MANDATORY] = True + ipr_attrs[CM_ATTR_IPR_DATE] = 1659308400000 + ipr_attrs[CM_ATTR_IPR_URL] = "https://creativecommons.org/licenses/by/2.0/" + client.update_custom_metadata_attributes(term.guid, ipr_attrs) t = client.retrieve_minimal(guid=term.guid, asset_type=AtlasGlossaryTerm) assert t @@ -472,9 +447,9 @@ def test_add_term_cm_dq( cm_name = make_unique("DQ") dq_attrs = term.get_custom_metadata(cm_name) _validate_dq_empty(dq_attrs) - setattr(dq_attrs, CM_ATTR_QUALITY_COUNT_RENAMED, 42) - setattr(dq_attrs, CM_ATTR_QUALITY_SQL_RENAMED, "SELECT * from SOMEWHERE;") - setattr(dq_attrs, CM_ATTR_QUALITY_TYPE_RENAMED, "Completeness") + dq_attrs[CM_ATTR_QUALITY_COUNT] = 42 + dq_attrs[CM_ATTR_QUALITY_SQL] = "SELECT * from SOMEWHERE;" + dq_attrs[CM_ATTR_QUALITY_TYPE] = "Completeness" client.update_custom_metadata_attributes(term.guid, dq_attrs) t = client.retrieve_minimal(guid=term.guid, asset_type=AtlasGlossaryTerm) assert t @@ -491,7 +466,7 @@ def test_update_term_cm_ipr( cm_name = make_unique("IPR") ipr = term.get_custom_metadata(cm_name) # Note: MUST access the getter / setter, not the underlying store - setattr(ipr, CM_ATTR_IPR_MANDATORY_RENAMED, False) + ipr[CM_ATTR_IPR_MANDATORY] = False client.update_custom_metadata_attributes(term.guid, ipr) t = client.retrieve_minimal(guid=term.guid, asset_type=AtlasGlossaryTerm) assert t @@ -514,11 +489,10 @@ def test_replace_term_cm_raci( CM_QUALITY = make_unique("DQ") raci = term.get_custom_metadata(CM_RACI) group1, group2 = _get_groups(client, make_unique) - # Note: MUST access the getter / setter, not the underlying store - setattr(raci, CM_ATTR_RACI_RESPONSIBLE_RENAMED, [FIXED_USER]) - setattr(raci, CM_ATTR_RACI_ACCOUNTABLE_RENAMED, FIXED_USER) - setattr(raci, CM_ATTR_RACI_CONSULTED_RENAMED, None) - setattr(raci, CM_ATTR_RACI_INFORMED_RENAMED, [group1.name, group2.name]) + raci[CM_ATTR_RACI_RESPONSIBLE] = [FIXED_USER] + raci[CM_ATTR_RACI_ACCOUNTABLE] = FIXED_USER + raci[CM_ATTR_RACI_CONSULTED] = None + raci[CM_ATTR_RACI_INFORMED] = [group1.name, group2.name] client.replace_custom_metadata(term.guid, raci) t = client.retrieve_minimal(guid=term.guid, asset_type=AtlasGlossaryTerm) assert t @@ -539,7 +513,8 @@ def test_replace_term_cm_ipr( CM_RACI = make_unique("RACI") CM_IPR = make_unique("IPR") CM_QUALITY = make_unique("DQ") - client.replace_custom_metadata(term.guid, term.get_custom_metadata(CM_IPR)) + term_cm_ipr = term.get_custom_metadata(CM_IPR) + client.replace_custom_metadata(term.guid, term_cm_ipr) t = client.retrieve_minimal(guid=term.guid, asset_type=AtlasGlossaryTerm) assert t _validate_raci_attributes_replacement( @@ -560,7 +535,7 @@ def test_search_by_any_accountable( be_active = Term.with_state("ACTIVE") be_a_term = Term.with_type_name("AtlasGlossaryTerm") have_attr = Exists.with_custom_metadata( - set_name=make_unique("RACI"), attr_name=CM_ATTR_RACI_ACCOUNTABLE_RENAMED + set_name=make_unique("RACI"), attr_name=CM_ATTR_RACI_ACCOUNTABLE ) query = Bool(must=[be_active, be_a_term, have_attr]) dsl = DSL(query=query) @@ -597,7 +572,7 @@ def test_search_by_specific_accountable( be_a_term = Term.with_type_name("AtlasGlossaryTerm") have_attr = Term.with_custom_metadata( set_name=make_unique("RACI"), - attr_name=CM_ATTR_RACI_ACCOUNTABLE_RENAMED, + attr_name=CM_ATTR_RACI_ACCOUNTABLE, value=FIXED_USER, ) query = Bool(must=[be_active, be_a_term, have_attr]) @@ -827,11 +802,11 @@ def test_update_replacing_cm( CM_QUALITY = make_unique("DQ") raci = term.get_custom_metadata(CM_RACI) group1, group2 = _get_groups(client, make_unique) - setattr(raci, CM_ATTR_RACI_RESPONSIBLE_RENAMED, [FIXED_USER]) - setattr(raci, CM_ATTR_RACI_ACCOUNTABLE_RENAMED, FIXED_USER) - setattr(raci, CM_ATTR_RACI_CONSULTED_RENAMED, [group1.name]) - setattr(raci, CM_ATTR_RACI_INFORMED_RENAMED, [group1.name, group2.name]) - setattr(raci, CM_ATTR_RACI_EXTRA_RENAMED, "something extra...") + raci[CM_ATTR_RACI_RESPONSIBLE] = [FIXED_USER] + raci[CM_ATTR_RACI_ACCOUNTABLE] = FIXED_USER + raci[CM_ATTR_RACI_CONSULTED] = [group1.name] + raci[CM_ATTR_RACI_INFORMED] = [group1.name, group2.name] + raci[CM_ATTR_RACI_EXTRA] = "something extra..." to_update = AtlasGlossaryTerm.create_for_modification( qualified_name=term.qualified_name, name=term.name, glossary_guid=glossary.guid ) @@ -853,7 +828,7 @@ def test_update_replacing_cm( assert x.qualified_name == term.qualified_name raci = x.get_custom_metadata(CM_RACI) _validate_raci_attributes(client, make_unique, raci) - assert getattr(raci, CM_ATTR_RACI_EXTRA_RENAMED) == "something extra..." + assert raci[CM_ATTR_RACI_EXTRA] == "something extra..." _validate_ipr_empty(x.get_custom_metadata(CM_IPR)) _validate_dq_empty(x.get_custom_metadata(CM_QUALITY)) @@ -861,22 +836,15 @@ def test_update_replacing_cm( # TODO: test entity audit retrieval and parsing, once available -def test_get_custom_metadata_when_name_is_invalid_then_raises_not_found_error(): - with pytest.raises( - NotFoundError, match="Custom metadata with name Bogs does not exist" - ): - CustomMetadataCache.get_custom_metadata(name="Bogs", asset_type=Table) - - def _validate_raci_attributes( - client: AtlanClient, make_unique: Callable[[str], str], cma: CustomMetadata + client: AtlanClient, make_unique: Callable[[str], str], cma: CustomMetadataDict ): assert cma # Note: MUST access the getter / setter, not the underlying store - responsible = getattr(cma, CM_ATTR_RACI_RESPONSIBLE_RENAMED) - accountable = getattr(cma, CM_ATTR_RACI_ACCOUNTABLE_RENAMED) - consulted = getattr(cma, CM_ATTR_RACI_CONSULTED_RENAMED) - informed = getattr(cma, CM_ATTR_RACI_INFORMED_RENAMED) + responsible = cma[CM_ATTR_RACI_RESPONSIBLE] + accountable = cma[CM_ATTR_RACI_ACCOUNTABLE] + consulted = cma[CM_ATTR_RACI_CONSULTED] + informed = cma[CM_ATTR_RACI_INFORMED] group1, group2 = _get_groups(client, make_unique) assert responsible assert len(responsible) == 1 @@ -888,14 +856,14 @@ def _validate_raci_attributes( def _validate_raci_attributes_replacement( - client: AtlanClient, make_unique: Callable[[str], str], cma: CustomMetadata + client: AtlanClient, make_unique: Callable[[str], str], cma: CustomMetadataDict ): assert cma # Note: MUST access the getter / setter, not the underlying store - responsible = getattr(cma, CM_ATTR_RACI_RESPONSIBLE_RENAMED) - accountable = getattr(cma, CM_ATTR_RACI_ACCOUNTABLE_RENAMED) - consulted = getattr(cma, CM_ATTR_RACI_CONSULTED_RENAMED) - informed = getattr(cma, CM_ATTR_RACI_INFORMED_RENAMED) + responsible = cma[CM_ATTR_RACI_RESPONSIBLE] + accountable = cma[CM_ATTR_RACI_ACCOUNTABLE] + consulted = cma[CM_ATTR_RACI_CONSULTED] + informed = cma[CM_ATTR_RACI_INFORMED] group1, group2 = _get_groups(client, make_unique) assert responsible assert responsible == [FIXED_USER] @@ -905,30 +873,27 @@ def _validate_raci_attributes_replacement( assert informed == [group1.name, group2.name] -def _validate_raci_empty(raci_attrs: CustomMetadata): - assert hasattr(raci_attrs, CM_ATTR_RACI_RESPONSIBLE_RENAMED) - assert hasattr(raci_attrs, CM_ATTR_RACI_ACCOUNTABLE_RENAMED) - assert hasattr(raci_attrs, CM_ATTR_RACI_CONSULTED_RENAMED) - assert hasattr(raci_attrs, CM_ATTR_RACI_INFORMED_RENAMED) - assert hasattr(raci_attrs, CM_ATTR_RACI_EXTRA_RENAMED) - assert not getattr( - raci_attrs, CM_ATTR_RACI_RESPONSIBLE_RENAMED - ) # could be empty list - assert getattr(raci_attrs, CM_ATTR_RACI_ACCOUNTABLE_RENAMED) is None - assert not getattr( - raci_attrs, CM_ATTR_RACI_CONSULTED_RENAMED - ) # could be empty list - assert not getattr(raci_attrs, CM_ATTR_RACI_INFORMED_RENAMED) # could be empty list - assert getattr(raci_attrs, CM_ATTR_RACI_EXTRA_RENAMED) is None - - -def _validate_ipr_attributes(cma: CustomMetadata, mandatory: bool = True): +def _validate_raci_empty(raci_attrs: CustomMetadataDict): + attribute_names = raci_attrs.attribute_names + assert CM_ATTR_RACI_RESPONSIBLE in attribute_names + assert CM_ATTR_RACI_ACCOUNTABLE in attribute_names + assert CM_ATTR_RACI_CONSULTED in attribute_names + assert CM_ATTR_RACI_INFORMED in attribute_names + assert CM_ATTR_RACI_EXTRA in attribute_names + assert not raci_attrs[CM_ATTR_RACI_RESPONSIBLE] + assert raci_attrs[CM_ATTR_RACI_ACCOUNTABLE] is None + assert not raci_attrs[CM_ATTR_RACI_CONSULTED] # could be empty list + assert not raci_attrs[CM_ATTR_RACI_INFORMED] # could be empty list + assert raci_attrs[CM_ATTR_RACI_EXTRA] is None + + +def _validate_ipr_attributes(cma: CustomMetadataDict, mandatory: bool = True): assert cma - license = getattr(cma, CM_ATTR_IPR_LICENSE_RENAMED) - v = getattr(cma, CM_ATTR_IPR_VERSION_RENAMED) - m = getattr(cma, CM_ATTR_IPR_MANDATORY_RENAMED) - d = getattr(cma, CM_ATTR_IPR_DATE_RENAMED) - u = getattr(cma, CM_ATTR_IPR_URL_RENAMED) + license = cma[CM_ATTR_IPR_LICENSE] + v = cma[CM_ATTR_IPR_VERSION] + m = cma[CM_ATTR_IPR_MANDATORY] + d = cma[CM_ATTR_IPR_DATE] + u = cma[CM_ATTR_IPR_URL] assert license assert license == "CC BY" assert v @@ -943,24 +908,25 @@ def _validate_ipr_attributes(cma: CustomMetadata, mandatory: bool = True): assert u == "https://creativecommons.org/licenses/by/2.0/" -def _validate_ipr_empty(ipr_attrs: CustomMetadata): - assert hasattr(ipr_attrs, CM_ATTR_IPR_LICENSE_RENAMED) - assert hasattr(ipr_attrs, CM_ATTR_IPR_VERSION_RENAMED) - assert hasattr(ipr_attrs, CM_ATTR_IPR_MANDATORY_RENAMED) - assert hasattr(ipr_attrs, CM_ATTR_IPR_DATE_RENAMED) - assert hasattr(ipr_attrs, CM_ATTR_IPR_URL_RENAMED) - assert getattr(ipr_attrs, CM_ATTR_IPR_LICENSE_RENAMED) is None - assert getattr(ipr_attrs, CM_ATTR_IPR_VERSION_RENAMED) is None - assert getattr(ipr_attrs, CM_ATTR_IPR_MANDATORY_RENAMED) is None - assert getattr(ipr_attrs, CM_ATTR_IPR_DATE_RENAMED) is None - assert getattr(ipr_attrs, CM_ATTR_IPR_URL_RENAMED) is None +def _validate_ipr_empty(ipr_attrs: CustomMetadataDict): + attribute_names = ipr_attrs.attribute_names + assert CM_ATTR_IPR_LICENSE in attribute_names + assert CM_ATTR_IPR_VERSION in attribute_names + assert CM_ATTR_IPR_MANDATORY in attribute_names + assert CM_ATTR_IPR_DATE in attribute_names + assert CM_ATTR_IPR_URL in attribute_names + assert ipr_attrs[CM_ATTR_IPR_LICENSE] is None + assert ipr_attrs[CM_ATTR_IPR_VERSION] is None + assert ipr_attrs[CM_ATTR_IPR_MANDATORY] is None + assert ipr_attrs[CM_ATTR_IPR_DATE] is None + assert ipr_attrs[CM_ATTR_IPR_URL] is None -def _validate_dq_attributes(cma: CustomMetadata): +def _validate_dq_attributes(cma: CustomMetadataDict): assert cma - c = getattr(cma, CM_ATTR_QUALITY_COUNT_RENAMED) - s = getattr(cma, CM_ATTR_QUALITY_SQL_RENAMED) - t = getattr(cma, CM_ATTR_QUALITY_TYPE_RENAMED) + c = cma[CM_ATTR_QUALITY_COUNT] + s = cma[CM_ATTR_QUALITY_SQL] + t = cma[CM_ATTR_QUALITY_TYPE] assert c assert c == 42 assert s @@ -969,13 +935,14 @@ def _validate_dq_attributes(cma: CustomMetadata): assert t == "Completeness" -def _validate_dq_empty(dq_attrs: CustomMetadata): - assert hasattr(dq_attrs, CM_ATTR_QUALITY_COUNT_RENAMED) - assert hasattr(dq_attrs, CM_ATTR_QUALITY_SQL_RENAMED) - assert hasattr(dq_attrs, CM_ATTR_QUALITY_TYPE_RENAMED) - assert getattr(dq_attrs, CM_ATTR_QUALITY_COUNT_RENAMED) is None - assert getattr(dq_attrs, CM_ATTR_QUALITY_SQL_RENAMED) is None - assert getattr(dq_attrs, CM_ATTR_QUALITY_TYPE_RENAMED) is None +def _validate_dq_empty(dq_attrs: CustomMetadataDict): + attribute_names = dq_attrs.attribute_names + assert CM_ATTR_QUALITY_COUNT in attribute_names + assert CM_ATTR_QUALITY_SQL in attribute_names + assert CM_ATTR_QUALITY_TYPE in attribute_names + assert dq_attrs[CM_ATTR_QUALITY_COUNT] is None + assert dq_attrs[CM_ATTR_QUALITY_SQL] is None + assert dq_attrs[CM_ATTR_QUALITY_TYPE] is None def _validate_raci_structure( @@ -1046,7 +1013,7 @@ def test_add_badge_cm_dq( badge = Badge.create( name=CM_ATTR_QUALITY_COUNT, cm_name=CM_QUALITY, - cm_attribute=CM_ATTR_QUALITY_COUNT_RENAMED, + cm_attribute=CM_ATTR_QUALITY_COUNT, badge_conditions=[ BadgeCondition.create( badge_condition_operator=BadgeComparisonOperator.GTE, diff --git a/tests/unit/test_custom_metadata.py b/tests/unit/test_custom_metadata.py index 6542fd1ea..208ef603a 100644 --- a/tests/unit/test_custom_metadata.py +++ b/tests/unit/test_custom_metadata.py @@ -74,28 +74,43 @@ def test_set_item_with_invalid_name_raises_key_error(self, sut): ): sut["garb"] = ATTR_FIRST_NAME_ID - def test_clear_set_all_attributes_to_none(self, sut): - sut.clear() - for name in sut.attribute_names: - assert sut[name] is None + @pytest.mark.parametrize("name", [ATTR_FIRST_NAME, ATTR_FIRST_NAME]) + def test_clear_all_set_all_attributes_to_none(self, sut, name): + sut.clear_all() + assert sut[name] is None assert sut.modified is True - def test_get_item_using_name_that_has_not_been_set_raises_key_err(self, sut): - with pytest.raises( - KeyError, - match="'First Name' must be set before trying to retrieve the value", - ): - sut[ATTR_FIRST_NAME] + @pytest.mark.parametrize( + "property_to_set, other_property", + [(ATTR_FIRST_NAME, ATTR_LAST_NAME), (ATTR_LAST_NAME, ATTR_FIRST_NAME)], + ) + def test_clear_unset_sets_unset_to_none(self, sut, property_to_set, other_property): + sut[property_to_set] = "bob" + sut.clear_unset() + assert sut[property_to_set] == "bob" + assert sut[other_property] is None + + def test_get_item_using_name_that_has_not_been_set_returns_none(self, sut): + assert sut[ATTR_FIRST_NAME] is None def test_business_attributes_when_no_changes(self, sut): - assert sut.business_attributes == {CM_ID: {}} + assert sut.business_attributes == {} def test_business_attributes_with_data(self, sut, mock_cache): mock_cache.get_attr_id_for_name.side_effect = get_attr_id_for_name alice = "alice" sut[ATTR_FIRST_NAME] = alice - assert sut.business_attributes == {CM_ID: {ATTR_FIRST_NAME_ID: alice}} + assert sut.business_attributes == {ATTR_FIRST_NAME_ID: alice} + + @pytest.mark.parametrize("name", [ATTR_FIRST_NAME, ATTR_FIRST_NAME]) + def test_is_unset_initially_returns_false(self, sut, name): + assert sut.is_set(name) is False + + @pytest.mark.parametrize("name", [ATTR_FIRST_NAME, ATTR_FIRST_NAME]) + def test_unset_after_update_returns_true(self, sut, name): + sut[name] = "bob" + assert sut.is_set(name) is True class TestCustomMetadataProxy: @@ -167,4 +182,4 @@ def test_create(self, mock_cache): cm = CustomMetadataDict(CM_NAME) request = CustomMetadataRequest.create(custom_metadata_dict=cm) - assert request.__root__ == {CM_ID: {}} + assert request.__root__ == {} diff --git a/tests/unit/test_model.py b/tests/unit/test_model.py index 390fcf640..dc3891302 100644 --- a/tests/unit/test_model.py +++ b/tests/unit/test_model.py @@ -5,15 +5,14 @@ from hashlib import md5 from inspect import signature from pathlib import Path -from unittest.mock import create_autospec, patch +from unittest.mock import create_autospec import pytest -from deepdiff import DeepDiff + +# from deepdiff import DeepDiff from pydantic.error_wrappers import ValidationError import pyatlan.cache.classification_cache -from pyatlan.cache.custom_metadata_cache import CustomMetadataCache -from pyatlan.error import NotFoundError from pyatlan.model.assets import ( SQL, AccessControl, @@ -150,7 +149,7 @@ View, validate_single_required_field, ) -from pyatlan.model.core import Announcement, AssetResponse +from pyatlan.model.core import Announcement from pyatlan.model.enums import ( ADLSAccessTier, ADLSLeaseState, @@ -178,7 +177,6 @@ QuickSightFolderType, SourceCostUnitType, ) -from pyatlan.model.response import AssetMutationResponse from pyatlan.model.structs import ( KafkaTopicConsumption, MCRuleComparison, @@ -670,71 +668,11 @@ def test_wrong_json(glossary_json): AtlasGlossaryTerm(**glossary_json) -def test_asset_response(glossary_category_json): - asset_response_json = {"referredEntities": {}, "entity": glossary_category_json} - glossary_category = AssetResponse[AtlasGlossaryCategory]( - **asset_response_json - ).entity - assert glossary_category == AtlasGlossaryCategory(**glossary_category_json) - - @pytest.fixture(scope="function") def the_json(request): return load_json(request.param) -@pytest.mark.parametrize( - "the_json, a_type", - [ - ("glossary.json", AtlasGlossary), - ("glossary_category.json", AtlasGlossaryCategory), - ("glossary_term.json", AtlasGlossaryTerm), - ("glossary_term2.json", AtlasGlossaryTerm), - ("asset_mutated_response_empty.json", AssetMutationResponse), - ("asset_mutated_response_update.json", AssetMutationResponse), - ], - indirect=["the_json"], -) -def test_constructor(the_json, a_type, monkeypatch): - def get_name_for_id(value): - return "PII" - - def get_id_for_name(value): - return "WQ6XGXwq9o7UnZlkWyKhQN" - - monkeypatch.setattr( - pyatlan.cache.classification_cache.ClassificationCache, - "get_id_for_name", - get_id_for_name, - ) - - monkeypatch.setattr( - pyatlan.cache.classification_cache.ClassificationCache, - "get_name_for_id", - get_name_for_id, - ) - - asset = a_type(**the_json) - assert not DeepDiff( - the_json, - json.loads(asset.json(by_alias=True, exclude_unset=True)), - ignore_order=True, - ) - - -def test_has_announcement(glossary): - assert glossary.has_announcement() == ( - bool(glossary.attributes.announcement_type) - or bool(glossary.attributes.announcement_title) - ) - - -def test_set_announcement(glossary, announcement): - glossary.set_announcement(announcement) - assert glossary.has_announcement() is True - assert announcement == glossary.get_announcment() - - def test_create_glossary(): glossary = AtlasGlossary( attributes=AtlasGlossary.Attributes( @@ -744,15 +682,6 @@ def test_create_glossary(): assert "AtlasGlossary" == glossary.type_name -def test_clear_announcement(glossary, announcement): - glossary.set_announcement(announcement) - glossary.remove_announcement() - assert not glossary.has_announcement() - assert glossary.attributes.announcement_title is None - assert glossary.attributes.announcement_type is None - assert glossary.attributes.announcement_message is None - - @pytest.mark.parametrize( "name, connector_type, admin_users, admin_groups, admin_roles, error", [ @@ -1297,136 +1226,6 @@ def test_glossary_term_attributes_create_sets_name_anchor(): assert sut.anchor == glossary -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_get_business_attributes_when_name_not_valid_raises_not_found_error( - mock_client, table, type_def_response -): - mock_client.return_value.get_typedefs.return_value = type_def_response - with pytest.raises( - NotFoundError, match="Custom metadata with name Zoro does not exist." - ): - table.get_custom_metadata("Zoro") - - -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_get_business_attributes_when_name_not_appropriate_for_asset_raises_value_error( - mock_client, table, type_def_response -): - mock_client.get_default_client.return_value = mock_client - mock_client.get_typedefs.return_value = type_def_response - with pytest.raises( - ValueError, match="Custom metadata attributes Moon are not applicable to Table" - ): - table.get_custom_metadata("Moon") - - -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_get_business_attributes_with_valid_name_returns_empty_attribute_when_table_does_not_have_attribute( - mock_client, table, type_def_response -): - mock_client.return_value.get_typedefs.return_value = type_def_response - - monte_carlo = table.get_custom_metadata("Monte Carlo") - - assert monte_carlo is not None - assert monte_carlo.freshness is None - assert monte_carlo.freshness_date is None - assert monte_carlo.table_url is None - - -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_get_business_attributes_with_valid_name_returns_attribute_when_table_has_attribute( - mock_client, table, type_def_response -): - mock_client.return_value.get_typedefs.return_value = type_def_response - table.business_attributes = { - MONTE_CARLO: { - FRESHNESS: "pass", - TABLE_URL: "https://getmontecarlo.com/catalog/", - } - } - - monte_carlo = table.get_custom_metadata("Monte Carlo") - - assert monte_carlo is not None - assert monte_carlo.freshness == "pass" - assert monte_carlo.table_url == "https://getmontecarlo.com/catalog/" - - -def test_set_busines_attributes_with_non_business_attributes_object_raises_value_error( - table, -): - with pytest.raises( - ValueError, - match="business_attributes must be an instance of CustomMetadata", - ): - table.set_custom_metadata({}) - - -def test_set_business_attributes_with_non_appropriate_meta_data_type_name_raises_value_error( - table, -): - with pytest.raises( - ValueError, - match="business_attributes must be an instance of CustomMetadata", - ): - table.set_custom_metadata({}) - - -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_assigning_to_invalid_business_attribute_raises_attribute_error( - mock_client, table, type_def_response -): - mock_client.return_value.get_typedefs.return_value = type_def_response - table.business_attributes = { - MONTE_CARLO: { - FRESHNESS: "pass", - TABLE_URL: "https://getmontecarlo.com/catalog/", - } - } - business_attributes = table.get_custom_metadata("Monte Carlo") - with pytest.raises(AttributeError, match="Attribute bogus does not exist"): - business_attributes.bogus = "123" - - -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_set_business_attributes_with_business_attribute_not_appropriate_to_asset_raises_value_error( - mock_client, table, type_def_response -): - mock_client.return_value.get_typedefs.return_value = type_def_response - - moon = CustomMetadataCache.get_type_for_id(MOON)() - - with pytest.raises( - ValueError, - match="Business attributes Moon are not applicable to Table", - ): - table.set_custom_metadata(moon) - - -@patch("pyatlan.cache.custom_metadata_cache.AtlanClient") -def test_set_business_attributes_with_appropriate_business_attribute_updates_dictionary( - mock_client, table, type_def_response -): - mock_client.return_value.get_typedefs.return_value = type_def_response - - table.business_attributes = { - MONTE_CARLO: { - FRESHNESS: "pass", - TABLE_URL: "https://getmontecarlo.com/catalog/", - } - } - monte_carlo = table.get_custom_metadata("Monte Carlo") - - monte_carlo.freshness = "fail" - monte_carlo.table_url = "http://anywhere.com" - table.set_custom_metadata(monte_carlo) - - monte_carlo_dict = table.business_attributes[MONTE_CARLO] - assert monte_carlo_dict[FRESHNESS] == monte_carlo.freshness - assert monte_carlo_dict[TABLE_URL] == monte_carlo.table_url - - @pytest.mark.parametrize( "cls, name, connection_qualified_name, aws_arn, msg", [