|
| 1 | +from unittest.mock import Mock |
| 2 | +from metadata.ingestion.source.database.snowflake.utils import _get_schema_unique_constraints |
| 3 | + |
| 4 | +def test_snowflake_unique_constraint_collision(): |
| 5 | + # Mocking self (SnowflakeDialect) |
| 6 | + dialect_mock = Mock() |
| 7 | + dialect_mock.normalize_name = lambda name: name.lower() if name else name |
| 8 | + |
| 9 | + # Mocking connection |
| 10 | + connection_mock = Mock() |
| 11 | + |
| 12 | + # Mocking the result of connection.execute(...) |
| 13 | + # Simulating two tables 'table_1' and 'table_2' inside the same schema |
| 14 | + # both sharing an identical constraint name like "unique_id" |
| 15 | + row1 = Mock() |
| 16 | + row1._mapping = { |
| 17 | + "constraint_name": "unique_id", |
| 18 | + "table_name": "table_1", |
| 19 | + "column_name": "id" |
| 20 | + } |
| 21 | + |
| 22 | + row2 = Mock() |
| 23 | + row2._mapping = { |
| 24 | + "constraint_name": "unique_id", |
| 25 | + "table_name": "table_2", |
| 26 | + "column_name": "id" |
| 27 | + } |
| 28 | + |
| 29 | + # Composite constraint on table_2 (second column of the same unique key) |
| 30 | + row3 = Mock() |
| 31 | + row3._mapping = { |
| 32 | + "constraint_name": "unique_id", |
| 33 | + "table_name": "table_2", |
| 34 | + "column_name": "email" |
| 35 | + } |
| 36 | + |
| 37 | + result_mock = [row1, row2, row3] |
| 38 | + connection_mock.execute.return_value = result_mock |
| 39 | + |
| 40 | + # Run the patched function |
| 41 | + output = _get_schema_unique_constraints(dialect_mock, connection_mock, "public") |
| 42 | + |
| 43 | + # Output should correctly split the constraints for table_1 and table_2 without collision |
| 44 | + assert "table_1" in output |
| 45 | + assert "table_2" in output |
| 46 | + |
| 47 | + assert len(output["table_1"]) == 1 |
| 48 | + assert output["table_1"][0]["name"] == "unique_id" |
| 49 | + assert output["table_1"][0]["column_names"] == ["id"] |
| 50 | + |
| 51 | + assert len(output["table_2"]) == 1 |
| 52 | + assert output["table_2"][0]["name"] == "unique_id" |
| 53 | + assert output["table_2"][0]["column_names"] == ["id", "email"] |
0 commit comments