# Copyright 2025 Collate # Licensed under the Collate Community License, Version 1.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ OpenMetadata high-level API Table test """ import logging import time from datetime import datetime from unittest import mock import pytest from metadata.generated.schema.entity.data.database import Database from metadata.generated.schema.entity.data.databaseSchema import DatabaseSchema from metadata.generated.schema.entity.data.table import ( Column, ColumnName, DataType, Table, ) from metadata.generated.schema.entity.services.databaseService import DatabaseService from metadata.generated.schema.entity.teams.team import Team from metadata.generated.schema.entity.teams.user import User from metadata.generated.schema.tests.testCase import TestCase as TestCaseEntity from metadata.generated.schema.tests.testCase import TestCaseParameterValue from metadata.generated.schema.tests.testDefinition import ( EntityType, TestCaseParameterDefinition, ) from metadata.generated.schema.type.basic import Markdown from metadata.generated.schema.type.entityReference import EntityReference from metadata.generated.schema.type.entityReferenceList import EntityReferenceList from metadata.generated.schema.type.tagLabel import ( LabelType, State, TagFQN, TagLabel, TagSource, ) from metadata.ingestion.models.patch_request import ( ALLOWED_COMMON_PATCH_FIELDS, RESTRICT_UPDATE_LIST, ) from metadata.ingestion.models.table_metadata import ColumnTag from metadata.utils.helpers import find_column_in_table from ..integration_base import ( # noqa: TID252 generate_name, get_create_entity, get_create_service, get_create_team_entity, get_create_test_case, get_create_test_definition, get_create_test_suite, get_create_user_entity, ) from .conftest import _safe_delete # noqa: TID252 def patch_with_retry(metadata, retries=3, delay=1, **kwargs): """Retry metadata.patch to handle transient MySQL deadlocks. The patch mixin catches server errors and returns None. We retry on None for calls expected to succeed. """ for attempt in range(retries): res = metadata.patch(**kwargs) if res is not None: return res if attempt < retries - 1: time.sleep(delay) return res def patch_owner_with_retry(metadata, retries=3, delay=1, **kwargs): """Retry patch_owner to handle transient MySQL deadlocks. Under parallel execution (--dist loadfile), concurrent writes to the tag_usage table from other test files can cause deadlocks during the server-side PATCH handler. patch_owner returns None on server errors, so we retry on None for calls expected to succeed. """ for attempt in range(retries): res = metadata.patch_owner(**kwargs) if res is not None: return res if attempt < retries - 1: time.sleep(delay) return res PII_TAG_LABEL = TagLabel( tagFQN=TagFQN("PII.Sensitive"), labelType=LabelType.Automated, state=State.Suggested.value, source=TagSource.Classification, ) TIER_TAG_LABEL = TagLabel( tagFQN=TagFQN("Tier.Tier2"), labelType=LabelType.Automated, state=State.Suggested.value, source=TagSource.Classification, ) @pytest.fixture(scope="module") def patch_service(metadata): """Module-scoped database service for patch tests.""" service_name = generate_name() create_service = get_create_service(entity=DatabaseService, name=service_name) service_entity = metadata.create_or_update(data=create_service) yield service_entity _safe_delete( metadata, entity=DatabaseService, entity_id=service_entity.id, recursive=True, hard_delete=True, ) @pytest.fixture(scope="module") def patch_database(metadata, patch_service): """Module-scoped database for patch tests.""" create_db = get_create_entity(entity=Database, reference=patch_service.fullyQualifiedName) return metadata.create_or_update(data=create_db) @pytest.fixture(scope="module") def patch_schema(metadata, patch_database): """Module-scoped database schema for patch tests.""" create_schema = get_create_entity(entity=DatabaseSchema, reference=patch_database.fullyQualifiedName) return metadata.create_or_update(data=create_schema) @pytest.fixture(scope="module") def patch_table(metadata, patch_schema): """Module-scoped table for patch tests.""" create = get_create_entity(entity=Table, reference=patch_schema.fullyQualifiedName) return metadata.create_or_update(data=create) @pytest.fixture(scope="module") def patch_test_table(metadata, patch_schema): """Module-scoped second table for patch-specific tests.""" create = get_create_entity(entity=Table, reference=patch_schema.fullyQualifiedName) return metadata.create_or_update(data=create) @pytest.fixture(scope="module") def patch_test_case(metadata, patch_table): """Module-scoped test case for patch tests.""" test_definition = metadata.create_or_update( get_create_test_definition( parameter_definition=[TestCaseParameterDefinition(name="foo")], entity_type=EntityType.TABLE, ) ) metadata.create_or_update_executable_test_suite( get_create_test_suite(executable_entity_reference=patch_table.fullyQualifiedName.root) ) return metadata.create_or_update( get_create_test_case( entity_link=f"<#E::table::{patch_table.fullyQualifiedName.root}>", test_definition=test_definition.fullyQualifiedName, parameter_values=[TestCaseParameterValue(name="foo", value="10")], ) ) @pytest.fixture(scope="module") def patch_user_1(metadata): """Module-scoped first user for patch owner tests.""" user = metadata.create_or_update(data=get_create_user_entity()) yield user _safe_delete(metadata, entity=User, entity_id=user.id, hard_delete=True) @pytest.fixture(scope="module") def patch_user_2(metadata): """Module-scoped second user for patch owner tests.""" user = metadata.create_or_update(data=get_create_user_entity()) yield user _safe_delete(metadata, entity=User, entity_id=user.id, hard_delete=True) @pytest.fixture(scope="module") def patch_team_1(metadata, patch_user_1, patch_user_2): """Module-scoped first team for patch owner tests.""" team = metadata.create_or_update(data=get_create_team_entity(users=[patch_user_1.id, patch_user_2.id])) yield team _safe_delete(metadata, entity=Team, entity_id=team.id, hard_delete=True) @pytest.fixture(scope="module") def patch_team_2(metadata, patch_user_2): """Module-scoped second team for patch owner tests.""" team = metadata.create_or_update(data=get_create_team_entity(users=[patch_user_2.id])) yield team _safe_delete(metadata, entity=Team, entity_id=team.id, hard_delete=True) @pytest.fixture(scope="module") def owner_user_1(patch_user_1): return EntityReferenceList(root=[EntityReference(id=patch_user_1.id, type="user")]) @pytest.fixture(scope="module") def owner_user_2(patch_user_2): return EntityReferenceList(root=[EntityReference(id=patch_user_2.id, type="user")]) @pytest.fixture(scope="module") def owner_team_1(patch_team_1): return EntityReferenceList(root=[EntityReference(id=patch_team_1.id, type="team")]) @pytest.fixture(scope="module") def owner_team_2(patch_team_2): return EntityReferenceList(root=[EntityReference(id=patch_team_2.id, type="team")]) @pytest.fixture(scope="module") def wait_for_patch_es_index(metadata, patch_user_1): """Wait for ES index to be updated with test user.""" logging.info("Checking ES index status...") tries = 0 res = None while not res and tries <= 5: res = metadata.es_search_from_fqn( entity_type=User, fqn_search_string=patch_user_1.name.root, ) if not res: tries += 1 time.sleep(1) return True class TestOMetaPatch: """ Patch API integration tests. Tests patch operations on tables, descriptions, tags, owners, and test cases. Uses fixtures from conftest: - metadata: OpenMetadata client (session scope) """ def test_patch_table( self, metadata, patch_test_table, owner_user_1, owner_user_2, wait_for_patch_es_index, ): new_patched_table = patch_test_table.model_copy(deep=True) new_patched_table.columns.append( Column(name=ColumnName("new_column"), dataType=DataType.BIGINT), ) new_patched_table.description = Markdown("This should get patched") new_patched_table.columns[0].description = Markdown(root="This column description should get patched") new_patched_table.tags = [PII_TAG_LABEL] new_patched_table.columns[0].tags = [PII_TAG_LABEL] new_patched_table.owners = owner_user_1 patched_table = patch_with_retry( metadata, entity=type(patch_test_table), source=patch_test_table, destination=new_patched_table, allowed_fields=ALLOWED_COMMON_PATCH_FIELDS, restrict_update_fields=RESTRICT_UPDATE_LIST, ) assert patched_table.description.root == "This should get patched" assert patched_table.columns[0].description.root == "This column description should get patched" assert patched_table.tags[0].tagFQN == PII_TAG_LABEL.tagFQN assert patched_table.columns[0].tags[0].tagFQN == PII_TAG_LABEL.tagFQN assert patched_table.owners.root[0].id == owner_user_1.root[0].id new_patched_table = patched_table.model_copy(deep=True) new_patched_table.description = Markdown("This should NOT get patched") new_patched_table.columns[0].description = Markdown(root="This column description should NOT get patched") new_patched_table.tags = [PII_TAG_LABEL, TIER_TAG_LABEL] new_patched_table.columns[0].tags = None new_patched_table.owners = owner_user_2 patched_table = patch_with_retry( metadata, entity=type(patched_table), source=patched_table, destination=new_patched_table, allowed_fields=ALLOWED_COMMON_PATCH_FIELDS, restrict_update_fields=RESTRICT_UPDATE_LIST, ) assert patched_table.description.root != "This should NOT get patched" assert patched_table.columns[0].description.root != "This column description should NOT get patched" assert patched_table.tags[0].tagFQN == PII_TAG_LABEL.tagFQN assert patched_table.tags[1].tagFQN == TIER_TAG_LABEL.tagFQN assert patched_table.columns[0].tags[0].tagFQN == PII_TAG_LABEL.tagFQN assert patched_table.owners.root[0].id == owner_user_1.root[0].id def test_patch_description(self, metadata, patch_table): """Update description and force""" updated: Table = metadata.patch_description(entity=Table, source=patch_table, description="New description") assert updated.description.root == "New description" not_updated = metadata.patch_description(entity=Table, source=patch_table, description="Not passing force") assert not not_updated force_updated: Table = metadata.patch_description( entity=Table, source=patch_table, description="Forced new", force=True, ) assert force_updated.description.root == "Forced new" def test_patch_description_TestCase(self, metadata, patch_test_case): # noqa: N802 """Update description and force""" new_description = "Description " + str(datetime.now()) updated: TestCaseEntity = metadata.patch_description( entity=TestCaseEntity, source=patch_test_case, description=new_description, force=True, ) assert updated.description.root == new_description not_updated = metadata.patch_description( entity=TestCaseEntity, source=patch_test_case, description="Not passing force", ) assert not not_updated force_updated: TestCaseEntity = metadata.patch_description( entity=TestCaseEntity, source=patch_test_case, description="Forced new", force=True, ) assert force_updated.description.root == "Forced new" def test_patch_column_description(self, metadata, patch_table): """Update column description and force""" updated: Table = metadata.patch_column_description( table=patch_table, description="New column description", column_fqn=patch_table.fullyQualifiedName.root + ".another", ) updated_col = find_column_in_table(column_name="another", table=updated) assert updated_col.description.root == "New column description" not_updated = metadata.patch_column_description( table=patch_table, description="Not passing force", column_fqn=patch_table.fullyQualifiedName.root + ".another", ) assert not not_updated force_updated: Table = metadata.patch_column_description( table=patch_table, description="Forced new", column_fqn=patch_table.fullyQualifiedName.root + ".another", force=True, ) updated_col = find_column_in_table(column_name="another", table=force_updated) assert updated_col.description.root == "Forced new" def test_patch_tags(self, metadata, patch_table): """Update table tags""" updated: Table = metadata.patch_tags( entity=Table, source=patch_table, tag_labels=[PII_TAG_LABEL, TIER_TAG_LABEL], ) assert updated.tags[0].tagFQN.root == "PII.Sensitive" assert updated.tags[1].tagFQN.root == "Tier.Tier2" def test_patch_column_tags(self, metadata, patch_table): """Update column tags""" updated: Table = metadata.patch_column_tags( entity=patch_table, column_tags=[ ColumnTag( column_fqn=patch_table.fullyQualifiedName.root + ".id", tag_label=PII_TAG_LABEL, ) ], ) updated_col = find_column_in_table(column_name="id", table=updated) assert updated_col.tags[0].tagFQN.root == "PII.Sensitive" updated_again: Table = metadata.patch_column_tags( entity=patch_table, column_tags=[ ColumnTag( column_fqn=patch_table.fullyQualifiedName.root + ".id", tag_label=TIER_TAG_LABEL, ) ], ) updated_again_col = find_column_in_table(column_name="id", table=updated_again) assert updated_again_col.tags[0].tagFQN.root == "PII.Sensitive" assert updated_again_col.tags[1].tagFQN.root == "Tier.Tier2" def test_patch_column_tags_retries_on_concurrent_modification(self, metadata, patch_table): """A concurrent modification between the SDK's read and its If-Match write must be retried (refetch + reapply), not silently lost. This exercises the optimistic- concurrency path on column-tag patching that guards a tag from landing on the wrong column when columns are changed by another writer between read and write.""" column_fqn = patch_table.fullyQualifiedName.root + ".another" table_path = f"/tables/{patch_table.id.root}" original_patch = metadata.patch state = {"injected": False} def patch_with_concurrent_writer(*args, **kwargs): # On the first attempt, another writer bumps the entity so the staged # If-Match goes stale and the server returns 412, triggering a retry. if not state["injected"]: state["injected"] = True metadata.client.patch( path=table_path, data='[{"op": "add", "path": "/description", "value": "concurrent writer"}]', ) return original_patch(*args, **kwargs) with mock.patch.object(metadata, "patch", side_effect=patch_with_concurrent_writer): updated: Table = metadata.patch_column_tags( entity=patch_table, column_tags=[ColumnTag(column_fqn=column_fqn, tag_label=PII_TAG_LABEL)], ) assert state["injected"], "the concurrent writer should have run" assert updated is not None, "patch must converge after retrying the stale write" updated_col = find_column_in_table(column_name="another", table=updated) assert any(tag.tagFQN.root == "PII.Sensitive" for tag in (updated_col.tags or [])), ( "column tag must persist after the optimistic-lock retry" ) def test_patch_column_falls_back_when_etag_unusable(self, metadata, patch_table): """If the client-derived If-Match can never match the server (e.g. a future client/server ETag-format drift), the column patch must fall back to a non-conditional last-write-wins write rather than silently dropping the change after exhausting retries. We force the mismatch by stubbing the ETag helper to a value the server will always reject.""" with mock.patch( "metadata.ingestion.ometa.mixins.patch_mixin._entity_etag", return_value='"unusable-etag"', ): updated = metadata.patch_column_description( table=patch_table, description="fallback-persisted description", column_fqn=patch_table.fullyQualifiedName.root + ".another", force=True, ) assert updated is not None, "patch must persist via the last-write-wins fallback" updated_col = find_column_in_table(column_name="another", table=updated) assert updated_col.description.root == "fallback-persisted description" def test_patch_owner( self, metadata, patch_table, patch_database, patch_schema, owner_user_1, owner_user_2, owner_team_1, owner_team_2, ): """Update owner""" # Database, no existing owner, owner is a User -> Modified updated: Database = patch_owner_with_retry( metadata, entity=Database, source=patch_database, owners=owner_user_1, ) assert updated is not None assert updated.owners.root[0].id == owner_user_1.root[0].id # Database, existing owner, owner is a User, no force -> Unmodified updated: Database = metadata.patch_owner( entity=Database, source=patch_database, owners=owner_user_2, ) assert updated is None # Database, existing owner, owner is a User, force -> Modified updated: Database = patch_owner_with_retry( metadata, entity=Database, source=patch_database, owners=owner_user_2, force=True, ) assert updated is not None assert updated.owners.root[0].id == owner_user_2.root[0].id # Database, existing owner, no owner, no force -> Unmodified updated: Database = metadata.patch_owner( entity=Database, source=patch_database, ) assert updated is None # Database, existing owner, no owner, force -> Modified updated: Database = patch_owner_with_retry( metadata, entity=Database, source=patch_database, force=True, ) assert updated is not None assert updated.owners == EntityReferenceList(root=[]) # DatabaseSchema, no existing owner, owner is Team -> Modified updated: DatabaseSchema = patch_owner_with_retry( metadata, entity=DatabaseSchema, source=patch_schema, owners=owner_team_1, ) assert updated is not None assert updated.owners.root[0].id == owner_team_1.root[0].id # DatabaseSchema, existing owner, owner is Team, no force -> Unmodified updated: DatabaseSchema = metadata.patch_owner( entity=DatabaseSchema, source=patch_schema, owners=owner_team_2, ) assert updated is None # DatabaseSchema, existing owner, owner is Team, force -> Modified updated: DatabaseSchema = patch_owner_with_retry( metadata, entity=DatabaseSchema, source=patch_schema, owners=owner_team_2, force=True, ) assert updated is not None assert updated.owners.root[0].id == owner_team_2.root[0].id # DatabaseSchema, existing owner, no owner, no force -> Unmodified updated: DatabaseSchema = metadata.patch_owner( entity=DatabaseSchema, source=patch_schema, ) assert updated is None # DatabaseSchema, existing owner, no owner, force -> Modified updated: DatabaseSchema = patch_owner_with_retry( metadata, entity=DatabaseSchema, source=patch_schema, force=True, ) assert updated is not None assert updated.owners == EntityReferenceList(root=[]) # Table, no existing owner, owner is a Team -> Modified updated: Table = patch_owner_with_retry( metadata, entity=Table, source=patch_table, owners=owner_team_1, ) assert updated is not None assert updated.owners.root[0].id == owner_team_1.root[0].id # Table, existing owner, owner is a Team, no force -> Unmodified updated: Table = metadata.patch_owner( entity=Table, source=patch_table, owners=owner_team_2, ) assert updated is None # Table, existing owner, owner is a Team, force -> Modified updated: Table = patch_owner_with_retry( metadata, entity=Table, source=patch_table, owners=owner_team_2, force=True, ) assert updated is not None assert updated.owners.root[0].id == owner_team_2.root[0].id # Table, existing owner, no owner, no force -> Unmodified updated: Table = metadata.patch_owner( entity=Table, source=patch_table, ) assert updated is None # Table, existing owner, no owner, no force -> Modified updated: Table = patch_owner_with_retry( metadata, entity=Table, source=patch_table, force=True, ) assert updated is not None assert updated.owners == EntityReferenceList(root=[]) # Table with non-existent id, force -> Unmodified non_existent_table = patch_table.model_copy(deep=True) non_existent_table.id = "9facb7b3-1dee-4017-8fca-1254b700afef" updated: Table = metadata.patch_owner( entity=Table, source=non_existent_table, force=True, ) assert updated is None def test_patch_nested_col(self, metadata, patch_schema): """Create a table with nested cols and run patch on it""" create = get_create_entity(entity=Table, reference=patch_schema.fullyQualifiedName) created: Table = metadata.create_or_update(create) with_tags: Table = metadata.patch_column_tags( entity=created, column_tags=[ ColumnTag( column_fqn=created.fullyQualifiedName.root + ".struct.id", tag_label=TIER_TAG_LABEL, ) ], ) assert with_tags.columns[2].children[0].tags[0].tagFQN.root == TIER_TAG_LABEL.tagFQN.root with_description: Table = metadata.patch_column_description( table=created, column_fqn=created.fullyQualifiedName.root + ".struct.name", description="I am so nested", ) assert with_description.columns[2].children[1].description.root == "I am so nested" def test_patch_when_inherited_owner(self, metadata, patch_database, owner_team_1): """PATCHing anything when owner is inherited, does not add the owner to the entity""" create_schema = get_create_entity(entity=DatabaseSchema, reference=patch_database.fullyQualifiedName) create_schema.owners = owner_team_1 db_schema_entity = metadata.create_or_update(data=create_schema) create_table = get_create_entity(entity=Table, reference=db_schema_entity.fullyQualifiedName) _table = metadata.create_or_update(data=create_table) table: Table = metadata.get_by_name(entity=Table, fqn=_table.fullyQualifiedName, fields=["owners"]) assert table.owners.root assert table.owners.root[0].inherited dest = table.model_copy(deep=True) dest.description = Markdown(root="potato") metadata.patch( entity=Table, source=table, destination=dest, ) patched_table = metadata.get_by_name(entity=Table, fqn=table.fullyQualifiedName, fields=["owners"]) assert patched_table.description.root == "potato" assert patched_table.owners.root assert patched_table.owners.root[0].inherited def test_patch_skip_on_failure_true(self, metadata, patch_table): """Test that patch operation skips failures when skip_on_failure=True.""" corrupted_destination = patch_table.model_copy(deep=True) corrupted_destination.description = Markdown("Modified description") with mock.patch.object(metadata.client, "patch") as mock_patch_client: mock_patch_client.side_effect = Exception("API error") result = metadata.patch( entity=Table, source=patch_table, destination=corrupted_destination, skip_on_failure=True, ) assert result is None mock_patch_client.assert_called_once() def test_patch_skip_on_failure_false(self, metadata, patch_table): """Test that patch operation raises exception when skip_on_failure=False.""" corrupted_destination = patch_table.model_copy(deep=True) corrupted_destination.description = Markdown("Modified description") with mock.patch.object(metadata.client, "patch") as mock_patch_client: mock_patch_client.side_effect = Exception("API error") with pytest.raises(RuntimeError) as context: metadata.patch( entity=Table, source=patch_table, destination=corrupted_destination, skip_on_failure=False, ) assert "API error" in str(context.value) assert "Failed to update" in str(context.value) mock_patch_client.assert_called_once() def test_patch_skip_on_failure_default_behavior(self, metadata, patch_table): """Test that patch operation defaults to skip_on_failure=True.""" corrupted_destination = patch_table.model_copy(deep=True) corrupted_destination.description = Markdown("Modified description") with mock.patch.object(metadata.client, "patch") as mock_patch_client: mock_patch_client.side_effect = Exception("API error") result = metadata.patch(entity=Table, source=patch_table, destination=corrupted_destination) assert result is None mock_patch_client.assert_called_once() def test_patch_description_skip_on_failure_true(self, metadata, patch_table): """Test that patch_description skips failures when skip_on_failure=True.""" with mock.patch.object(metadata, "_fetch_entity_if_exists") as mock_fetch: mock_fetch.side_effect = Exception("Database error") result = metadata.patch_description( entity=Table, source=patch_table, description="New description", skip_on_failure=True, ) assert result is None mock_fetch.assert_called_once() def test_patch_description_skip_on_failure_false(self, metadata, patch_table): """Test that patch_description raises exception when skip_on_failure=False.""" with mock.patch.object(metadata, "_fetch_entity_if_exists") as mock_fetch: mock_fetch.side_effect = Exception("Database error") with pytest.raises(Exception) as context: metadata.patch_description( entity=Table, source=patch_table, description="New description", skip_on_failure=False, ) assert str(context.value) == "Database error" mock_fetch.assert_called_once() def test_patch_tags_skip_on_failure_true(self, metadata, patch_table): """Test that patch_tags skips failures when skip_on_failure=True.""" with mock.patch.object(metadata, "_fetch_entity_if_exists") as mock_fetch: mock_fetch.side_effect = Exception("Database error") result = metadata.patch_tags( entity=Table, source=patch_table, tag_labels=[PII_TAG_LABEL], skip_on_failure=True, ) assert result is None mock_fetch.assert_called_once() def test_patch_tags_skip_on_failure_false(self, metadata, patch_table): """Test that patch_tags raises exception when skip_on_failure=False.""" with mock.patch.object(metadata, "_fetch_entity_if_exists") as mock_fetch: mock_fetch.side_effect = Exception("Database error") with pytest.raises(Exception) as context: metadata.patch_tags( entity=Table, source=patch_table, tag_labels=[PII_TAG_LABEL], skip_on_failure=False, ) assert str(context.value) == "Database error" mock_fetch.assert_called_once()