bf2343b7e4
Integration Tests - MySQL + Elasticsearch / Detect Changes (push) Has been cancelled
Integration Tests - MySQL + Elasticsearch / integration-tests-mysql-elasticsearch (push) Has been cancelled
Integration Tests - PostgreSQL + Elasticsearch + Redis / Detect Changes (push) Has been cancelled
Integration Tests - PostgreSQL + Elasticsearch + Redis / integration-tests-postgres-elasticsearch-redis (push) Has been cancelled
Integration Tests - PostgreSQL + OpenSearch / Detect Changes (push) Has been cancelled
Integration Tests - PostgreSQL + OpenSearch / integration-tests-postgres-opensearch (push) Has been cancelled
Java Checkstyle / java-checkstyle (push) Has been cancelled
Maven Collate Tests / maven-collate-ci (push) Has been cancelled
OpenMetadata Service Unit Tests / openmetadata-service-unit-tests-status (push) Has been cancelled
Publish Package to Maven Central Repository / publish-maven-packages (push) Has been cancelled
OpenMetadata Service Unit Tests / Detect Changes (push) Has been cancelled
OpenMetadata Service Unit Tests / openmetadata-service-unit-tests (push) Has been cancelled
OpenMetadata Service Unit Tests / k8s_operator-unit-tests (push) Has been cancelled
282 lines
9.7 KiB
Python
282 lines
9.7 KiB
Python
# 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.
|
|
"""Glue test-connection: its checks match its shipped definition (non-Engine: boto3 client)."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from botocore.exceptions import ClientError
|
|
|
|
from metadata.core.connections.test_connection.check import CheckError, collect_checks
|
|
from metadata.generated.schema.entity.services.connections.database.glueConnection import (
|
|
GlueConnection as GlueConnectionConfig,
|
|
)
|
|
from metadata.generated.schema.security.credentials.awsCredentials import AWSCredentials
|
|
from metadata.ingestion.connections.connection import BaseConnection
|
|
from metadata.ingestion.source.database.glue.connection import (
|
|
GLUE_ERRORS,
|
|
GlueChecks,
|
|
GlueConnection,
|
|
list_databases,
|
|
)
|
|
|
|
CONNECTION_MODULE = "metadata.ingestion.source.database.glue.connection"
|
|
|
|
_SEED = (
|
|
Path(__file__).parents[6]
|
|
/ "openmetadata-service/src/main/resources/json/data"
|
|
/ "testConnections/database/glue.json"
|
|
)
|
|
|
|
|
|
def _config() -> GlueConnectionConfig:
|
|
return GlueConnectionConfig(awsConfig=AWSCredentials(awsRegion="us-east-1"))
|
|
|
|
|
|
def _client(pages: dict[str, list[dict]] | None = None) -> MagicMock:
|
|
"""A Glue client whose paginators yield one page per operation."""
|
|
responses = pages or {}
|
|
client = MagicMock()
|
|
client.get_paginator.side_effect = lambda operation: _paginator(responses.get(operation, [{}]))
|
|
return client
|
|
|
|
|
|
def _paginator(pages: list[dict]) -> MagicMock:
|
|
paginator = MagicMock()
|
|
paginator.paginate.return_value = pages
|
|
return paginator
|
|
|
|
|
|
def _checks(client=None) -> GlueChecks:
|
|
return GlueChecks(connect=lambda: client)
|
|
|
|
|
|
def _check_names() -> set[str]:
|
|
return {step.value for step in collect_checks(_checks())}
|
|
|
|
|
|
def _client_error(code: str, operation: str = "GetDatabases") -> ClientError:
|
|
return ClientError({"Error": {"Code": code, "Message": "denied"}}, operation)
|
|
|
|
|
|
def test_glue_connection_is_base_connection():
|
|
assert issubclass(GlueConnection, BaseConnection)
|
|
|
|
|
|
def test_get_client_builds_the_glue_client():
|
|
with patch(f"{CONNECTION_MODULE}.AWSClient") as mock_aws:
|
|
client = GlueConnection(_config()).client
|
|
|
|
mock_aws.return_value.get_glue_client.assert_called_once_with()
|
|
assert client is mock_aws.return_value.get_glue_client.return_value
|
|
|
|
|
|
def test_glue_checks_cover_expected_steps():
|
|
assert _check_names() == {"GetDatabases", "GetTables"}
|
|
|
|
|
|
def test_glue_checks_match_definition_seed():
|
|
definition_steps = {step["name"] for step in json.loads(_SEED.read_text())["steps"]}
|
|
assert _check_names() == definition_steps
|
|
|
|
|
|
def test_get_databases_is_the_connection_gate():
|
|
steps = json.loads(_SEED.read_text())["steps"]
|
|
assert steps[0]["name"] == "GetDatabases"
|
|
assert steps[0]["category"] == "ConnectionGate"
|
|
|
|
|
|
def test_building_checks_does_not_build_the_client():
|
|
with patch.object(GlueConnection, "_get_client") as mock_build:
|
|
GlueConnection(_config()).checks()
|
|
|
|
mock_build.assert_not_called()
|
|
|
|
|
|
def test_get_databases_enumerates_the_catalog():
|
|
client = _client({"get_databases": [{"DatabaseList": [{"Name": "a"}, {"Name": "b"}]}]})
|
|
|
|
evidence = _checks(client).get_databases()
|
|
|
|
assert evidence.summary == "2 databases enumerated"
|
|
assert evidence.command == "glue:GetDatabases"
|
|
assert evidence.caveat is None
|
|
|
|
|
|
def test_get_databases_caps_reported_count_and_flags_more():
|
|
paginator = _paginator([{"DatabaseList": [{"Name": "a"}, {"Name": "b"}, {"Name": "c"}]}])
|
|
client = MagicMock()
|
|
client.get_paginator.return_value = paginator
|
|
|
|
evidence = list_databases(client, limit=2)
|
|
|
|
assert evidence.summary == "2 databases enumerated (showing first 2; more exist)"
|
|
client.get_paginator.assert_called_once_with("get_databases")
|
|
paginator.paginate.assert_called_once_with(PaginationConfig={"MaxItems": 3})
|
|
|
|
|
|
def test_get_databases_warns_when_catalog_is_empty():
|
|
client = _client({"get_databases": [{"DatabaseList": []}]})
|
|
|
|
evidence = _checks(client).get_databases()
|
|
|
|
assert evidence.summary == "0 databases enumerated"
|
|
assert evidence.caveat is not None
|
|
assert "No databases visible" in evidence.caveat.title
|
|
|
|
|
|
def test_get_databases_failure_reports_the_command_it_ran():
|
|
client = _client()
|
|
cause = _client_error("AccessDeniedException")
|
|
client.get_paginator.side_effect = cause
|
|
|
|
with pytest.raises(CheckError) as failure:
|
|
_checks(client).get_databases()
|
|
|
|
assert failure.value.cause is cause
|
|
assert failure.value.evidence.command == "glue:GetDatabases"
|
|
|
|
|
|
def test_get_tables_probes_the_first_database():
|
|
client = _client(
|
|
{
|
|
"get_databases": [{"DatabaseList": [{"Name": "sales"}]}],
|
|
"get_tables": [{"TableList": [{"Name": "orders"}]}],
|
|
}
|
|
)
|
|
|
|
evidence = _checks(client).get_tables()
|
|
|
|
assert evidence.summary == "1 table in database 'sales'"
|
|
assert evidence.command == "glue:GetTables (DatabaseName=sales)"
|
|
|
|
|
|
def test_get_tables_reports_a_caveat_when_no_database_can_be_probed():
|
|
client = _client({"get_databases": [{"DatabaseList": []}]})
|
|
|
|
evidence = _checks(client).get_tables()
|
|
|
|
assert evidence.command == "glue:GetDatabases"
|
|
assert evidence.caveat is not None
|
|
assert evidence.caveat.title == "No tables visible"
|
|
|
|
|
|
def test_get_tables_skips_an_empty_leading_database():
|
|
# A catalog commonly leads with an empty 'default'; the probe must not stop there.
|
|
client = MagicMock()
|
|
tables = {"default": [{"TableList": []}], "ecommerce": [{"TableList": [{"Name": "orders"}]}]}
|
|
calls = {"n": 0}
|
|
|
|
def paginator(operation):
|
|
if operation == "get_databases":
|
|
return _paginator([{"DatabaseList": [{"Name": "default"}, {"Name": "ecommerce"}]}])
|
|
calls["n"] += 1
|
|
return _paginator(tables["default" if calls["n"] == 1 else "ecommerce"])
|
|
|
|
client.get_paginator.side_effect = paginator
|
|
|
|
evidence = _checks(client).get_tables()
|
|
|
|
assert evidence.summary == "1 table in database 'ecommerce'"
|
|
assert evidence.command == "glue:GetTables (DatabaseName=ecommerce)"
|
|
assert evidence.caveat is None
|
|
|
|
|
|
def test_get_tables_warns_when_no_database_exposes_a_table():
|
|
client = _client(
|
|
{
|
|
"get_databases": [{"DatabaseList": [{"Name": "a"}, {"Name": "b"}]}],
|
|
"get_tables": [{"TableList": []}],
|
|
}
|
|
)
|
|
|
|
evidence = _checks(client).get_tables()
|
|
|
|
assert evidence.summary == "no tables in the first 2 databases"
|
|
assert evidence.caveat is not None
|
|
assert evidence.caveat.title == "No tables visible"
|
|
|
|
|
|
def test_get_tables_failure_names_the_database_it_probed():
|
|
cause = _client_error("EntityNotFoundException", "GetTables")
|
|
client = _client({"get_databases": [{"DatabaseList": [{"Name": "sales"}]}]})
|
|
tables_paginator = MagicMock()
|
|
tables_paginator.paginate.side_effect = cause
|
|
client.get_paginator.side_effect = lambda operation: (
|
|
tables_paginator if operation == "get_tables" else _paginator([{"DatabaseList": [{"Name": "sales"}]}])
|
|
)
|
|
|
|
with pytest.raises(CheckError) as failure:
|
|
_checks(client).get_tables()
|
|
|
|
assert failure.value.cause is cause
|
|
assert failure.value.evidence.command == "glue:GetTables (DatabaseName=sales)"
|
|
|
|
|
|
def test_error_pack_access_denied():
|
|
diagnosis = GLUE_ERRORS.classify(
|
|
ClientError(
|
|
{
|
|
"Error": {
|
|
"Code": "AccessDeniedException",
|
|
"Message": "User is not authorized to perform: glue:GetDatabases",
|
|
}
|
|
},
|
|
"GetDatabases",
|
|
)
|
|
)
|
|
assert diagnosis is not None
|
|
assert "authorized" in diagnosis.title.lower()
|
|
|
|
|
|
def test_error_pack_entity_not_found():
|
|
diagnosis = GLUE_ERRORS.classify(_client_error("EntityNotFoundException", "GetTables"))
|
|
assert diagnosis is not None
|
|
assert diagnosis.title == "Glue database not found"
|
|
|
|
|
|
def test_error_pack_inherits_the_shared_aws_diagnoses():
|
|
# Glue is json-protocol: expired creds arrive as ExpiredTokenException.
|
|
diagnosis = GLUE_ERRORS.classify(_client_error("ExpiredTokenException"))
|
|
assert diagnosis is not None
|
|
assert diagnosis.title == "AWS session token expired"
|
|
|
|
|
|
def test_error_pack_classifies_an_unknown_access_key():
|
|
diagnosis = GLUE_ERRORS.classify(_client_error("UnrecognizedClientException"))
|
|
assert diagnosis is not None
|
|
assert diagnosis.title == "AWS access key not recognized"
|
|
|
|
|
|
def test_error_pack_classifies_a_wrong_secret():
|
|
diagnosis = GLUE_ERRORS.classify(_client_error("InvalidSignatureException"))
|
|
assert diagnosis is not None
|
|
assert diagnosis.title == "AWS secret key does not match"
|
|
|
|
|
|
def test_error_pack_classifies_an_sts_assume_role_denial():
|
|
# The assume-role leg denies with the suffix-less code.
|
|
diagnosis = GLUE_ERRORS.classify(_client_error("AccessDenied", "AssumeRole"))
|
|
assert diagnosis is not None
|
|
assert diagnosis.title == "Not authorized"
|
|
|
|
|
|
def test_error_pack_inherits_the_network_pack_through_aws_errors():
|
|
diagnosis = GLUE_ERRORS.classify(ConnectionRefusedError("refused"))
|
|
assert diagnosis is not None
|
|
assert "refused" in diagnosis.title.lower()
|
|
|
|
|
|
def test_error_pack_unmatched_returns_none():
|
|
assert GLUE_ERRORS.classify(Exception("novel error")) is None
|