KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_connection_url method in testcontainers-python

Best Python code snippet using testcontainers-python_python

test_source_connection.py

Source:test_source_connection.py

...94            scheme=DatabricksScheme.databricks_connector,95            hostPort="1.1.1.1:443",96            token="KlivDTACWXKmZVfN1qIM",97        )98        assert expected_result == get_connection_url(databricks_conn_obj)99    def test_databricks_url_with_db(self):100        expected_result = (101            "databricks+connector://token:KlivDTACWXKmZVfN1qIM@1.1.1.1:443"102        )103        databricks_conn_obj = DatabricksConnection(104            scheme=DatabricksScheme.databricks_connector,105            hostPort="1.1.1.1:443",106            token="KlivDTACWXKmZVfN1qIM",107        )108        assert expected_result == get_connection_url(databricks_conn_obj)109    def test_hive_url(self):110        expected_result = "hive://localhost:10000"111        hive_conn_obj = HiveConnection(112            scheme=HiveScheme.hive, hostPort="localhost:10000"113        )114        assert expected_result == get_connection_url(hive_conn_obj)115        expected_http_result = "hive+http://localhost:1000"116        http_conn_obj = HiveConnection(117            scheme=HiveScheme.hive_http, hostPort="localhost:1000"118        )119        assert expected_http_result == get_connection_url(http_conn_obj)120        exptected_https_result = "hive+https://localhost:1000"121        http_conn_obj = HiveConnection(122            scheme=HiveScheme.hive_https, hostPort="localhost:1000"123        )124        assert exptected_https_result == get_connection_url(http_conn_obj)125    def test_hive_url_custom_auth(self):126        expected_result = "hive://username:password@localhost:10000"127        hive_conn_obj = HiveConnection(128            scheme=HiveScheme.hive.value,129            username="username",130            password="password",131            hostPort="localhost:10000",132            connectionArguments={"auth": "CUSTOM"},133        )134        assert expected_result == get_connection_url(hive_conn_obj)135        # Passing @ in username and password136        expected_result = "hive://username%40444:password%40333@localhost:10000"137        hive_conn_obj = HiveConnection(138            scheme=HiveScheme.hive.value,139            username="username@444",140            password="password@333",141            hostPort="localhost:10000",142            connectionArguments={"auth": "CUSTOM"},143        )144        assert expected_result == get_connection_url(hive_conn_obj)145    def test_hive_url_conn_options_with_db(self):146        expected_result = "hive://localhost:10000/test_db?Key=Value"147        hive_conn_obj = HiveConnection(148            hostPort="localhost:10000",149            databaseSchema="test_db",150            connectionOptions={"Key": "Value"},151        )152        assert expected_result == get_connection_url(hive_conn_obj)153    def test_hive_url_conn_options_without_db(self):154        expected_result = "hive://localhost:10000?Key=Value"155        hive_conn_obj = HiveConnection(156            hostPort="localhost:10000",157            connectionOptions={"Key": "Value"},158        )159        assert expected_result == get_connection_url(hive_conn_obj)160    def test_hive_url_with_kerberos_auth(self):161        expected_result = "hive://localhost:10000"162        hive_conn_obj = HiveConnection(163            scheme=HiveScheme.hive.value,164            hostPort="localhost:10000",165            connectionArguments={166                "auth": "KERBEROS",167                "kerberos_service_name": "hive",168            },169        )170        assert expected_result == get_connection_url(hive_conn_obj)171    def test_hive_url_with_ldap_auth(self):172        expected_result = "hive://username:password@localhost:10000"173        hive_conn_obj = HiveConnection(174            scheme=HiveScheme.hive.value,175            username="username",176            password="password",177            hostPort="localhost:10000",178            connectionArguments={"auth": "LDAP"},179        )180        assert expected_result == get_connection_url(hive_conn_obj)181    def test_hive_url_without_auth(self):182        expected_result = "hive://localhost:10000"183        hive_conn_obj = HiveConnection(184            scheme=HiveScheme.hive.value,185            username="username",186            password="password",187            hostPort="localhost:10000",188            connectionArguments={"customKey": "value"},189        )190        assert expected_result == get_connection_url(hive_conn_obj)191    def test_trino_url_without_params(self):192        expected_url = "trino://username:pass@localhost:443/catalog"193        trino_conn_obj = TrinoConnection(194            scheme=TrinoScheme.trino,195            hostPort="localhost:443",196            username="username",197            password="pass",198            catalog="catalog",199        )200        assert expected_url == get_connection_url(trino_conn_obj)201        # Passing @ in username and password202        expected_url = "trino://username%40444:pass%40111@localhost:443/catalog"203        trino_conn_obj = TrinoConnection(204            scheme=TrinoScheme.trino,205            hostPort="localhost:443",206            username="username@444",207            password="pass@111",208            catalog="catalog",209        )210        assert expected_url == get_connection_url(trino_conn_obj)211    def test_trino_conn_arguments(self):212        # connection arguments without connectionArguments and without proxies213        expected_args = {}214        trino_conn_obj = TrinoConnection(215            username="user",216            password=None,217            hostPort="localhost:443",218            catalog="tpcds",219            connectionArguments=None,220            scheme=TrinoScheme.trino,221        )222        assert expected_args == get_connection_args(trino_conn_obj)223        # connection arguments with connectionArguments and without proxies224        expected_args = {"user": "user-to-be-impersonated"}225        trino_conn_obj = TrinoConnection(226            username="user",227            password=None,228            hostPort="localhost:443",229            catalog="tpcds",230            connectionArguments={"user": "user-to-be-impersonated"},231            scheme=TrinoScheme.trino,232        )233        assert expected_args == get_connection_args(trino_conn_obj)234        # connection arguments without connectionArguments and with proxies235        expected_args = {}236        trino_conn_obj = TrinoConnection(237            username="user",238            password=None,239            hostPort="localhost:443",240            catalog="tpcds",241            connectionArguments=None,242            proxies={"http": "foo.bar:3128", "http://host.name": "foo.bar:4012"},243            scheme=TrinoScheme.trino,244        )245        conn_args = get_connection_args(trino_conn_obj)246        assert "http_session" in conn_args247        conn_args.pop("http_session")248        assert expected_args == conn_args249        # connection arguments with connectionArguments and with proxies250        expected_args = {"user": "user-to-be-impersonated"}251        trino_conn_obj = TrinoConnection(252            username="user",253            password=None,254            hostPort="localhost:443",255            catalog="tpcds",256            connectionArguments={"user": "user-to-be-impersonated"},257            proxies={"http": "foo.bar:3128", "http://host.name": "foo.bar:4012"},258            scheme=TrinoScheme.trino,259        )260        conn_args = get_connection_args(trino_conn_obj)261        assert "http_session" in conn_args262        conn_args.pop("http_session")263        assert expected_args == conn_args264    def test_trino_url_with_params(self):265        expected_url = "trino://username:pass@localhost:443/catalog?param=value"266        trino_conn_obj = TrinoConnection(267            scheme=TrinoScheme.trino,268            hostPort="localhost:443",269            username="username",270            password="pass",271            catalog="catalog",272            params={"param": "value"},273        )274        assert expected_url == get_connection_url(trino_conn_obj)275    def test_trino_with_proxies(self):276        test_proxies = {"http": "http_proxy", "https": "https_proxy"}277        trino_conn_obj = TrinoConnection(278            scheme=TrinoScheme.trino,279            hostPort="localhost:443",280            username="username",281            password="pass",282            catalog="catalog",283            proxies=test_proxies,284        )285        assert (286            test_proxies287            == get_connection_args(trino_conn_obj).get("http_session").proxies288        )289    def test_trino_without_catalog(self):290        # Test trino url without catalog291        expected_url = "trino://username:pass@localhost:443"292        trino_conn_obj = TrinoConnection(293            scheme=TrinoScheme.trino,294            hostPort="localhost:443",295            username="username",296            password="pass",297        )298        assert expected_url == get_connection_url(trino_conn_obj)299    def test_vertica_url(self):300        expected_url = (301            "vertica+vertica_python://username:password@localhost:5443/database"302        )303        vertica_conn_obj = VerticaConnection(304            scheme=VerticaScheme.vertica_vertica_python,305            hostPort="localhost:5443",306            username="username",307            password="password",308            database="database",309        )310        assert expected_url == get_connection_url(vertica_conn_obj)311        # Passing @ in username and password312        expected_url = "vertica+vertica_python://username%40444:password%40123@localhost:5443/database"313        vertica_conn_obj = VerticaConnection(314            scheme=VerticaScheme.vertica_vertica_python,315            hostPort="localhost:5443",316            username="username@444",317            password="password@123",318            database="database",319        )320        assert expected_url == get_connection_url(vertica_conn_obj)321    def test_druid_url(self):322        expected_url = "druid://localhost:8082/druid/v2/sql"323        druid_conn_obj = DruidConnection(324            scheme=DruidScheme.druid, hostPort="localhost:8082"325        )326        assert expected_url == get_connection_url(druid_conn_obj)327    def test_pinotdb_url(self):328        expected_url = (329            "pinot://localhost:8099/query/sql?controller=http://localhost:9000/"330        )331        pinot_conn_obj = PinotDBConnection(332            scheme=PinotDBScheme.pinot,333            hostPort="localhost:8099",334            pinotControllerHost="http://localhost:9000/",335        )336        assert expected_url == get_connection_url(pinot_conn_obj)337    def test_mysql_url(self):338        # connection arguments without db339        expected_url = "mysql+pymysql://openmetadata_user:@localhost:3306"340        mysql_conn_obj = MysqlConnection(341            username="openmetadata_user",342            hostPort="localhost:3306",343            scheme=MySQLScheme.mysql_pymysql,344        )345        assert expected_url == get_connection_url(mysql_conn_obj)346        # connection arguments with db347        expected_url = "mysql+pymysql://openmetadata_user:@localhost:3306"348        mysql_conn_obj = MysqlConnection(349            username="openmetadata_user",350            hostPort="localhost:3306",351            scheme=MySQLScheme.mysql_pymysql,352        )353        assert expected_url == get_connection_url(mysql_conn_obj)354    def test_clickhouse_url(self):355        # connection arguments without db356        expected_url = "clickhouse+http://username:@localhost:8123"357        clickhouse_conn_obj = ClickhouseConnection(358            username="username",359            hostPort="localhost:8123",360            scheme=ClickhouseScheme.clickhouse_http,361            databaseSchema=None,362        )363        assert expected_url == get_connection_url(clickhouse_conn_obj)364        # connection arguments with db365        expected_url = "clickhouse+http://username:@localhost:8123/default"366        clickhouse_conn_obj = ClickhouseConnection(367            username="username",368            hostPort="localhost:8123",369            scheme=ClickhouseScheme.clickhouse_http,370            databaseSchema="default",371        )372        assert expected_url == get_connection_url(clickhouse_conn_obj)373        expected_url = (374            "clickhouse+http://username:@localhost:8123/default?protocol=https"375        )376        clickhouse_conn_obj = ClickhouseConnection(377            username="username",378            hostPort="localhost:8123",379            scheme=ClickhouseScheme.clickhouse_http,380            connectionOptions=dict(protocol="https"),381            databaseSchema="default",382        )383        assert expected_url == get_connection_url(clickhouse_conn_obj)384    def test_mariadb_url(self):385        # connection arguments without db386        expected_url = "mysql+pymysql://openmetadata_user:@localhost:3306"387        mariadb_conn_obj = MariaDBConnection(388            username="openmetadata_user",389            hostPort="localhost:3306",390            scheme=MariaDBScheme.mysql_pymysql,391        )392        assert expected_url == get_connection_url(mariadb_conn_obj)393        # connection arguments with db394        expected_url = "mysql+pymysql://openmetadata_user:@localhost:3306"395        mariadb_conn_obj = MariaDBConnection(396            username="openmetadata_user",397            hostPort="localhost:3306",398            scheme=MariaDBScheme.mysql_pymysql,399        )400        assert expected_url == get_connection_url(mariadb_conn_obj)401    def test_postgres_url(self):402        # connection arguments without db403        expected_url = "postgresql+psycopg2://openmetadata_user:@localhost:5432"404        postgres_conn_obj = PostgresConnection(405            username="openmetadata_user",406            hostPort="localhost:5432",407            scheme=PostgresScheme.postgresql_psycopg2,408            database=None,409        )410        assert expected_url == get_connection_url(postgres_conn_obj)411        # connection arguments witho db412        expected_url = "postgresql+psycopg2://openmetadata_user:@localhost:5432/default"413        postgres_conn_obj = PostgresConnection(414            username="openmetadata_user",415            hostPort="localhost:5432",416            scheme=PostgresScheme.postgresql_psycopg2,417            database="default",418        )419        assert expected_url == get_connection_url(postgres_conn_obj)420    def test_redshift_url(self):421        # connection arguments witho db422        expected_url = "redshift+psycopg2://username:strong_password@cluster.name.region.redshift.amazonaws.com:5439/dev"423        redshift_conn_obj = RedshiftConnection(424            username="username",425            password="strong_password",426            hostPort="cluster.name.region.redshift.amazonaws.com:5439",427            scheme=RedshiftScheme.redshift_psycopg2,428            database="dev",429        )430        assert expected_url == get_connection_url(redshift_conn_obj)431    def test_singleStore_url(self):432        # connection arguments without db433        expected_url = "mysql+pymysql://openmetadata_user:@localhost:5432"434        singleStore_conn_obj = SingleStoreConnection(435            username="openmetadata_user",436            hostPort="localhost:5432",437            scheme=SingleStoreScheme.mysql_pymysql,438        )439        assert expected_url == get_connection_url(singleStore_conn_obj)440        # connection arguments with db441        expected_url = "mysql+pymysql://openmetadata_user:@localhost:5432"442        singleStore_conn_obj = SingleStoreConnection(443            username="openmetadata_user",444            hostPort="localhost:5432",445            scheme=SingleStoreScheme.mysql_pymysql,446        )447        assert expected_url == get_connection_url(singleStore_conn_obj)448    def test_db2_url(self):449        # connection arguments without db450        expected_url = "db2+ibm_db://openmetadata_user:@localhost:50000"451        db2_conn_obj = Db2Connection(452            scheme=Db2Scheme.db2_ibm_db,453            username="openmetadata_user",454            hostPort="localhost:50000",455        )456        assert expected_url == get_connection_url(db2_conn_obj)457        # connection arguments with db458        expected_url = "db2+ibm_db://openmetadata_user:@localhost:50000"459        db2_conn_obj = Db2Connection(460            username="openmetadata_user",461            hostPort="localhost:50000",462            scheme=Db2Scheme.db2_ibm_db,463        )464        assert expected_url == get_connection_url(db2_conn_obj)465    def test_snowflake_url(self):466        # connection arguments without db467        expected_url = "snowflake://coding:Abhi@ue18849.us-east-2.aws?account=ue18849.us-east-2.aws&warehouse=COMPUTE_WH"468        snowflake_conn_obj = SnowflakeConnection(469            scheme=SnowflakeScheme.snowflake,470            username="coding",471            password="Abhi",472            warehouse="COMPUTE_WH",473            account="ue18849.us-east-2.aws",474        )475        assert expected_url == get_connection_url(snowflake_conn_obj)476    def test_snowflake_url(self):477        # Passing @ in username and password478        expected_url = "snowflake://coding%40444:Abhi%40123@ue18849.us-east-2.aws?account=ue18849.us-east-2.aws&warehouse=COMPUTE_WH"479        snowflake_conn_obj = SnowflakeConnection(480            scheme=SnowflakeScheme.snowflake,481            username="coding@444",482            password="Abhi@123",483            warehouse="COMPUTE_WH",484            account="ue18849.us-east-2.aws",485        )486        assert expected_url == get_connection_url(snowflake_conn_obj)487        # connection arguments with db488        expected_url = "snowflake://coding:Abhi@ue18849.us-east-2.aws/testdb?account=ue18849.us-east-2.aws&warehouse=COMPUTE_WH"489        snowflake_conn_obj = SnowflakeConnection(490            scheme=SnowflakeScheme.snowflake,491            username="coding",492            password="Abhi",493            database="testdb",494            warehouse="COMPUTE_WH",495            account="ue18849.us-east-2.aws",496        )497        assert expected_url == get_connection_url(snowflake_conn_obj)498    def test_mysql_conn_arguments(self):499        # connection arguments without connectionArguments500        expected_args = {}501        mysql_conn_obj = MysqlConnection(502            username="user",503            password=None,504            hostPort="localhost:443",505            connectionArguments=None,506            scheme=MySQLScheme.mysql_pymysql,507        )508        assert expected_args == get_connection_args(mysql_conn_obj)509        # connection arguments with connectionArguments510        expected_args = {"user": "user-to-be-impersonated"}511        mysql_conn_obj = MysqlConnection(512            username="user",513            password=None,514            hostPort="localhost:443",515            connectionArguments={"user": "user-to-be-impersonated"},516            scheme=MySQLScheme.mysql_pymysql,517        )518        assert expected_args == get_connection_args(mysql_conn_obj)519    def test_clickhouse_conn_arguments(self):520        # connection arguments without connectionArguments521        expected_args = {}522        clickhouse_conn_obj = ClickhouseConnection(523            username="user",524            password=None,525            hostPort="localhost:443",526            databaseSchema=None,527            connectionArguments=None,528            scheme=ClickhouseScheme.clickhouse_http,529        )530        assert expected_args == get_connection_args(clickhouse_conn_obj)531        # connection arguments with connectionArguments532        expected_args = {"user": "user-to-be-impersonated"}533        clickhouse_conn_obj = ClickhouseConnection(534            username="user",535            password=None,536            hostPort="localhost:443",537            databaseSchema="tiny",538            connectionArguments={"user": "user-to-be-impersonated"},539            scheme=ClickhouseScheme.clickhouse_http,540        )541        assert expected_args == get_connection_args(clickhouse_conn_obj)542    def test_mariadb_conn_arguments(self):543        # connection arguments without connectionArguments544        expected_args = {}545        mariadb_conn_obj = MariaDBConnection(546            username="user",547            password=None,548            hostPort="localhost:443",549            connectionArguments=None,550            scheme=MariaDBScheme.mysql_pymysql,551        )552        assert expected_args == get_connection_args(mariadb_conn_obj)553        # connection arguments with connectionArguments554        expected_args = {"user": "user-to-be-impersonated"}555        mariadb_conn_obj = MariaDBConnection(556            username="user",557            password=None,558            hostPort="localhost:443",559            connectionArguments={"user": "user-to-be-impersonated"},560            scheme=MariaDBScheme.mysql_pymysql,561        )562        assert expected_args == get_connection_args(mariadb_conn_obj)563    def test_postgres_conn_arguments(self):564        # connection arguments without connectionArguments565        expected_args = {}566        postgres_conn_obj = PostgresConnection(567            username="user",568            password=None,569            hostPort="localhost:443",570            database=None,571            connectionArguments=None,572            scheme=PostgresScheme.postgresql_psycopg2,573        )574        assert expected_args == get_connection_args(postgres_conn_obj)575        # connection arguments with connectionArguments576        expected_args = {"user": "user-to-be-impersonated"}577        postgres_conn_obj = PostgresConnection(578            username="user",579            password=None,580            hostPort="localhost:443",581            database="tiny",582            connectionArguments={"user": "user-to-be-impersonated"},583            scheme=PostgresScheme.postgresql_psycopg2,584        )585        assert expected_args == get_connection_args(postgres_conn_obj)586    def test_redshift_conn_arguments(self):587        # connection arguments without connectionArguments588        expected_args = {}589        redshift_conn_obj = RedshiftConnection(590            username="user",591            password=None,592            hostPort="localhost:443",593            database="tiny",594            connectionArguments=None,595            scheme=RedshiftScheme.redshift_psycopg2,596        )597        assert expected_args == get_connection_args(redshift_conn_obj)598        # connection arguments with connectionArguments599        expected_args = {"user": "user-to-be-impersonated"}600        redshift_conn_obj = RedshiftConnection(601            username="user",602            password=None,603            hostPort="localhost:443",604            database="tiny",605            connectionArguments={"user": "user-to-be-impersonated"},606            scheme=RedshiftScheme.redshift_psycopg2,607        )608        assert expected_args == get_connection_args(redshift_conn_obj)609    def test_singleStore_conn_arguments(self):610        # connection arguments without connectionArguments611        expected_args = {}612        singleStore_conn_obj = SingleStoreConnection(613            username="user",614            password=None,615            hostPort="localhost:443",616            connectionArguments=None,617            scheme=SingleStoreScheme.mysql_pymysql,618        )619        assert expected_args == get_connection_args(singleStore_conn_obj)620        # connection arguments with connectionArguments621        expected_args = {"user": "user-to-be-impersonated"}622        singleStore_conn_obj = SingleStoreConnection(623            username="user",624            password=None,625            hostPort="localhost:443",626            connectionArguments={"user": "user-to-be-impersonated"},627            scheme=SingleStoreScheme.mysql_pymysql,628        )629        assert expected_args == get_connection_args(singleStore_conn_obj)630    def test_db2_conn_arguments(self):631        # connection arguments without connectionArguments632        expected_args = {}633        db2_conn_obj = Db2Connection(634            username="user",635            password=None,636            hostPort="localhost:443",637            connectionArguments=None,638            scheme=Db2Scheme.db2_ibm_db,639        )640        assert expected_args == get_connection_args(db2_conn_obj)641        # connection arguments with connectionArguments642        expected_args = {"user": "user-to-be-impersonated"}643        db2_conn_obj = Db2Connection(644            username="user",645            password=None,646            hostPort="localhost:443",647            connectionArguments={"user": "user-to-be-impersonated"},648            scheme=Db2Scheme.db2_ibm_db,649        )650        assert expected_args == get_connection_args(db2_conn_obj)651    def test_snowflake_conn_arguments(self):652        # connection arguments without connectionArguments653        expected_args = {}654        snowflake_conn_obj = SnowflakeConnection(655            username="user",656            password="test-pwd",657            database="tiny",658            warehouse="COMPUTE_WH",659            scheme=SnowflakeScheme.snowflake,660            account="account.region_name.cloud_service",661        )662        assert expected_args == get_connection_args(snowflake_conn_obj)663        # connection arguments with connectionArguments664        expected_args = {"user": "user-to-be-impersonated"}665        snowflake_conn_obj = SnowflakeConnection(666            username="user",667            password="test-pwd",668            database="tiny",669            warehouse="COMPUTE_WH",670            connectionArguments={"user": "user-to-be-impersonated"},671            scheme=SnowflakeScheme.snowflake,672            account="account.region_name.cloud_service",673        )674        assert expected_args == get_connection_args(snowflake_conn_obj)675    def test_athena_url(self):676        # connection arguments without db677        awsCreds = awsCredentials.AWSCredentials(678            awsAccessKeyId="key", awsRegion="us-east-2", awsSecretAccessKey="secret_key"679        )680        expected_url = "awsathena+rest://key:secret_key@athena.us-east-2.amazonaws.com:443?s3_staging_dir=s3athena-postgres&work_group=primary"681        athena_conn_obj = AthenaConnection(682            awsConfig=awsCreds,683            s3StagingDir="s3athena-postgres",684            workgroup="primary",685            scheme=AthenaScheme.awsathena_rest,686        )687        assert expected_url == get_connection_url(athena_conn_obj)688        # connection arguments witho db689        expected_url = "awsathena+rest://key:secret_key@athena.us-east-2.amazonaws.com:443?s3_staging_dir=s3athena-postgres&work_group=primary"690        athena_conn_obj = AthenaConnection(691            awsConfig=awsCreds,692            s3StagingDir="s3athena-postgres",693            workgroup="primary",694            scheme=AthenaScheme.awsathena_rest,695        )696        assert expected_url == get_connection_url(athena_conn_obj)697    def test_mssql_url(self):698        # connection arguments without db699        expected_url = "mssql+pytds://sa:password@localhost:1433"700        mssql_conn_obj = MssqlConnection(701            username="sa",702            password="password",703            hostPort="localhost:1433",704            scheme=MssqlScheme.mssql_pytds,705            database=None,706        )707        assert expected_url == get_connection_url(mssql_conn_obj)708    def test_mssql_url(self):709        # Passing @ in username and password710        expected_url = "mssql+pytds://sa%40123:password%40444@localhost:1433"711        mssql_conn_obj = MssqlConnection(712            username="sa@123",713            password="password@444",714            hostPort="localhost:1433",715            scheme=MssqlScheme.mssql_pytds,716            database=None,717        )718        assert expected_url == get_connection_url(mssql_conn_obj)719        # connection arguments witho db720        expected_url = "mssql+pytds://sa:password@localhost:1433/catalog_test"721        mssql_conn_obj = MssqlConnection(722            username="sa",723            password="password",724            hostPort="localhost:1433",725            scheme=MssqlScheme.mssql_pytds,726            database="catalog_test",727        )728        assert expected_url == get_connection_url(mssql_conn_obj)729    def test_presto_url(self):730        # connection arguments without db731        expected_url = "presto://admin@localhost:8080/test_catalog"732        presto_conn_obj = PrestoConnection(733            username="admin",734            hostPort="localhost:8080",735            scheme=PrestoScheme.presto,736            catalog="test_catalog",737        )738        assert expected_url == get_connection_url(presto_conn_obj)739        # Passing @ in username and password740        expected_url = "presto://admin%40333:pass%40111@localhost:8080/test_catalog"741        presto_conn_obj = PrestoConnection(742            username="admin@333",743            password="pass@111",744            hostPort="localhost:8080",745            scheme=PrestoScheme.presto,746            catalog="test_catalog",747        )748        assert expected_url == get_connection_url(presto_conn_obj)749    def test_presto_without_catalog(self):750        # Test presto url without catalog751        expected_url = "presto://username:pass@localhost:8080"752        presto_conn_obj = PrestoConnection(753            scheme=PrestoScheme.presto,754            hostPort="localhost:8080",755            username="username",756            password="pass",757        )758        assert expected_url == get_connection_url(presto_conn_obj)759    def test_oracle_url(self):760        # oracle with db761        expected_url = "oracle+cx_oracle://admin:password@localhost:1541/testdb"762        oracle_conn_obj = OracleConnection(763            username="admin",764            password="password",765            hostPort="localhost:1541",766            scheme=OracleScheme.oracle_cx_oracle,767            oracleConnectionType=OracleDatabaseSchema(databaseSchema="testdb"),768        )769        assert expected_url == get_connection_url(oracle_conn_obj)770        # oracle with service name771        expected_url = (772            "oracle+cx_oracle://admin:password@localhost:1541/?service_name=testdb"773        )774        oracle_conn_obj = OracleConnection(775            username="admin",776            password="password",777            hostPort="localhost:1541",778            scheme=OracleScheme.oracle_cx_oracle,779            oracleConnectionType=OracleServiceName(oracleServiceName="testdb"),780        )781        assert expected_url == get_connection_url(oracle_conn_obj)782        # oracle with db & connection options783        expected_url = [784            "oracle+cx_oracle://admin:password@localhost:1541/testdb?test_key_2=test_value_2&test_key_1=test_value_1",785            "oracle+cx_oracle://admin:password@localhost:1541/testdb?test_key_1=test_value_1&test_key_2=test_value_2",786        ]787        oracle_conn_obj = OracleConnection(788            username="admin",789            password="password",790            hostPort="localhost:1541",791            scheme=OracleScheme.oracle_cx_oracle,792            oracleConnectionType=OracleDatabaseSchema(databaseSchema="testdb"),793            connectionOptions=dict(794                test_key_1="test_value_1", test_key_2="test_value_2"795            ),796        )797        assert get_connection_url(oracle_conn_obj) in expected_url798        # oracle with service name & connection options799        expected_url = [800            "oracle+cx_oracle://admin:password@localhost:1541/?service_name=testdb&test_key_2=test_value_2&test_key_1=test_value_1",801            "oracle+cx_oracle://admin:password@localhost:1541/?service_name=testdb&test_key_1=test_value_1&test_key_2=test_value_2",802        ]803        oracle_conn_obj = OracleConnection(804            username="admin",805            password="password",806            hostPort="localhost:1541",807            scheme=OracleScheme.oracle_cx_oracle,808            oracleConnectionType=OracleServiceName(oracleServiceName="testdb"),809            connectionOptions=dict(810                test_key_1="test_value_1", test_key_2="test_value_2"811            ),812        )...

source_connections.py

Source:source_connections.py

...117        )118        url = f"{url}?{params}"119    return url120@singledispatch121def get_connection_url(connection):122    """123    Single dispatch method to get the source connection url124    """125    raise NotImplementedError(126        f"Connection URL build not implemented for type {type(connection)}: {connection}"127    )128@get_connection_url.register(MariaDBConnection)129@get_connection_url.register(PostgresConnection)130@get_connection_url.register(RedshiftConnection)131@get_connection_url.register(MysqlConnection)132@get_connection_url.register(SalesforceConnection)133@get_connection_url.register(ClickhouseConnection)134@get_connection_url.register(SingleStoreConnection)135@get_connection_url.register(Db2Connection)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.