diff --git a/ingestion/tests/integration/ometa/conftest.py b/ingestion/tests/integration/ometa/conftest.py index f128dd148ba..d8e099a42c7 100644 --- a/ingestion/tests/integration/ometa/conftest.py +++ b/ingestion/tests/integration/ometa/conftest.py @@ -15,6 +15,13 @@ import uuid import pytest +from metadata.generated.schema.api.data.createGlossary import CreateGlossaryRequest +from metadata.generated.schema.api.data.createGlossaryTerm import ( + CreateGlossaryTermRequest, +) +from metadata.generated.schema.api.teams.createUser import CreateUserRequest +from metadata.generated.schema.entity.data.glossary import Glossary +from metadata.generated.schema.entity.data.glossaryTerm import GlossaryTerm from metadata.generated.schema.entity.services.connections.database.common.basicAuth import ( BasicAuth, ) @@ -22,6 +29,7 @@ from metadata.generated.schema.entity.services.connections.database.mysqlConnect MysqlConnection, ) from metadata.generated.schema.entity.services.databaseService import DatabaseService +from metadata.generated.schema.entity.teams.user import User from metadata.workflow.metadata import MetadataWorkflow from ..containers import MySqlContainerConfigs, get_mysql_container @@ -78,3 +86,61 @@ def workflow(metadata, service, mysql_container): ) workflow_config["ingestionPipelineFQN"] = f"{service_name}.ingestion" return MetadataWorkflow.create(workflow_config) + + +@pytest.fixture +def create_glossary(metadata): + glossaries = [] + + def _create_glossary(create_request: CreateGlossaryRequest): + glossary = metadata.create_or_update(data=create_request) + glossaries.append(glossary) + return glossary + + def teardown(): + for glossary in glossaries: + metadata.delete(entity=Glossary, entity_id=glossary.id, hard_delete=True) + + yield _create_glossary + + teardown() + + +@pytest.fixture +def create_glossary_term(metadata, create_glossary): + glossary_terms = [] + + def _create_glossary_term(create_request: CreateGlossaryTermRequest): + glossary_term = metadata.create_or_update(data=create_request) + + glossary_terms.append(glossary_term) + return glossary_term + + def teardown(): + glossary_terms.reverse() + for glossary_term in glossary_terms: + metadata.delete( + entity=GlossaryTerm, entity_id=glossary_term.id, hard_delete=True + ) + + yield _create_glossary_term + + teardown() + + +@pytest.fixture +def create_user(metadata, request): + users = [] + + def _create_user(create_request: CreateUserRequest): + user = metadata.create_or_update(data=create_request) + users.append(user) + return user + + def teardown(): + for user in users: + metadata.delete(entity=User, entity_id=user.id, hard_delete=True) + + request.addfinalizer(teardown) + + return _create_user diff --git a/ingestion/tests/integration/ometa/test_ometa_glossary.py b/ingestion/tests/integration/ometa/test_ometa_glossary.py index 1766492c101..b3598edd01f 100644 --- a/ingestion/tests/integration/ometa/test_ometa_glossary.py +++ b/ingestion/tests/integration/ometa/test_ometa_glossary.py @@ -12,10 +12,7 @@ """ OpenMetadata high-level API Glossary test """ -import logging -import time from copy import deepcopy -from unittest import TestCase from metadata.generated.schema.api.data.createGlossary import CreateGlossaryRequest from metadata.generated.schema.api.data.createGlossaryTerm import ( @@ -27,14 +24,6 @@ from metadata.generated.schema.entity.data.glossaryTerm import ( GlossaryTerm, TermReference, ) -from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import ( - OpenMetadataConnection, -) -from metadata.generated.schema.entity.teams.user import User -from metadata.generated.schema.security.client.openMetadataJWTClientConfig import ( - OpenMetadataJWTClientConfig, -) -from metadata.generated.schema.type import basic from metadata.generated.schema.type.basic import ( Email, EntityName, @@ -43,575 +32,495 @@ from metadata.generated.schema.type.basic import ( ) from metadata.generated.schema.type.entityReference import EntityReference from metadata.generated.schema.type.entityReferenceList import EntityReferenceList -from metadata.ingestion.ometa.ometa_api import OpenMetadata -from metadata.ingestion.ometa.utils import model_str from metadata.utils import fqn -class OMetaGlossaryTest(TestCase): +class TestOMetaGlossary: """ Run this integration test with the local API available Install the ingestion package before running the tests """ - service_entity_id = None - glossary_entity_id: basic.Uuid = None - create_glossary: CreateGlossaryRequest = None - create_glossary_term_1: CreateGlossaryTermRequest = None - create_glossary_term_2: CreateGlossaryTermRequest = None - create_glossary_term_3: CreateGlossaryTermRequest = None - glossary: Glossary = None - glossary_term_1: GlossaryTerm = None - glossary_term_2: GlossaryTerm = None - glossary_term_3: GlossaryTerm = None - user_1: User = None - user_2: User = None - user_3: User = None - - server_config = OpenMetadataConnection( - hostPort="http://localhost:8585/api", - authProvider="openmetadata", - securityConfig=OpenMetadataJWTClientConfig( - jwtToken="eyJraWQiOiJHYjM4OWEtOWY3Ni1nZGpzLWE5MmotMDI0MmJrOTQzNTYiLCJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiIsImlzQm90IjpmYWxzZSwiaXNzIjoib3Blbi1tZXRhZGF0YS5vcmciLCJpYXQiOjE2NjM5Mzg0NjIsImVtYWlsIjoiYWRtaW5Ab3Blbm1ldGFkYXRhLm9yZyJ9.tS8um_5DKu7HgzGBzS1VTA5uUjKWOCU0B_j08WXBiEC0mr0zNREkqVfwFDD-d24HlNEbrqioLsBuFRiwIWKc1m_ZlVQbG7P36RUxhuv2vbSp80FKyNM-Tj93FDzq91jsyNmsQhyNv_fNr3TXfzzSPjHt8Go0FMMP66weoKMgW2PbXlhVKwEuXUHyakLLzewm9UMeQaEiRzhiTMU3UkLXcKbYEJJvfNFcLwSl9W8JCO_l0Yj3ud-qt_nQYEZwqW6u5nfdQllN133iikV4fM5QZsMCnm8Rq1mvLR0y9bmJiD7fwM1tmJ791TUWqmKaTnP49U493VanKpUAfzIiOiIbhg" - ), - ) - metadata = OpenMetadata(server_config) - - assert metadata.health_check() - - @classmethod - def check_es_index(cls) -> None: + def test_create_glossary(self, create_glossary, create_user): """ - Wait until the index has been updated with the test user. + Create a Glossary """ - logging.info("Checking ES index status...") - tries = 0 - - res = None - while not res and tries <= 5: # Kill in 5 seconds - res = cls.metadata.es_search_from_fqn( - entity_type=User, - fqn_search_string="Levy", - ) - if not res: - tries += 1 - time.sleep(1) - - @classmethod - def setUpClass(cls) -> None: - """ - Prepare ingredients - """ - - cls.user_1 = cls.metadata.create_or_update( - data=CreateUserRequest( - name=EntityName("test.user.1"), - email=Email(root="test.user.1@getcollate.io"), - ), + create_user_request = CreateUserRequest( + name=EntityName("test.user.1"), + email=Email(root="test.user.1@getcollate.io"), ) - cls.user_2 = cls.metadata.create_or_update( - data=CreateUserRequest( - name=EntityName("test.user.2"), - email=Email(root="test.user.2@getcollate.io"), - ), - ) + user = create_user(create_user_request) - cls.user_3 = cls.metadata.create_or_update( - data=CreateUserRequest( - name=EntityName("test.user.3"), - email=Email(root="test.user.3@getcollate.io"), - ), - ) - - cls.check_es_index() - - cls.create_glossary = CreateGlossaryRequest( + create_glossary_request = CreateGlossaryRequest( name=EntityName("test-glossary"), displayName="test-glossary", description=Markdown("Description of test glossary"), owners=EntityReferenceList( root=[ EntityReference( - id=cls.user_1.id, + id=user.id, type="user", ) ], ), ) - cls.create_glossary_term_1 = CreateGlossaryTermRequest( - glossary=FullyQualifiedEntityName(cls.create_glossary.name.root), - name=EntityName("GT1"), - displayName="Glossary Term 1", - description=Markdown("Test glossary term 1"), - owners=EntityReferenceList( - root=[ - EntityReference( - id=cls.user_1.id, - type="user", - ) - ], - ), - ) + glossary = create_glossary(create_glossary_request) - cls.create_glossary_term_2 = CreateGlossaryTermRequest( - glossary=FullyQualifiedEntityName(cls.create_glossary.name.root), - name=EntityName("GT2"), - displayName="Glossary Term 2", - description=Markdown("Test glossary term 2"), - synonyms=[ - EntityName("GT2S1"), - EntityName("GT2S2"), - EntityName("GT2S3"), - ], - owners=EntityReferenceList( - root=[ - EntityReference( - id=cls.user_1.id, - type="user", - ) - ], - ), - ) + assert glossary is not None + assert create_glossary_request.name == glossary.name - cls.create_glossary_term_3 = CreateGlossaryTermRequest( - glossary=FullyQualifiedEntityName(cls.create_glossary.name.root), - name=EntityName("GT3"), - displayName="Glossary Term 3", - description=Markdown("Test glossary term 3"), - synonyms=[ - EntityName("GT2S1"), - EntityName("GT2S2"), - EntityName("GT2S3"), - ], - owners=EntityReferenceList( - root=[ - EntityReference( - id=cls.user_1.id, - type="user", - ) - ], - ), - ) - - # Leave some time for indexes to get updated, otherwise this happens too fast - cls.check_es_index() - - @classmethod - def tearDownClass(cls) -> None: - """ - Clean up - """ - - cls.metadata.delete( - entity=User, - entity_id=cls.user_1.id, - recursive=True, - hard_delete=True, - ) - - cls.metadata.delete( - entity=User, - entity_id=cls.user_2.id, - hard_delete=True, - ) - - cls.metadata.delete( - entity=User, - entity_id=cls.user_3.id, - hard_delete=True, - ) - - if cls.glossary_term_3 is not None: - cls.metadata.delete( - entity=GlossaryTerm, - entity_id=cls.glossary_term_3.id, - hard_delete=True, - ) - - if cls.glossary_term_2 is not None: - cls.metadata.delete( - entity=GlossaryTerm, - entity_id=cls.glossary_term_2.id, - hard_delete=True, - ) - - if cls.glossary_term_1 is not None: - cls.metadata.delete( - entity=GlossaryTerm, - entity_id=cls.glossary_term_1.id, - hard_delete=True, - ) - - # glossary: Glossary = cls.metadata.get_by_name( - # entity=Glossary, - # fqn=model_str(cls.create_glossary.name), - # ) - if cls.glossary_entity_id is not None: - cls.metadata.delete( - entity=Glossary, - entity_id=cls.glossary_entity_id, - hard_delete=True, - ) - - def test_create_glossary(self): - """ - Create a Glossary - """ - self.glossary = self.metadata.create_or_update(self.create_glossary) - self.assertIsNotNone(self.glossary) - self.assertEqual(self.create_glossary.name, self.glossary.name) - if self.glossary_entity_id is None: - self.glossary_entity_id = self.glossary.id - - def test_create_glossary_term(self): + def test_create_glossary_term(self, create_glossary, create_glossary_term): """ Test the creation of a glossary term """ - if OMetaGlossaryTest.glossary_entity_id is None: - glossary: Glossary = self.metadata.create_or_update(self.create_glossary) - OMetaGlossaryTest.glossary_entity_id = glossary.id - - # Create without parent - res: GlossaryTerm = self.metadata.create_or_update(self.create_glossary_term_1) - self.assertIsNotNone(res) - self.assertEqual(self.create_glossary_term_1.name, res.name) - self.assertEqual( - f"{self.create_glossary.name.root}.{res.name.root}", - res.fullyQualifiedName.root, + glossary = create_glossary( + CreateGlossaryRequest( + name=EntityName("test-glossary"), + displayName="test-glossary", + description=Markdown("Description of test glossary"), + ) ) - # Create with parent - if OMetaGlossaryTest.glossary_term_1 is None: - OMetaGlossaryTest.glossary_term_1 = res - self.create_glossary_term_2.parent = self.glossary_term_1.fullyQualifiedName - res = self.metadata.create_or_update(self.create_glossary_term_2) - self.assertIsNotNone(res) - self.assertEqual(self.create_glossary_term_2.name, res.name) - self.assertEqual(model_str(self.create_glossary_term_1.name), res.parent.name) - if OMetaGlossaryTest.glossary_term_2 is None: - OMetaGlossaryTest.glossary_term_2 = res + # Create Glossary Term without Parent + create_glossary_term_1 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT1"), + displayName="Glossary Term 1", + description=Markdown("Test glossary term 1"), + ) + glossary_term_1 = create_glossary_term(create_glossary_term_1) - def test_patch_glossary_term_parent(self): + assert glossary_term_1 is not None + assert glossary_term_1.name == create_glossary_term_1.name + assert ( + glossary_term_1.fullyQualifiedName.root + == f"{glossary.name.root}.{glossary_term_1.name.root}" + ) + + # Create Glossary Term with Parent + create_glossary_term_2 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT2"), + displayName="Glossary Term 2", + description=Markdown("Test glossary term 2"), + parent=glossary_term_1.fullyQualifiedName, + ) + + glossary_term_2 = create_glossary_term(create_glossary_term_2) + + assert glossary_term_2 is not None + assert glossary_term_2.name == create_glossary_term_2.name + assert glossary_term_2.parent.name == glossary_term_1.name.root + + def test_patch_glossary_term_parent( + self, metadata, create_glossary, create_glossary_term + ): """ Update parent via PATCH """ - if OMetaGlossaryTest.glossary_entity_id is None: - glossary: Glossary = self.metadata.create_or_update(self.create_glossary) - OMetaGlossaryTest.glossary_entity_id = glossary.id - if self.glossary_term_1 is None: - OMetaGlossaryTest.glossary_term_1 = self.metadata.create_or_update( - self.create_glossary_term_1 - ) - if self.glossary_term_2 is None: - self.create_glossary_term_2.parent = self.glossary_term_1.fullyQualifiedName - OMetaGlossaryTest.glossary_term_2 = self.metadata.create_or_update( - self.create_glossary_term_2 - ) - if self.glossary_term_3 is None: - OMetaGlossaryTest.glossary_term_3 = self.metadata.create_or_update( - self.create_glossary_term_3 + glossary = create_glossary( + CreateGlossaryRequest( + name=EntityName("test-glossary"), + displayName="test-glossary", + description=Markdown("Description of test glossary"), ) + ) - dest_glossary_term_3 = deepcopy(self.glossary_term_3) - dest_glossary_term_3.parent = EntityReference( - id=self.glossary_term_2.id, type="glossaryTerm" + # Create Glossary Term without Parent + create_glossary_term_1 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT1"), + displayName="Glossary Term 1", + description=Markdown("Test glossary term 1"), + ) + glossary_term_1 = create_glossary_term(create_glossary_term_1) + + # Create Glossary Term with Parent + create_glossary_term_2 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT2"), + displayName="Glossary Term 2", + description=Markdown("Test glossary term 2"), + parent=glossary_term_1.fullyQualifiedName, + ) + glossary_term_2 = create_glossary_term(create_glossary_term_2) + + create_glossary_term_3 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT3"), + displayName="Glossary Term 3", + description=Markdown("Test glossary term 3"), + ) + glossary_term_3 = create_glossary_term(create_glossary_term_3) + + updated_glossary_term_3 = deepcopy(glossary_term_3) + updated_glossary_term_3.parent = EntityReference( + id=glossary_term_2.id, type="glossaryTerm" ) # Add parent - res: GlossaryTerm = self.metadata.patch( + patched_glossary_term_3 = metadata.patch( entity=GlossaryTerm, - source=self.glossary_term_3, - destination=dest_glossary_term_3, + source=glossary_term_3, + destination=updated_glossary_term_3, ) - self.assertIsNotNone(res) - self.assertEqual(self.glossary_term_2.id, res.parent.id) + + assert patched_glossary_term_3 is not None + assert patched_glossary_term_3.parent.id == glossary_term_2.id # Move parent - dest_glossary_term_3.parent = EntityReference( - id=self.glossary_term_1.id, type="glossaryTerm" - ) - res: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_3, - destination=dest_glossary_term_3, + updated_glossary_term_3.parent = EntityReference( + id=glossary_term_1.id, type="glossaryTerm" ) - self.assertIsNotNone(res) - self.assertEqual(self.glossary_term_1.id, res.parent.id) - dest_glossary_term_3 = deepcopy(res) + patched_glossary_term_3 = metadata.patch( + entity=GlossaryTerm, + source=glossary_term_3, + destination=updated_glossary_term_3, + ) + + assert patched_glossary_term_3 is not None + assert patched_glossary_term_3.parent.id == glossary_term_1.id + # Delete parent - dest_glossary_term_3.parent = None - res: GlossaryTerm = self.metadata.patch( + updated_glossary_term_3.parent = None + patched_glossary_term_3 = metadata.patch( entity=GlossaryTerm, - source=res, - destination=dest_glossary_term_3, + source=patched_glossary_term_3, + destination=updated_glossary_term_3, ) - self.assertIsNotNone(res) - self.assertIsNone(res.parent) - def test_patch_glossary_term_related_terms(self): + assert patched_glossary_term_3 is not None + assert patched_glossary_term_3.parent is None + + def test_patch_glossary_term_related_terms( + self, metadata, create_glossary, create_glossary_term + ): """ Update related terms via PATCH """ - if OMetaGlossaryTest.glossary_entity_id is None: - glossary: Glossary = self.metadata.create_or_update(self.create_glossary) - OMetaGlossaryTest.glossary_entity_id = glossary.id - if self.glossary_term_1 is None: - OMetaGlossaryTest.glossary_term_1 = self.metadata.create_or_update( - self.create_glossary_term_1 - ) - if self.glossary_term_2 is None: - OMetaGlossaryTest.glossary_term_2 = self.metadata.create_or_update( - self.create_glossary_term_2 - ) - elif self.glossary_term_2.parent is not None: - dest_glossary_term_2 = deepcopy(self.glossary_term_2) - dest_glossary_term_2.parent = EntityReference( - id=self.glossary_term_1.id, type="glossaryTerm" - ) - self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_2, - destination=dest_glossary_term_2, + glossary = create_glossary( + CreateGlossaryRequest( + name=EntityName("test-glossary"), + displayName="test-glossary", + description=Markdown("Description of test glossary"), ) + ) - if self.glossary_term_3 is None: - OMetaGlossaryTest.glossary_term_3 = self.metadata.create_or_update( - self.create_glossary_term_3 - ) - elif self.glossary_term_3.parent is not None: - dest_glossary_term_3 = deepcopy(self.glossary_term_3) - dest_glossary_term_3.parent = None - self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_3, - destination=dest_glossary_term_3, - ) + create_glossary_term_1 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT1"), + displayName="Glossary Term 1", + description=Markdown("Test glossary term 1"), + ) + glossary_term_1 = create_glossary_term(create_glossary_term_1) + + create_glossary_term_2 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT2"), + displayName="Glossary Term 2", + description=Markdown("Test glossary term 2"), + ) + glossary_term_2 = create_glossary_term(create_glossary_term_2) # Add related term - dest_glossary_term_1 = deepcopy(self.glossary_term_1) - dest_glossary_term_1.relatedTerms = None - if dest_glossary_term_1.relatedTerms is not None: - dest_glossary_term_1.relatedTerms = None - dest_glossary_term_1.relatedTerms = EntityReferenceList( - root=[EntityReference(id=self.glossary_term_2.id, type="glossaryTerm")] + updated_glossary_term_1 = deepcopy(glossary_term_1) + updated_glossary_term_1.relatedTerms = EntityReferenceList( + root=[EntityReference(id=glossary_term_2.id, type="glossaryTerm")] ) - res: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_1, - destination=dest_glossary_term_1, - ) - self.assertIsNotNone(res) - self.assertEqual(1, len(res.relatedTerms.root)) - self.assertEqual(self.glossary_term_2.id, res.relatedTerms.root[0].id) - def test_patch_reviewer(self): + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=glossary_term_1, + destination=updated_glossary_term_1, + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.relatedTerms.root) == 1 + assert patched_glossary_term_1.relatedTerms.root[0].id == glossary_term_2.id + + def test_patch_reviewer( + self, metadata, create_glossary, create_glossary_term, create_user + ): """ Update reviewers via PATCH """ - if OMetaGlossaryTest.glossary is None: - glossary: Glossary = self.metadata.create_or_update(self.create_glossary) - OMetaGlossaryTest.glossary_entity_id = glossary.id - self.glossary = glossary - if self.glossary_term_1 is None: - OMetaGlossaryTest.glossary_term_1 = self.metadata.create_or_update( - self.create_glossary_term_1 - ) - # Add Glossary Reviewer - dest_glossary = deepcopy(self.glossary) - if dest_glossary.reviewers is None: - dest_glossary.reviewers = [] - dest_glossary.reviewers.append(EntityReference(id=self.user_1.id, type="user")) - res_glossary: Glossary = self.metadata.patch( - entity=Glossary, source=self.glossary, destination=dest_glossary - ) - - self.assertIsNotNone(res_glossary) - self.assertEqual(1, len(res_glossary.reviewers)) - self.assertEqual(self.user_1.id, res_glossary.reviewers[0].id) - - # Remove only Glossary reviewer - dest_glossary = deepcopy(res_glossary) - dest_glossary.reviewers.pop(0) - res_glossary: Glossary = self.metadata.patch( - entity=Glossary, source=res_glossary, destination=dest_glossary - ) - self.assertIsNotNone(res_glossary) - self.assertEqual(0, len(res_glossary.reviewers)) - dest_glossary = deepcopy(res_glossary) - dest_glossary.reviewers.append(EntityReference(id=self.user_1.id, type="user")) - dest_glossary.reviewers.append(EntityReference(id=self.user_2.id, type="user")) - dest_glossary.reviewers.append(EntityReference(id=self.user_3.id, type="user")) - res_glossary: Glossary = self.metadata.patch( - entity=Glossary, source=res_glossary, destination=dest_glossary - ) - # Remove one Glossary reviewer when there are many - # delete self.user_3 - dest_glossary = deepcopy(res_glossary) - dest_glossary.reviewers.pop(2) - res_glossary: Glossary = self.metadata.patch( - entity=Glossary, source=res_glossary, destination=dest_glossary - ) - self.assertIsNotNone(res_glossary) - self.assertEqual(2, len(res_glossary.reviewers)) - self.assertEqual(self.user_1.id, res_glossary.reviewers[0].id) - self.assertEqual(self.user_2.id, res_glossary.reviewers[1].id) - - # Add GlossaryTerm Reviewer - dest_glossary_term_1 = deepcopy(self.glossary_term_1) - dest_glossary_term_1.reviewers.root.append( - EntityReference(id=self.user_1.id, type="user") - ) - res_glossary_term: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_1, - destination=dest_glossary_term_1, - ) - - self.assertIsNotNone(res_glossary_term) - self.assertEqual(2, len(res_glossary_term.reviewers.root)) - self.assertTrue( - any( - reviewer.id == self.user_1.id - for reviewer in res_glossary_term.reviewers.root + glossary = create_glossary( + CreateGlossaryRequest( + name=EntityName("test-glossary"), + displayName="test-glossary", + description=Markdown("Description of test glossary"), ) ) - dest_glossary_term_1 = deepcopy(res_glossary_term) - dest_glossary_term_1.reviewers.root.pop(0) - res_glossary_term = self.metadata.patch( - entity=GlossaryTerm, - source=res_glossary_term, - destination=dest_glossary_term_1, - ) - self.assertIsNotNone(res_glossary_term) - # inherited reviewers from glossary - self.assertEqual(2, len(res_glossary_term.reviewers.root)) - - def test_patch_glossary_term_synonyms(self): - """ - Update synonyms via PATCH - """ - if OMetaGlossaryTest.glossary_entity_id is None: - glossary: Glossary = self.metadata.create_or_update(self.create_glossary) - OMetaGlossaryTest.glossary_entity_id = glossary.id - if self.glossary_term_1 is None: - OMetaGlossaryTest.glossary_term_1 = self.metadata.create_or_update( - self.create_glossary_term_1 - ) - dest_glossary_term_1 = deepcopy(self.glossary_term_1) - if dest_glossary_term_1.synonyms is None: - dest_glossary_term_1.synonyms = list() - - if dest_glossary_term_1.synonyms is None: - dest_glossary_term_1.synonyms = [] - dest_glossary_term_1.synonyms.append(EntityName("GT1S1")) - - # Add GlossaryTerm synonym - res: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_1, - destination=dest_glossary_term_1, - ) - self.assertIsNotNone(res) - self.assertEqual(1, len(res.synonyms)) - self.assertEqual("GT1S1", model_str(res.synonyms[0])) - self.glossary_term_1 = self.metadata.get_by_id( - entity=GlossaryTerm, entity_id=self.glossary_term_1.id, fields=["*"] - ) - dest_glossary_term_1 = deepcopy(res) - # Remove GlossaryTerm synonym - dest_glossary_term_1.synonyms.pop(0) - res = self.metadata.patch( - entity=GlossaryTerm, - source=res, - destination=dest_glossary_term_1, - ) - self.assertIsNotNone(res) - self.assertEqual(0, len(res.synonyms)) - dest_glossary_term_1 = deepcopy(res) - dest_glossary_term_1.synonyms.append(EntityName("GT1S1")) - dest_glossary_term_1.synonyms.append(EntityName("GT1S2")) - dest_glossary_term_1.synonyms.append(EntityName("GT1S3")) - - res: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, - source=res, - destination=dest_glossary_term_1, - ) - self.assertIsNotNone(res) - self.assertEqual(3, len(res.synonyms)) - self.assertEqual("GT1S2", model_str(res.synonyms[1])) - - def test_patch_glossary_term_references(self): - """ - Update GlossaryTerm references via PATCH - """ - - if OMetaGlossaryTest.glossary_entity_id is None: - glossary: Glossary = self.metadata.create_or_update(self.create_glossary) - OMetaGlossaryTest.glossary_entity_id = glossary.id - self.glossary = glossary - if self.glossary_term_1 is None: - OMetaGlossaryTest.glossary_term_1 = self.metadata.create_or_update( - self.create_glossary_term_1 - ) - - dest_glossary_term_1 = deepcopy(self.glossary_term_1) - if dest_glossary_term_1.references is None: - dest_glossary_term_1.references = [] - dest_glossary_term_1.references.append( - TermReference(name="GT1S1", endpoint="https://www.getcollate.io") - ) - # Add reference - res: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, - source=self.glossary_term_1, - destination=dest_glossary_term_1, - ) - self.assertIsNotNone(res) - self.assertEqual(1, len(res.references)) - self.assertEqual("GT1S1", res.references[0].name) - - # Remove reference - dest_glossary_term_1_noref = deepcopy(res) - dest_glossary_term_1_noref.references = [] - - self.metadata.patch( - entity=GlossaryTerm, source=res, destination=dest_glossary_term_1_noref - ) - - res: GlossaryTerm = self.metadata.get_by_name( - entity=GlossaryTerm, - fqn=fqn._build( - self.create_glossary.name.root, - self.create_glossary_term_1.name.root, + user_1 = create_user( + CreateUserRequest( + name=EntityName("test.user.1"), + email=Email(root="test.user.1@getcollate.io"), ), ) - self.assertIsNotNone(res) - self.assertEqual(0, len(res.references)) + # Add Glossary Reviewer + updated_glossary = deepcopy(glossary) + if updated_glossary.reviewers is None: + updated_glossary.reviewers = [] + updated_glossary.reviewers.append(EntityReference(id=user_1.id, type="user")) + patched_glossary = metadata.patch( + entity=Glossary, source=glossary, destination=updated_glossary + ) - # Remove reference when there are many - dest_glossary_term_1 = deepcopy(res) - dest_glossary_term_1.references.append( + assert patched_glossary is not None + assert len(patched_glossary.reviewers) == 1 + assert patched_glossary.reviewers[0].id == user_1.id + + # Remove only Glossary reviewer + updated_glossary = deepcopy(patched_glossary) + updated_glossary.reviewers.pop(0) + patched_glossary = metadata.patch( + entity=Glossary, source=patched_glossary, destination=updated_glossary + ) + + assert patched_glossary is not None + assert len(patched_glossary.reviewers) == 0 + + user_2 = create_user( + CreateUserRequest( + name=EntityName("test.user.2"), + email=Email(root="test.user.2@getcollate.io"), + ), + ) + + user_3 = create_user( + CreateUserRequest( + name=EntityName("test.user.3"), + email=Email(root="test.user.3@getcollate.io"), + ), + ) + + # Add Reviewers + updated_glossary = deepcopy(patched_glossary) + updated_glossary.reviewers.append(EntityReference(id=user_1.id, type="user")) + updated_glossary.reviewers.append(EntityReference(id=user_2.id, type="user")) + updated_glossary.reviewers.append(EntityReference(id=user_3.id, type="user")) + patched_glossary = metadata.patch( + entity=Glossary, source=patched_glossary, destination=updated_glossary + ) + + # Remove one Glossary reviewer when there are many + # delete user_3 + updated_glossary = deepcopy(patched_glossary) + updated_glossary.reviewers.pop(2) + patched_glossary = metadata.patch( + entity=Glossary, source=patched_glossary, destination=updated_glossary + ) + + assert patched_glossary is not None + assert len(patched_glossary.reviewers) == 2 + assert patched_glossary.reviewers[0].id == user_1.id + assert patched_glossary.reviewers[1].id == user_2.id + + # Add GlossaryTerm Reviewer + create_glossary_term_1 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT1"), + displayName="Glossary Term 1", + description=Markdown("Test glossary term 1"), + ) + glossary_term_1 = create_glossary_term(create_glossary_term_1) + + updated_glossary_term_1 = deepcopy(glossary_term_1) + updated_glossary_term_1.reviewers.root.append( + EntityReference(id=user_1.id, type="user") + ) + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=glossary_term_1, + destination=updated_glossary_term_1, + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.reviewers.root) == 2 + assert any( + reviewer.id == user_1.id + for reviewer in patched_glossary_term_1.reviewers.root + ) + + updated_glossary_term_1 = deepcopy(patched_glossary_term_1) + updated_glossary_term_1.reviewers.root.pop(0) + metadata.patch( + entity=GlossaryTerm, + source=patched_glossary_term_1, + destination=updated_glossary_term_1, + ) + + patched_glossary_term_1 = metadata.get_by_name( + entity=GlossaryTerm, + fqn=fqn._build( + glossary.name.root, + glossary_term_1.name.root, + ), + fields=["reviewers"], + ) + + assert patched_glossary_term_1 is not None + + # inherited reviewers from glossary + assert len(patched_glossary_term_1.reviewers.root) == 2 + + def test_patch_glossary_term_synonyms( + self, metadata, create_glossary, create_glossary_term + ): + """ + Update synonyms via PATCH + """ + glossary = create_glossary( + CreateGlossaryRequest( + name=EntityName("test-glossary"), + displayName="test-glossary", + description=Markdown("Description of test glossary"), + ) + ) + + create_glossary_term_1 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT1"), + displayName="Glossary Term 1", + description=Markdown("Test glossary term 1"), + ) + glossary_term_1 = create_glossary_term(create_glossary_term_1) + + # Add GlossaryTerm synonym + updated_glossary_term_1 = deepcopy(glossary_term_1) + if updated_glossary_term_1.synonyms is None: + updated_glossary_term_1.synonyms = [] + + if updated_glossary_term_1.synonyms is None: + updated_glossary_term_1.synonyms = [] + + updated_glossary_term_1.synonyms.append(EntityName("GT1S1")) + + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=glossary_term_1, + destination=updated_glossary_term_1, + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.synonyms) == 1 + assert patched_glossary_term_1.synonyms[0].root == "GT1S1" + # self.glossary_term_1 = self.metadata.get_by_id( + # entity=GlossaryTerm, entity_id=self.glossary_term_1.id, fields=["*"] + # ) + + # Remove GlossaryTerm synonym + updated_glossary_term_1 = deepcopy(patched_glossary_term_1) + updated_glossary_term_1.synonyms.pop(0) + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=patched_glossary_term_1, + destination=updated_glossary_term_1, + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.synonyms) == 0 + + updated_glossary_term_1 = deepcopy(patched_glossary_term_1) + updated_glossary_term_1.synonyms.append(EntityName("GT1S1")) + updated_glossary_term_1.synonyms.append(EntityName("GT1S2")) + updated_glossary_term_1.synonyms.append(EntityName("GT1S3")) + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=patched_glossary_term_1, + destination=updated_glossary_term_1, + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.synonyms) == 3 + assert patched_glossary_term_1.synonyms[1].root == "GT1S2" + + def test_patch_glossary_term_references( + self, metadata, create_glossary, create_glossary_term + ): + """ + Update GlossaryTerm references via PATCH + """ + glossary = create_glossary( + CreateGlossaryRequest( + name=EntityName("test-glossary"), + displayName="test-glossary", + description=Markdown("Description of test glossary"), + ) + ) + + create_glossary_term_1 = CreateGlossaryTermRequest( + glossary=FullyQualifiedEntityName(glossary.name.root), + name=EntityName("GT1"), + displayName="Glossary Term 1", + description=Markdown("Test glossary term 1"), + ) + glossary_term_1 = create_glossary_term(create_glossary_term_1) + + # Add reference + updated_glossary_term_1 = deepcopy(glossary_term_1) + if updated_glossary_term_1.references is None: + updated_glossary_term_1.references = [] + updated_glossary_term_1.references.append( TermReference(name="GT1S1", endpoint="https://www.getcollate.io") ) - dest_glossary_term_1.references.append( + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=glossary_term_1, + destination=updated_glossary_term_1, + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.references) == 1 + assert patched_glossary_term_1.references[0].name == "GT1S1" + + # Remove reference + updated_glossary_term_1 = deepcopy(patched_glossary_term_1) + updated_glossary_term_1.references = [] + + metadata.patch( + entity=GlossaryTerm, + source=patched_glossary_term_1, + destination=updated_glossary_term_1, + ) + + patched_glossary_term_1 = metadata.get_by_name( + entity=GlossaryTerm, + fqn=fqn._build( + glossary.name.root, + glossary_term_1.name.root, + ), + ) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.references) == 0 + + # Add many references + updated_glossary_term_1 = deepcopy(patched_glossary_term_1) + updated_glossary_term_1.references.append( + TermReference(name="GT1S1", endpoint="https://www.getcollate.io") + ) + updated_glossary_term_1.references.append( TermReference(name="GT1S2", endpoint="https://open-metadata.org/") ) - dest_glossary_term_1.references.append( + updated_glossary_term_1.references.append( TermReference( name="GT1S3", endpoint="https://github.com/open-metadata/OpenMetadata" ) ) - res: GlossaryTerm = self.metadata.patch( - entity=GlossaryTerm, source=res, destination=dest_glossary_term_1 + patched_glossary_term_1 = metadata.patch( + entity=GlossaryTerm, + source=patched_glossary_term_1, + destination=updated_glossary_term_1, ) - self.assertIsNotNone(res) - self.assertEqual(3, len(res.references)) - self.assertEqual("GT1S2", res.references[1].name) + + assert patched_glossary_term_1 is not None + assert len(patched_glossary_term_1.references) == 3 + assert patched_glossary_term_1.references[1].name == "GT1S2"