# 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. """ Test Greenplum using the topology """ from unittest import TestCase from unittest.mock import patch from metadata.generated.schema.metadataIngestion.workflow import ( OpenMetadataWorkflowConfig, ) from metadata.ingestion.source.database.greenplum.metadata import GreenplumSource mock_greenplum_config = { "source": { "type": "greenplum", "serviceName": "local_greenplum1", "serviceConnection": { "config": { "type": "Greenplum", "username": "username", "authType": { "password": "password", }, "hostPort": "localhost:5432", "database": "greenplum", } }, "sourceConfig": { "config": { "type": "DatabaseMetadata", } }, }, "sink": { "type": "metadata-rest", "config": {}, }, "workflowConfig": { "openMetadataServerConfig": { "hostPort": "http://localhost:8585/api", "authProvider": "openmetadata", "securityConfig": {"jwtToken": "greenplum"}, } }, } class greenplumUnitTest(TestCase): @patch( "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.test_connection" ) def __init__(self, methodName, test_connection) -> None: super().__init__(methodName) test_connection.return_value = False self.config = OpenMetadataWorkflowConfig.model_validate(mock_greenplum_config) self.greenplum_source = GreenplumSource.create( mock_greenplum_config["source"], self.config.workflowConfig.openMetadataServerConfig, ) @patch("sqlalchemy.engine.base.Engine") @patch( "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.connection" ) def test_close_connection(self, engine, connection): connection.return_value = True self.greenplum_source.close()