#75 Fixed the bucketfs_location tests

test/integration/conftest.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
+from typing import Any
 import pytest
 import click
 import requests
+from urllib.parse import urlparse
 from contextlib import ExitStack, contextmanager
 import pyexasol
 import exasol.bucketfs as bfs
 
+from exasol.python_extension_common.cli.std_options import StdParams
 from exasol.python_extension_common.deployment.language_container_deployer import (
     LanguageContainerDeployer,
 )
@@ -95,3 +98,53 @@ def create_deployer(create_test_schema: bool = False, open_test_schema: bool = T
                 open_schema(pyexasol_connection, db_schema)
             yield LanguageContainerDeployer(pyexasol_connection, language_alias, bucketfs_path)
     return create_deployer
+
+
+@pytest.fixture(scope='session')
+def onprem_db_params(backend_aware_onprem_database,
+                     exasol_config) -> dict[str, Any]:
+    return {
+        StdParams.dsn.name: f'{exasol_config.host}:{exasol_config.port}',
+        StdParams.db_user.name: exasol_config.username,
+        StdParams.db_password.name: exasol_config.password,
+        StdParams.use_ssl_cert_validation.name: False
+    }
+
+
+@pytest.fixture(scope='session')
+def onprem_bfs_params(backend_aware_onprem_database,
+                      bucketfs_config) -> dict[str, Any]:
+    parsed_url = urlparse(bucketfs_config.url)
+    host, port = parsed_url.netloc.split(":")
+    return {
+        StdParams.bucketfs_host.name: host,
+        StdParams.bucketfs_port.name: port,
+        StdParams.bucketfs_use_https.name: parsed_url.scheme.lower() == 'https',
+        StdParams.bucketfs_user.name: bucketfs_config.username,
+        StdParams.bucketfs_password.name: bucketfs_config.password,
+        StdParams.bucketfs_name.name: 'bfsdefault',
+        StdParams.bucket.name: 'default',
+        StdParams.use_ssl_cert_validation.name: False
+    }
+
+
+@pytest.fixture(scope='session')
+def saas_params_id(saas_host,
+                   saas_pat,
+                   saas_account_id,
+                   backend_aware_saas_database_id) -> dict[str, Any]:
+    return {
+        StdParams.saas_url.name: saas_host,
+        StdParams.saas_account_id.name: saas_account_id,
+        StdParams.saas_database_id.name: backend_aware_saas_database_id,
+        StdParams.saas_token.name: saas_pat,
+    }
+
+
+@pytest.fixture(scope='session')
+def saas_params_name(saas_params_id,
+                     database_name) -> dict[str, Any]:
+    saas_params = dict(saas_params_id)
+    saas_params.pop(StdParams.saas_database_id.name)
+    saas_params[StdParams.saas_database_name.name] = database_name
+    return saas_params

test/integration/connections/test_bucketfs_location.py

@@ -1,5 +1,4 @@
 from typing import Any
-from urllib.parse import urlparse
 from unittest.mock import patch
 from contextlib import contextmanager
 
@@ -17,45 +16,6 @@
 TEST_FILE_CONTENT = b'A rose by any other name would smell as sweet.'
 
 
-@pytest.fixture(scope='session')
-def onprem_params(backend_aware_onprem_database,
-                  bucketfs_config) -> dict[str, Any]:
-    parsed_url = urlparse(bucketfs_config.url)
-    host, port = parsed_url.netloc.split(":")
-    return {
-        StdParams.bucketfs_host.name: host,
-        StdParams.bucketfs_port.name: port,
-        StdParams.bucketfs_use_https.name: parsed_url.scheme.lower() == 'https',
-        StdParams.bucketfs_user.name: bucketfs_config.username,
-        StdParams.bucketfs_password.name: bucketfs_config.password,
-        StdParams.bucketfs_name.name: 'bfsdefault',
-        StdParams.bucket.name: 'default',
-        StdParams.use_ssl_cert_validation.name: False
-    }
-
-
-@pytest.fixture(scope='session')
-def saas_params_with_id(saas_host,
-                        saas_pat,
-                        saas_account_id,
-                        backend_aware_saas_database_id) -> dict[str, Any]:
-    return {
-        StdParams.saas_url.name: saas_host,
-        StdParams.saas_account_id.name: saas_account_id,
-        StdParams.saas_database_id.name: backend_aware_saas_database_id,
-        StdParams.saas_token.name: saas_pat,
-    }
-
-
-@pytest.fixture(scope='session')
-def saas_params_with_name(saas_params_with_id,
-                          database_name) -> dict[str, Any]:
-    saas_params = dict(saas_params_with_id)
-    saas_params.pop(StdParams.saas_database_id.name)
-    saas_params[StdParams.saas_database_name.name] = database_name
-    return saas_params
-
-
 @pytest.fixture
 def rubbish_params() -> dict[str, Any]:
     return {
@@ -90,34 +50,34 @@ def validate_conn_object(pyexasol_connection: pyexasol.ExaConnection,
 
 
 def test_create_bucketfs_location_onprem(use_onprem,
-                                         onprem_params):
+                                         onprem_bfs_params):
     if not use_onprem:
         pytest.skip("The test is not configured to use ITDE.")
 
     extra_params = {StdParams.path_in_bucket.name: 'test_create_location'}
-    bfs_path = create_bucketfs_location(**onprem_params, **extra_params)
+    bfs_path = create_bucketfs_location(**onprem_bfs_params, **extra_params)
     with write_test_file(bfs_path):
         validate_test_file(bfs_path)
 
 
 def test_create_bucketfs_location_saas_db_id(use_saas,
-                                             saas_params_with_id):
+                                             saas_params_id):
     if not use_saas:
         pytest.skip("The test is not configured to use SaaS.")
 
     extra_params = {StdParams.path_in_bucket.name: 'test_create_location_with_id'}
-    bfs_path = create_bucketfs_location(**saas_params_with_id, **extra_params)
+    bfs_path = create_bucketfs_location(**saas_params_id, **extra_params)
     with write_test_file(bfs_path):
         validate_test_file(bfs_path)
 
 
 def test_create_bucketfs_location_saas_db_name(use_saas,
-                                               saas_params_with_name):
+                                               saas_params_name):
     if not use_saas:
         pytest.skip("The test is not configured to use SaaS.")
 
     extra_params = {StdParams.path_in_bucket.name: 'test_create_location_with_name'}
-    bfs_path = create_bucketfs_location(**saas_params_with_name, **extra_params)
+    bfs_path = create_bucketfs_location(**saas_params_name, **extra_params)
     with write_test_file(bfs_path):
         validate_test_file(bfs_path)
 
@@ -130,46 +90,47 @@ def test_create_bucketfs_location_error(rubbish_params):
 @patch('exasol.python_extension_common.connections.bucketfs_location.write_bucketfs_conn_object')
 def test_create_bucketfs_conn_object_onprem(write_conn_object_mock,
                                             use_onprem,
-                                            onprem_params):
+                                            onprem_db_params,
+                                            onprem_bfs_params):
     if not use_onprem:
         pytest.skip("The test is not configured to use ITDE.")
 
     write_conn_object_mock.side_effect = validate_conn_object
     extra_params = {StdParams.path_in_bucket.name: 'test_create_conn_object'}
-    bfs_path = create_bucketfs_location(**onprem_params, **extra_params)
+    bfs_path = create_bucketfs_location(**onprem_bfs_params, **extra_params)
     with write_test_file(bfs_path):
         create_bucketfs_conn_object(conn_name='ONPREM_TEST_BFS',
-                                    **onprem_params, **extra_params)
+                                    **onprem_db_params, **onprem_bfs_params, **extra_params)
 
 
 @patch('exasol.python_extension_common.connections.bucketfs_location.write_bucketfs_conn_object')
 def test_create_bucketfs_conn_object_saas_db_id(write_conn_object_mock,
                                                 use_saas,
-                                                saas_params_with_id):
+                                                saas_params_id):
     if not use_saas:
         pytest.skip("The test is not configured to use SaaS.")
 
     write_conn_object_mock.side_effect = validate_conn_object
     extra_params = {StdParams.path_in_bucket.name: 'test_create_conn_object_with_id'}
-    bfs_path = create_bucketfs_location(**saas_params_with_id, **extra_params)
+    bfs_path = create_bucketfs_location(**saas_params_id, **extra_params)
     with write_test_file(bfs_path):
-        create_bucketfs_conn_object(conn_name='SAAS_TEST_BFS_WITH_ID',
-                                    **saas_params_with_id, **extra_params)
+        create_bucketfs_conn_object(conn_name='SAAS_TEST_BFS_ID',
+                                    **saas_params_id, **extra_params)
 
 
 @patch('exasol.python_extension_common.connections.bucketfs_location.write_bucketfs_conn_object')
 def test_create_bucketfs_conn_object_saas_db_name(write_conn_object_mock,
                                                   use_saas,
-                                                  saas_params_with_name):
+                                                  saas_params_name):
     if not use_saas:
         pytest.skip("The test is not configured to use SaaS.")
 
     write_conn_object_mock.side_effect = validate_conn_object
     extra_params = {StdParams.path_in_bucket.name: 'test_create_conn_object_with_name'}
-    bfs_path = create_bucketfs_location(**saas_params_with_name, **extra_params)
+    bfs_path = create_bucketfs_location(**saas_params_name, **extra_params)
     with write_test_file(bfs_path):
-        create_bucketfs_conn_object(conn_name='SAAS_TEST_BFS_WITH_NAME',
-                                    **saas_params_with_name, **extra_params)
+        create_bucketfs_conn_object(conn_name='SAAS_TEST_BFS_NAME',
+                                    **saas_params_name, **extra_params)
 
 
 def test_create_bucketfs_conn_object_error(rubbish_params):

test/integration/connections/test_pyexasol_connection.py

@@ -11,55 +11,29 @@ def validate_connection(conn: pyexasol.ExaConnection) -> None:
 
 
 def test_open_pyexasol_connection_onprem(use_onprem,
-                                         backend_aware_onprem_database,
-                                         exasol_config):
+                                         onprem_db_params):
     if not use_onprem:
         pytest.skip("The test is not configured to use ITDE.")
 
-    kwargs = {
-        StdParams.dsn.name: f'{exasol_config.host}:{exasol_config.port}',
-        StdParams.db_user.name: exasol_config.username,
-        StdParams.db_password.name: exasol_config.password,
-        StdParams.use_ssl_cert_validation.name: False
-    }
-    with open_pyexasol_connection(**kwargs) as conn:
+    with open_pyexasol_connection(**onprem_db_params) as conn:
         validate_connection(conn)
 
 
 def test_open_pyexasol_connection_saas_db_id(use_saas,
-                                             saas_host,
-                                             saas_pat,
-                                             saas_account_id,
-                                             backend_aware_saas_database_id):
+                                             saas_params_id):
     if not use_saas:
         pytest.skip("The test is not configured to use SaaS.")
 
-    kwargs = {
-        StdParams.saas_url.name: saas_host,
-        StdParams.saas_account_id.name: saas_account_id,
-        StdParams.saas_database_id.name: backend_aware_saas_database_id,
-        StdParams.saas_token.name: saas_pat
-    }
-    with open_pyexasol_connection(**kwargs) as conn:
+    with open_pyexasol_connection(**saas_params_id) as conn:
         validate_connection(conn)
 
 
 def test_open_pyexasol_connection_saas_db_name(use_saas,
-                                               saas_host,
-                                               saas_pat,
-                                               saas_account_id,
-                                               backend_aware_saas_database_id,
-                                               database_name):
+                                               saas_params_name):
     if not use_saas:
         pytest.skip("The test is not configured to use SaaS.")
 
-    kwargs = {
-        StdParams.saas_url.name: saas_host,
-        StdParams.saas_account_id.name: saas_account_id,
-        StdParams.saas_database_name.name: database_name,
-        StdParams.saas_token.name: saas_pat
-    }
-    with open_pyexasol_connection(**kwargs) as conn:
+    with open_pyexasol_connection(**saas_params_name) as conn:
         validate_connection(conn)