OpenMetadata/ingestion/tests/unit/test_source_url.py

74 lines
2.4 KiB
Python
Raw Normal View History

# Copyright 2021 Collate
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# 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 source URL building tests
"""
2022-04-07 08:52:38 -07:00
from unittest import TestCase
from metadata.generated.schema.entity.services.connections.database.mssqlConnection import (
MssqlConnection,
MssqlScheme,
)
from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
MysqlConnection,
)
2022-04-07 08:52:38 -07:00
from metadata.generated.schema.entity.services.connections.database.redshiftConnection import (
RedshiftConnection,
)
from metadata.ingestion.connections.builders import get_connection_url_common
2022-04-07 08:52:38 -07:00
class TestConfig(TestCase):
def test_mysql_url(self):
"""
Validate MySQL URL building
"""
connection = MysqlConnection(
username="username",
password="password",
hostPort="localhost:1234",
)
url = get_connection_url_common(connection)
2022-04-07 08:52:38 -07:00
assert url == "mysql+pymysql://username:password@localhost:1234"
def test_redshift_url(self):
"""
Validate Redshift URL building
"""
connection = RedshiftConnection(
username="username",
password="password",
hostPort="localhost:1234",
database="dev",
)
url = get_connection_url_common(connection)
2022-04-07 08:52:38 -07:00
assert url == "redshift+psycopg2://username:password@localhost:1234/dev"
def test_mssql_url(self):
"""
Validate URL building for MSSQL
"""
from metadata.ingestion.source.database.mssql.connection import (
get_connection_url,
)
expected_url = "mssql+pytds://sa:password@james\\bond:1433"
mssql_conn_obj = MssqlConnection(
username="sa",
password="password",
hostPort="james\\bond:1433",
scheme=MssqlScheme.mssql_pytds,
database=None,
)
assert expected_url == get_connection_url(mssql_conn_obj)