SNOW-2306184: config refactor - e2e fix attempt 3

Author: sfc-gh-mraba

tests_e2e/conftest.py

@@ -117,11 +117,14 @@ def isolate_environment_variables(monkeypatch):
     """
     Clear Snowflake-specific environment variables that could interfere with e2e tests.
     This ensures tests run in a clean environment and only use the config files they specify.
+    Exception: Keep INTEGRATION connection vars for e2e testing.
     """
-    # Clear all SNOWFLAKE_CONNECTIONS_* environment variables
+    # Clear all SNOWFLAKE_CONNECTIONS_* environment variables except INTEGRATION
     for env_var in list(os.environ.keys()):
         if env_var.startswith(("SNOWFLAKE_CONNECTIONS_", "SNOWSQL_")):
-            monkeypatch.delenv(env_var, raising=False)
+            # Preserve all INTEGRATION connection environment variables
+            if not env_var.startswith("SNOWFLAKE_CONNECTIONS_INTEGRATION_"):
+                monkeypatch.delenv(env_var, raising=False)
 
 
 def _create_venv(tmp_dir: Path) -> None:

tests_e2e/test_import_snowsql_connections.py

@@ -6,32 +6,37 @@
 from tests_e2e.conftest import subprocess_check_output, subprocess_run
 
 
-@pytest.fixture()
-def _assert_json_output_matches_snapshot(snapshot):
-    def f(cmd, stdin: Optional[str] = None):
-        output = subprocess_check_output(cmd, stdin)
-        parsed_json = json.loads(output)
-        snapshot.assert_match(json.dumps(parsed_json))
-
-    return f
-
-
-@pytest.mark.e2e
-def test_import_of_snowsql_connections(
-    snowcli, test_root_path, empty_config_file, _assert_json_output_matches_snapshot
-):
-    _assert_json_output_matches_snapshot(
+def _get_connections_list(snowcli, config_file) -> list:
+    """Helper function to get connections list as parsed JSON."""
+    output = subprocess_check_output(
         [
             snowcli,
             "--config-file",
-            empty_config_file,
+            config_file,
             "connection",
             "list",
             "--format",
             "json",
-        ],
+        ]
     )
+    return json.loads(output)
+
+
+def _find_connection(connections: list, name: str) -> Optional[dict]:
+    """Helper function to find a connection by name."""
+    for conn in connections:
+        if conn["connection_name"] == name:
+            return conn
+    return None
+
+
+@pytest.mark.e2e
+def test_import_of_snowsql_connections(snowcli, test_root_path, empty_config_file):
+    # Initially should have empty or minimal connections list
+    initial_connections = _get_connections_list(snowcli, empty_config_file)
+    initial_count = len(initial_connections)
 
+    # Import snowsql connections
     result = subprocess_run(
         [
             snowcli,
@@ -47,35 +52,49 @@ def test_import_of_snowsql_connections(
     )
     assert result.returncode == 0
 
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            empty_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ]
-    )
+    # After import, should have multiple connections
+    final_connections = _get_connections_list(snowcli, empty_config_file)
+
+    # Should have more connections than initially
+    assert len(final_connections) > initial_count
+
+    # Check that expected connections exist
+    connection_names = {conn["connection_name"] for conn in final_connections}
+    expected_names = {"snowsql1", "snowsql2", "snowsql3", "example", "default"}
+    assert expected_names.issubset(connection_names)
+
+    # Check specific connection details
+    snowsql1 = _find_connection(final_connections, "snowsql1")
+    assert snowsql1 is not None
+    assert snowsql1["parameters"]["account"] == "a1"
+    assert snowsql1["parameters"]["user"] == "u1"
+    assert snowsql1["parameters"]["host"] == "h1_override"  # From overriding config
+    assert snowsql1["is_default"] is False
+
+    snowsql2 = _find_connection(final_connections, "snowsql2")
+    assert snowsql2 is not None
+    assert snowsql2["parameters"]["account"] == "a2"
+    assert snowsql2["parameters"]["port"] == 1234
+    assert snowsql2["is_default"] is False
+
+    default_conn = _find_connection(final_connections, "default")
+    assert default_conn is not None
+    assert default_conn["parameters"]["account"] == "default_connection_account"
+    assert (
+        default_conn["parameters"]["database"] == "default_connection_database_override"
+    )  # From overriding config
+    assert default_conn["is_default"] is True
 
 
 @pytest.mark.e2e
 def test_import_prompt_for_different_default_connection_name_on_conflict(
-    snowcli, test_root_path, empty_config_file, _assert_json_output_matches_snapshot
+    snowcli, test_root_path, empty_config_file
 ):
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            empty_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ],
-    )
+    # Initially should have empty or minimal connections list
+    initial_connections = _get_connections_list(snowcli, empty_config_file)
+    initial_count = len(initial_connections)
 
+    # Import with different default connection name
     result = subprocess_run(
         [
             snowcli,
@@ -94,38 +113,39 @@ def test_import_prompt_for_different_default_connection_name_on_conflict(
     )
     assert result.returncode == 0
 
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            empty_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ]
-    )
+    # After import, snowsql2 should be the default
+    final_connections = _get_connections_list(snowcli, empty_config_file)
+
+    # Should have more connections than initially
+    assert len(final_connections) > initial_count
+
+    snowsql2 = _find_connection(final_connections, "snowsql2")
+    assert snowsql2 is not None
+    assert snowsql2["is_default"] is True
+
+    default_conn = _find_connection(final_connections, "default")
+    assert default_conn is not None
+    assert default_conn["is_default"] is False
 
 
 @pytest.mark.e2e
 def test_import_confirm_on_conflict_with_existing_cli_connection(
     snowcli,
     test_root_path,
     example_connection_config_file,
-    _assert_json_output_matches_snapshot,
 ):
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            example_connection_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ],
-    )
+    # Initially should have example and integration connections
+    initial_connections = _get_connections_list(snowcli, example_connection_config_file)
 
+    example_conn = _find_connection(initial_connections, "example")
+    assert example_conn is not None
+    assert example_conn["parameters"]["user"] == "u1"
+    assert example_conn["parameters"]["authenticator"] == "SNOWFLAKE_JWT"
+
+    integration_conn = _find_connection(initial_connections, "integration")
+    assert integration_conn is not None
+
+    # Import with confirmation (y)
     result = subprocess_run(
         [
             snowcli,
@@ -142,38 +162,32 @@ def test_import_confirm_on_conflict_with_existing_cli_connection(
     )
     assert result.returncode == 0
 
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            example_connection_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ],
-    )
+    # After import, example connection should be overwritten with snowsql data
+    final_connections = _get_connections_list(snowcli, example_connection_config_file)
+
+    example_conn = _find_connection(final_connections, "example")
+    assert example_conn is not None
+    assert example_conn["parameters"]["account"] == "accountname"
+    assert example_conn["parameters"]["user"] == "username"
+    # Should not have the old JWT authenticator
+    assert "authenticator" not in example_conn["parameters"]
 
 
 @pytest.mark.e2e
 def test_import_reject_on_conflict_with_existing_cli_connection(
     snowcli,
     test_root_path,
     example_connection_config_file,
-    _assert_json_output_matches_snapshot,
 ):
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            example_connection_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ],
-    )
+    # Initially should have example and integration connections
+    initial_connections = _get_connections_list(snowcli, example_connection_config_file)
+
+    example_conn = _find_connection(initial_connections, "example")
+    assert example_conn is not None
+    original_user = example_conn["parameters"]["user"]
+    original_auth = example_conn["parameters"]["authenticator"]
 
+    # Import with rejection (n)
     result = subprocess_run(
         [
             snowcli,
@@ -190,17 +204,18 @@ def test_import_reject_on_conflict_with_existing_cli_connection(
     )
     assert result.returncode == 0
 
-    _assert_json_output_matches_snapshot(
-        [
-            snowcli,
-            "--config-file",
-            example_connection_config_file,
-            "connection",
-            "list",
-            "--format",
-            "json",
-        ],
-    )
+    # After import, example connection should remain unchanged
+    final_connections = _get_connections_list(snowcli, example_connection_config_file)
+
+    example_conn = _find_connection(final_connections, "example")
+    assert example_conn is not None
+    assert example_conn["parameters"]["user"] == original_user
+    assert example_conn["parameters"]["authenticator"] == original_auth
+
+    # But other connections should still be imported
+    snowsql1 = _find_connection(final_connections, "snowsql1")
+    assert snowsql1 is not None
+    assert snowsql1["parameters"]["account"] == "a1"
 
 
 @pytest.mark.e2e
@@ -218,6 +233,7 @@ def test_connection_imported_from_snowsql(snowcli, test_root_path, empty_config_
     )
     assert result.returncode == 0
 
+    # Test that the imported integration connection works
     result = subprocess_run(
         [
             snowcli,