SNOW-2306184: config refactor - e2e fix attempt 4

Author: sfc-gh-mraba

tests_e2e/conftest.py

@@ -210,3 +210,31 @@ def example_connection_config_file(test_root_path, prepare_test_config_file):
     yield prepare_test_config_file(
         SecurePath(test_root_path) / "config" / "example_connection.toml"
     )
+
+
+@pytest.fixture
+def config_mode(request, monkeypatch):
+    """
+    Fixture to switch between legacy and config_ng modes.
+
+    When parameterized with ["legacy", "config_ng"], this fixture sets the
+    appropriate environment variable to enable/disable the new config system.
+    Each parameter value creates a separate test instance with its own snapshot.
+
+    Usage:
+        @pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
+        def test_something(config_mode, snapshot):
+            # Test runs twice: once with legacy, once with config_ng
+            # Each gets its own snapshot: test_something[legacy] and test_something[config_ng]
+            ...
+    """
+    mode = getattr(request, "param", "config_ng")  # default to config_ng
+
+    if mode == "config_ng":
+        # Enable new config system
+        monkeypatch.setenv("SNOWFLAKE_CLI_CONFIG_V2_ENABLED", "true")
+    else:
+        # Ensure new config system is disabled (legacy mode)
+        monkeypatch.delenv("SNOWFLAKE_CLI_CONFIG_V2_ENABLED", raising=False)
+
+    return mode

tests_e2e/test_import_snowsql_connections.py

@@ -1,14 +1,11 @@
-import json
-from typing import Optional
-
 import pytest
 
 from tests_e2e.conftest import subprocess_check_output, subprocess_run
 
 
-def _get_connections_list(snowcli, config_file) -> list:
-    """Helper function to get connections list as parsed JSON."""
-    output = subprocess_check_output(
+def _get_connections_list_output(snowcli, config_file) -> str:
+    """Helper function to get connections list output as string."""
+    return subprocess_check_output(
         [
             snowcli,
             "--config-file",
@@ -19,22 +16,17 @@ def _get_connections_list(snowcli, config_file) -> list:
             "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.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.e2e
-def test_import_of_snowsql_connections(snowcli, test_root_path, empty_config_file):
+def test_import_of_snowsql_connections(
+    snowcli, test_root_path, empty_config_file, snapshot, config_mode
+):
+    """Test connection import with both legacy and config_ng systems."""
     # Initially should have empty or minimal connections list
-    initial_connections = _get_connections_list(snowcli, empty_config_file)
-    initial_count = len(initial_connections)
+    initial_output = _get_connections_list_output(snowcli, empty_config_file)
+    assert initial_output == snapshot
 
     # Import snowsql connections
     result = subprocess_run(
@@ -53,46 +45,19 @@ def test_import_of_snowsql_connections(snowcli, test_root_path, empty_config_fil
     assert result.returncode == 0
 
     # 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
+    final_output = _get_connections_list_output(snowcli, empty_config_file)
+    assert final_output == snapshot
 
 
+@pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.e2e
 def test_import_prompt_for_different_default_connection_name_on_conflict(
-    snowcli, test_root_path, empty_config_file
+    snowcli, test_root_path, empty_config_file, snapshot, config_mode
 ):
+    """Test importing with different default connection name."""
     # Initially should have empty or minimal connections list
-    initial_connections = _get_connections_list(snowcli, empty_config_file)
-    initial_count = len(initial_connections)
+    initial_output = _get_connections_list_output(snowcli, empty_config_file)
+    assert initial_output == snapshot
 
     # Import with different default connection name
     result = subprocess_run(
@@ -114,36 +79,25 @@ def test_import_prompt_for_different_default_connection_name_on_conflict(
     assert result.returncode == 0
 
     # 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
+    final_output = _get_connections_list_output(snowcli, empty_config_file)
+    assert final_output == snapshot
 
 
+@pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.e2e
 def test_import_confirm_on_conflict_with_existing_cli_connection(
     snowcli,
     test_root_path,
     example_connection_config_file,
+    snapshot,
+    config_mode,
 ):
+    """Test import with confirmation on conflict."""
     # 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
+    initial_output = _get_connections_list_output(
+        snowcli, example_connection_config_file
+    )
+    assert initial_output == snapshot
 
     # Import with confirmation (y)
     result = subprocess_run(
@@ -163,29 +117,25 @@ def test_import_confirm_on_conflict_with_existing_cli_connection(
     assert result.returncode == 0
 
     # 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"]
+    final_output = _get_connections_list_output(snowcli, example_connection_config_file)
+    assert final_output == snapshot
 
 
+@pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.e2e
 def test_import_reject_on_conflict_with_existing_cli_connection(
     snowcli,
     test_root_path,
     example_connection_config_file,
+    snapshot,
+    config_mode,
 ):
+    """Test import with rejection on conflict."""
     # 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"]
+    initial_output = _get_connections_list_output(
+        snowcli, example_connection_config_file
+    )
+    assert initial_output == snapshot
 
     # Import with rejection (n)
     result = subprocess_run(
@@ -205,21 +155,17 @@ def test_import_reject_on_conflict_with_existing_cli_connection(
     assert result.returncode == 0
 
     # 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"
+    final_output = _get_connections_list_output(snowcli, example_connection_config_file)
+    assert final_output == snapshot
 
 
+@pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.e2e
-def test_connection_imported_from_snowsql(snowcli, test_root_path, empty_config_file):
+def test_connection_imported_from_snowsql(
+    snowcli, test_root_path, empty_config_file, config_mode
+):
+    """Test that imported connection works."""
     result = subprocess_run(
         [
             snowcli,

tests_integration/conftest.py

@@ -308,3 +308,32 @@ def enable_snowpark_glob_support_feature_flag():
 def global_setup(monkeypatch):
     width = 81 if IS_WINDOWS else 80
     monkeypatch.setenv("COLUMNS", str(width))
+
+
+@pytest.fixture
+def config_mode(request, monkeypatch):
+    """
+    Fixture to switch between legacy and config_ng modes.
+
+    When parameterized with ["legacy", "config_ng"], this fixture sets the
+    appropriate environment variable to enable/disable the new config system.
+    Each parameter value creates a separate test instance with its own snapshot.
+
+    Usage:
+        @pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
+        @pytest.mark.integration
+        def test_something(runner, config_mode, snapshot):
+            # Test runs twice: once with legacy, once with config_ng
+            # Each gets its own snapshot
+            ...
+    """
+    mode = getattr(request, "param", "config_ng")  # default to config_ng
+
+    if mode == "config_ng":
+        # Enable new config system
+        monkeypatch.setenv("SNOWFLAKE_CLI_CONFIG_V2_ENABLED", "true")
+    else:
+        # Ensure new config system is disabled (legacy mode)
+        monkeypatch.delenv("SNOWFLAKE_CLI_CONFIG_V2_ENABLED", raising=False)
+
+    return mode

tests_integration/plugin/test_broken_plugin.py

@@ -12,13 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json
-
 import pytest
 
 
+@pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.integration
-def test_broken_command_path_plugin(runner, test_root_path, _install_plugin, caplog):
+def test_broken_command_path_plugin(
+    runner, test_root_path, _install_plugin, caplog, snapshot, config_mode
+):
+    """Test broken plugin with both legacy and config_ng systems."""
     config_path = (
         test_root_path / "config" / "plugin_tests" / "broken_plugin_config.toml"
     )
@@ -34,16 +36,9 @@ def test_broken_command_path_plugin(runner, test_root_path, _install_plugin, cap
         in caplog.messages
     )
 
-    # Parse JSON output and check for test connection existence
-    connections = json.loads(result.output)
-
-    # Find the 'test' connection
-    test_connection = next(
-        (conn for conn in connections if conn["connection_name"] == "test"), None
-    )
-    assert test_connection is not None, "Expected 'test' connection not found in output"
-    assert test_connection["parameters"] == {"account": "test"}
-    assert test_connection["is_default"] is False
+    # Use snapshot to capture the output
+    # Each config_mode gets its own snapshot automatically
+    assert result.output == snapshot
 
 
 @pytest.fixture(scope="module")

tests_integration/plugin/test_failing_plugin.py

@@ -12,13 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json
-
 import pytest
 
 
+@pytest.mark.parametrize("config_mode", ["legacy", "config_ng"], indirect=True)
 @pytest.mark.integration
-def test_failing_plugin(runner, test_root_path, _install_plugin, caplog):
+def test_failing_plugin(
+    runner, test_root_path, _install_plugin, caplog, snapshot, config_mode
+):
+    """Test failing plugin with both legacy and config_ng systems."""
     config_path = (
         test_root_path / "config" / "plugin_tests" / "failing_plugin_config.toml"
     )
@@ -32,16 +34,9 @@ def test_failing_plugin(runner, test_root_path, _install_plugin, caplog):
         in caplog.messages
     )
 
-    # Parse JSON output and check for test connection existence
-    connections = json.loads(result.output)
-
-    # Find the 'test' connection
-    test_connection = next(
-        (conn for conn in connections if conn["connection_name"] == "test"), None
-    )
-    assert test_connection is not None, "Expected 'test' connection not found in output"
-    assert test_connection["parameters"] == {"account": "test"}
-    assert test_connection["is_default"] is False
+    # Use snapshot to capture the output
+    # Each config_mode gets its own snapshot automatically
+    assert result.output == snapshot
 
 
 @pytest.fixture(scope="module")