#133 Addressed review comments

test/integration/test_sagemaker_extension_wrapper.py

@@ -8,7 +8,7 @@
     assert_connection_exists,
     assert_run_empty_udf,
     get_script_counts,
-    set_language_definition,
+    language_definition_context,
 )
 
 
@@ -23,17 +23,18 @@ def test_initialize_sme_extension(
     secrets.save(CKey.aws_access_key_id, "FAKEKEYIDDONTUSEIT")
     secrets.save(CKey.aws_secret_access_key, "FakeSecretAccessKeyDontTryToUseIt")
 
+    language_alias = 'PYTHON3_SME_TEST'
+
     with open_pyexasol_connection(secrets) as pyexasol_connection:
-        # Set pre-existing language definition with the chosen alias.
-        language_alias = 'PYTHON3_SME'
-        set_language_definition(language_alias, pyexasol_connection)
 
-        # Run the extension deployment.
-        initialize_sme_extension(secrets, language_alias=language_alias)
+        with language_definition_context(pyexasol_connection, language_alias):
+
+            # Run the extension deployment.
+            initialize_sme_extension(secrets, language_alias=language_alias)
 
-        activate_languages(pyexasol_connection, secrets)
-        assert_run_empty_udf("PYTHON3_SME", pyexasol_connection, secrets)
-        script_counts = get_script_counts(pyexasol_connection, secrets)
-        assert script_counts["SCRIPTING"] >= 4
-        assert script_counts["UDF"] >= 5
-        assert_connection_exists(secrets.get(CKey.sme_aws_connection), pyexasol_connection)
+            activate_languages(pyexasol_connection, secrets)
+            assert_run_empty_udf(language_alias, pyexasol_connection, secrets)
+            script_counts = get_script_counts(pyexasol_connection, secrets)
+            assert script_counts["SCRIPTING"] >= 4
+            assert script_counts["UDF"] >= 5
+            assert_connection_exists(secrets.get(CKey.sme_aws_connection), pyexasol_connection)

test/integration/test_transformers_extension_wrapper.py

@@ -11,31 +11,28 @@
     assert_connection_exists,
     assert_run_empty_udf,
     get_script_counts,
-    set_language_definition,
+    language_definition_context,
 )
 
 
 @pytest.mark.skip(reason="The expected functionality is not yet implemented in the Transformers Extension")
 def test_initialize_te_extension(
-    request: FixtureRequest,
     secrets: Secrets,
     setup_itde
 ):
-    test_name: str = request.node.name
-    language_alias = f"PYTHON3_TE_{test_name.upper()}"
+    language_alias = f"PYTHON3_TE_TEST"
     secrets.save(CKey.huggingface_token, "abc")
 
     with open_pyexasol_connection(secrets) as pyexasol_connection:
 
-        # Set pre-existing language definition with the chosen alias.
-        set_language_definition(language_alias, pyexasol_connection)
+        with language_definition_context(pyexasol_connection, language_alias):
 
-        # Run the extension deployment.
-        initialize_te_extension(secrets, language_alias=language_alias)
+            # Run the extension deployment.
+            initialize_te_extension(secrets, language_alias=language_alias)
 
-        activate_languages(pyexasol_connection, secrets)
-        assert_run_empty_udf(language_alias, pyexasol_connection, secrets)
-        script_counts = get_script_counts(pyexasol_connection, secrets)
-        assert script_counts["UDF"] > 5
-        assert_connection_exists(secrets.get(CKey.te_bfs_connection), pyexasol_connection)
-        assert_connection_exists(secrets.get(CKey.te_hf_connection), pyexasol_connection)
+            activate_languages(pyexasol_connection, secrets)
+            assert_run_empty_udf(language_alias, pyexasol_connection, secrets)
+            script_counts = get_script_counts(pyexasol_connection, secrets)
+            assert script_counts["UDF"] > 5
+            assert_connection_exists(secrets.get(CKey.te_bfs_connection), pyexasol_connection)
+            assert_connection_exists(secrets.get(CKey.te_hf_connection), pyexasol_connection)

test/unit/test_extension_wrapper_common.py

@@ -17,13 +17,13 @@ def filled_secrets(secrets) -> Secrets:
     secrets.save(CKey.db_host_name, DB_HOST)
     secrets.save(CKey.db_port, '8888')
     secrets.save(CKey.db_user, 'user')
-    secrets.save(CKey.db_password, 'password')
+    secrets.save(CKey.db_password, 'my_db_password')
     secrets.save(CKey.bfs_port, '6666')
     secrets.save(CKey.bfs_encryption, 'True')
     secrets.save(CKey.bfs_service, 'bfsdefault')
     secrets.save(CKey.bfs_bucket, 'default')
     secrets.save(CKey.bfs_user, 'user'),
-    secrets.save(CKey.bfs_password, 'password')
+    secrets.save(CKey.bfs_password, 'my_bfs_password')
     return secrets
 
 
@@ -69,7 +69,7 @@ def test_bucketfs_credentials_default(mock_connect, filled_secrets):
         ['username'], ['user']
     ))
     validate_params(query_params['BUCKETFS_PASSWORD'], (
-        ['password'], ['password']
+        ['password'], ['my_bfs_password']
     ))
 
 

test/utils/integration_test_utils.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
 import textwrap
-from typing import Dict
+from contextlib import contextmanager
 import pytest
 
 from pyexasol import ExaConnection
@@ -71,7 +72,7 @@ def run(ctx):
 
 def get_script_counts(
     pyexasol_connection: ExaConnection, secrets: Secrets
-) -> Dict[str, int]:
+) -> dict[str, int]:
     """
     Returns numbers of installed scripts of different types.
     """
@@ -101,10 +102,33 @@ def assert_connection_exists(
     assert result
 
 
-def set_language_definition(language_alias: str, pyexasol_connection: ExaConnection
-                            ) -> None:
-    for alter_type in ['SYSTEM', 'SESSION']:
-        sql = (f"ALTER {alter_type} SET SCRIPT_LANGUAGES='PYTHON=builtin_python "
-               "R=builtin_r JAVA=builtin_java PYTHON3=builtin_python3 "
-               f"{language_alias}=builtin_python3';")
+@contextmanager
+def language_definition_context(pyexasol_connection: ExaConnection,
+                                language_alias: str | None = None) -> None:
+    """
+    A context manager that preserves the current language definitions at both
+    SESSION and SYSTEM levels. Optionally creates a definition for the specified
+    alias to test the ability to override an existing definition.
+    """
+    def alter_language_settings(alter_type: str, lang_definition: str):
+        sql = f"ALTER {alter_type} SET SCRIPT_LANGUAGES='{lang_definition}';"
         pyexasol_connection.execute(sql)
+
+    # Remember the current language settings.
+    alter_types = ['SYSTEM', 'SESSION']
+    sql0 = (f"""SELECT "{', '.join(alter_type + '_VALUE' for alter_type in alter_types)}" """
+            "FROM SYS.EXA_PARAMETERS WHERE PARAMETER_NAME='SCRIPT_LANGUAGES';")
+    current_definitions = pyexasol_connection.execute(sql0).fetchall()[0]
+
+    for alter_type in alter_types:
+        # Creates a trivial language definition for the specified alias.
+        if language_alias:
+            lang_def = ('PYTHON=builtin_python R=builtin_r JAVA=builtin_java '
+                        f'PYTHON3=builtin_python3 {language_alias}=builtin_python3')
+            alter_language_settings(alter_type, lang_def)
+    try:
+        yield
+    finally:
+        # Restore language settings.
+        for alter_type, lang_def in zip(alter_types, current_definitions):
+            alter_language_settings(alter_type, lang_def)