#75 Finished the bucketfs_conn_object tests

exasol/python_extension_common/connections/bucketfs_location.py

@@ -1,5 +1,6 @@
 from typing import Any
 from enum import Enum, auto
+from dataclasses import dataclass
 import json
 import pyexasol     # type: ignore
 import exasol.bucketfs as bfs   # type: ignore
@@ -107,22 +108,31 @@ def create_bucketfs_location(**kwargs) -> bfs.path.PathLike:
         return bfs.path.build_path(**_convert_saas_bfs_params(kwargs))
 
 
+@dataclass
+class ConnObjectData:
+    """
+    This is not a connection object. It's just a structure to keep together the data
+    required for creating a BucketFs connection object. Useful for testing.
+    """
+    to: str
+    user: str
+    password: str
+
+
 def _to_json_str(bucketfs_params: dict[str, Any], selected: list[str]) -> str:
     filtered_kwargs = {k: v for k, v in bucketfs_params.items()
                        if (k in selected) and (v is not None)}
     return json.dumps(filtered_kwargs)
 
 
-def _write_bucketfs_conn_object(pyexasol_connection: pyexasol.ExaConnection,
-                                conn_name: str,
-                                conn_to: str,
-                                conn_user: str,
-                                conn_password: str) -> None:
+def write_bucketfs_conn_object(pyexasol_connection: pyexasol.ExaConnection,
+                               conn_name: str,
+                               conn_obj: ConnObjectData) -> None:
 
     query = (f"CREATE OR REPLACE  CONNECTION {conn_name} "
-             f"TO '{conn_to}' "
-             f"USER '{conn_user}' "
-             f"IDENTIFIED BY '{conn_password}'")
+             f"TO '{conn_obj.to}' "
+             f"USER '{conn_obj.user}' "
+             f"IDENTIFIED BY '{conn_obj.password}'")
     pyexasol_connection.execute(query)
 
 
@@ -146,8 +156,8 @@ def create_bucketfs_conn_object_onprem(pyexasol_connection: pyexasol.ExaConnecti
     conn_user = _to_json_str(bucketfs_params, ['username'])
     conn_password = _to_json_str(bucketfs_params, ['password'])
 
-    _write_bucketfs_conn_object(pyexasol_connection, conn_name,
-                                conn_to, conn_user, conn_password)
+    write_bucketfs_conn_object(pyexasol_connection, conn_name,
+                               ConnObjectData(conn_to, conn_user, conn_password))
 
 
 def create_bucketfs_conn_object_saas(pyexasol_connection: pyexasol.ExaConnection,
@@ -169,8 +179,8 @@ def create_bucketfs_conn_object_saas(pyexasol_connection: pyexasol.ExaConnection
     conn_user = _to_json_str(bucketfs_params, ['account_id', 'database_id'])
     conn_password = _to_json_str(bucketfs_params, ['pat'])
 
-    _write_bucketfs_conn_object(pyexasol_connection, conn_name,
-                                conn_to, conn_user, conn_password)
+    write_bucketfs_conn_object(pyexasol_connection, conn_name,
+                               ConnObjectData(conn_to, conn_user, conn_password))
 
 
 def create_bucketfs_conn_object(conn_name: str, **kwargs) -> None:

test/integration/connections/test_bucketfs_location.py

@@ -1,13 +1,20 @@
 from typing import Any
 from urllib.parse import urlparse
+from unittest.mock import patch
+from contextlib import contextmanager
+
+import pyexasol
 import pytest
 import exasol.bucketfs as bfs
 
 from exasol.python_extension_common.cli.std_options import StdParams
 from exasol.python_extension_common.connections.bucketfs_location import (
     create_bucketfs_location,
     create_bucketfs_conn_object,
-    create_bucketfs_location_from_conn_object)
+    create_bucketfs_location_from_conn_object,
+    ConnObjectData)
+
+TEST_FILE_CONTENT = b'A rose by any other name would smell as sweet.'
 
 
 @pytest.fixture(scope='session')
@@ -44,19 +51,41 @@ def saas_params_with_id(saas_host,
 
 @pytest.fixture(scope='session')
 def saas_params_with_name(saas_params_with_id,
-                          database_name):
+                          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
 
 
-def validate_bfs_path(bfs_path: bfs.path.PathLike) -> None:
-    file_content = b'A rose by any other name would smell as sweet.'
-    bfs_path.write(file_content)
-    data_back = b''.join(bfs_path.read())
-    bfs_path.rm()
-    assert data_back == file_content
+@pytest.fixture
+def rubbish_params() -> dict[str, Any]:
+    return {
+        StdParams.bucketfs_name.name: 'bfsdefault',
+        StdParams.bucket.name: 'default',
+        StdParams.saas_url.name: 'my_saas_url',
+    }
+
+
+@contextmanager
+def write_test_file(bfs_path: bfs.path.PathLike) -> None:
+    bfs_path.write(TEST_FILE_CONTENT)
+    try:
+        yield
+    finally:
+        bfs_path.rm()
+
+
+def validate_test_file(bfs_path: bfs.path.PathLike) -> None:
+    file_content = b''.join(bfs_path.read())
+    assert file_content == TEST_FILE_CONTENT
+
+
+def validate_conn_object(pyexasol_connection: pyexasol.ExaConnection,
+                         conn_name: str,
+                         conn_obj: ConnObjectData):
+    bfs_path = create_bucketfs_location_from_conn_object(conn_obj)
+    validate_test_file(bfs_path)
 
 
 def test_create_bucketfs_location_onprem(use_onprem,
@@ -65,7 +94,8 @@ def test_create_bucketfs_location_onprem(use_onprem,
         pytest.skip("The test is not configured to use ITDE.")
 
     bfs_path = create_bucketfs_location(**onprem_params)
-    validate_bfs_path(bfs_path)
+    with write_test_file(bfs_path):
+        validate_test_file(bfs_path)
 
 
 def test_create_bucketfs_location_saas_db_id(use_saas,
@@ -74,7 +104,8 @@ def test_create_bucketfs_location_saas_db_id(use_saas,
         pytest.skip("The test is not configured to use SaaS.")
 
     bfs_path = create_bucketfs_location(**saas_params_with_id)
-    validate_bfs_path(bfs_path)
+    with write_test_file(bfs_path):
+        validate_test_file(bfs_path)
 
 
 def test_create_bucketfs_location_saas_db_name(use_saas,
@@ -83,25 +114,54 @@ def test_create_bucketfs_location_saas_db_name(use_saas,
         pytest.skip("The test is not configured to use SaaS.")
 
     bfs_path = create_bucketfs_location(**saas_params_with_name)
-    validate_bfs_path(bfs_path)
+    with write_test_file(bfs_path):
+        validate_test_file(bfs_path)
 
 
-def test_create_bucketfs_location_error():
-    kwargs = {
-        StdParams.bucketfs_name.name: 'bfsdefault',
-        StdParams.bucket.name: 'default',
-        StdParams.saas_url.name: 'my_saas_url',
-    }
+def test_create_bucketfs_location_error(rubbish_params):
     with pytest.raises(ValueError):
-        create_bucketfs_location(**kwargs)
+        create_bucketfs_location(**rubbish_params)
 
 
-def test_create_bucketfs_conn_object_onprem(use_onprem,
+@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):
     if not use_onprem:
         pytest.skip("The test is not configured to use ITDE.")
 
-    conn_name = 'ONPREM_TEST_BFS'
-    create_bucketfs_conn_object(conn_name=conn_name, **onprem_params)
-    # bfs_path = create_bucketfs_location_from_conn_object()
-    # validate_bfs_path( bfs_path)
+    write_conn_object_mock.side_effect = validate_conn_object
+    bfs_path = create_bucketfs_location(**onprem_params)
+    with write_test_file(bfs_path):
+        create_bucketfs_conn_object(conn_name='ONPREM_TEST_BFS', **onprem_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):
+    if not use_saas:
+        pytest.skip("The test is not configured to use SaaS.")
+
+    write_conn_object_mock.side_effect = validate_conn_object
+    bfs_path = create_bucketfs_location(**saas_params_with_id)
+    with write_test_file(bfs_path):
+        create_bucketfs_conn_object(conn_name='SAAS_TEST_BFS_WITH_ID', **saas_params_with_id)
+
+
+@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):
+    if not use_saas:
+        pytest.skip("The test is not configured to use SaaS.")
+
+    write_conn_object_mock.side_effect = validate_conn_object
+    bfs_path = create_bucketfs_location(**saas_params_with_name)
+    with write_test_file(bfs_path):
+        create_bucketfs_conn_object(conn_name='SAAS_TEST_BFS_WITH_NAME', **saas_params_with_name)
+
+
+def test_create_bucketfs_conn_object_error(rubbish_params):
+    with pytest.raises(ValueError):
+        create_bucketfs_conn_object(conn_name='WHATEVER', **rubbish_params)