SNOW-2306184: config refactor - connections.toml legacy behaviour

Author: sfc-gh-mraba

src/snowflake/cli/_plugins/connection/commands.py

@@ -108,12 +108,12 @@ def list_connections(
     Lists configured connections.
     """
     from snowflake.cli.api.config_provider import (
-        _is_alternative_config_enabled,
         get_config_provider_singleton,
+        is_alternative_config_enabled,
     )
 
     # Use provider directly for config_ng to pass the flag
-    if _is_alternative_config_enabled():
+    if is_alternative_config_enabled():
         provider = get_config_provider_singleton()
         connections = provider.get_all_connections(include_env_connections=all_sources)
     else:

src/snowflake/cli/api/config.py

@@ -322,35 +322,17 @@ def config_section_exists(*path) -> bool:
 
 
 def get_all_connections() -> dict[str, ConnectionConfig]:
-    # Use config provider if available
-    try:
-        from snowflake.cli.api.config_provider import get_config_provider_singleton
+    from snowflake.cli.api.config_provider import get_config_provider_singleton
 
-        provider = get_config_provider_singleton()
-        return provider.get_all_connections()
-    except Exception:
-        # Fall back to legacy implementation
-        return {
-            k: ConnectionConfig.from_dict(connection_dict)
-            for k, connection_dict in get_config_section("connections").items()
-        }
+    provider = get_config_provider_singleton()
+    return provider.get_all_connections()
 
 
 def get_connection_dict(connection_name: str) -> dict:
-    # Use config provider if available
-    try:
-        from snowflake.cli.api.config_provider import get_config_provider_singleton
+    from snowflake.cli.api.config_provider import get_config_provider_singleton
 
-        provider = get_config_provider_singleton()
-        return provider.get_connection_dict(connection_name)
-    except Exception:
-        # Fall back to legacy implementation
-        try:
-            return get_config_section(CONNECTIONS_SECTION, connection_name)
-        except KeyError:
-            raise MissingConfigurationError(
-                f"Connection {connection_name} is not configured"
-            )
+    provider = get_config_provider_singleton()
+    return provider.get_connection_dict(connection_name)
 
 
 def get_default_connection_name() -> str:

src/snowflake/cli/api/config_ng/resolver.py

@@ -349,7 +349,8 @@ def resolve(self, key: Optional[str] = None, default: Any = None) -> Dict[str, A
         1. Iterate sources in order (lowest to highest priority)
         2. Record all discovered values in history
         3. For connection keys (connections.{name}.{param}):
-           - Merge connection-by-connection: later sources extend/overwrite individual params
+           - If connections.toml defines a connection, it REPLACES cli_config_toml only
+           - SnowSQL config, environment vars, and CLI parameters still override
         4. For flat keys: later sources overwrite earlier sources
         5. Mark which value was selected
         6. Return final resolved values
@@ -365,6 +366,19 @@ def resolve(self, key: Optional[str] = None, default: Any = None) -> Dict[str, A
         # Track connection values separately for intelligent merging
         connections: Dict[str, Dict[str, ConfigValue]] = defaultdict(dict)
 
+        # Identify sources that connections.toml replaces
+        # connections.toml only replaces cli_config_toml, not SnowSQL config
+        cli_config_source = "cli_config_toml"
+        connections_file_source = None
+        connections_to_replace: set[str] = set()
+
+        # First pass: find connections.toml and identify connections to replace
+        for source in self._sources:
+            if hasattr(source, "is_connections_file") and source.is_connections_file:
+                connections_file_source = source
+                connections_to_replace = source.get_defined_connections()
+                break
+
         # Process sources in order (first = lowest priority, last = highest)
         for source in self._sources:
             try:
@@ -384,6 +398,22 @@ def resolve(self, key: Optional[str] = None, default: Any = None) -> Dict[str, A
                             param = parts[2]
                             param_key = f"connections.{conn_name}.{param}"
 
+                            # Replacement logic: Skip cli_config_toml if connection is in connections.toml
+                            # SnowSQL config is NOT replaced by connections.toml
+                            is_cli_config = source.source_name == cli_config_source
+                            connection_in_connections_toml = (
+                                conn_name in connections_to_replace
+                            )
+
+                            if is_cli_config and connection_in_connections_toml:
+                                # Skip this value - connections.toml replaces cli_config_toml
+                                log.debug(
+                                    "Skipping %s from %s (replaced by connections.toml)",
+                                    param_key,
+                                    source.source_name,
+                                )
+                                continue
+
                             # Merge at parameter level: later source overwrites/extends
                             connections[conn_name][param_key] = config_value
                     else:

src/snowflake/cli/api/config_ng/sources.py

@@ -312,6 +312,42 @@ def __init__(self):
     def source_name(self) -> str:
         return "connections_toml"
 
+    @property
+    def is_connections_file(self) -> bool:
+        """Mark this as the dedicated connections file source."""
+        return True
+
+    def get_defined_connections(self) -> set[str]:
+        """
+        Return set of connection names that are defined in connections.toml.
+        This is used by the resolver to implement replacement behavior.
+        """
+        if not self._file_path.exists():
+            return set()
+
+        try:
+            with open(self._file_path, "rb") as f:
+                data = tomllib.load(f)
+
+            connection_names = set()
+
+            # Check for direct connection sections (legacy format)
+            for section_name, section_data in data.items():
+                if isinstance(section_data, dict) and section_name != "connections":
+                    connection_names.add(section_name)
+
+            # Check for nested [connections] section format
+            connections_section = data.get("connections", {})
+            if isinstance(connections_section, dict):
+                for conn_name in connections_section.keys():
+                    connection_names.add(conn_name)
+
+            return connection_names
+
+        except Exception as e:
+            log.debug("Failed to read connections.toml: %s", e)
+            return set()
+
     def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
         """
         Read connections.toml if it exists.

src/snowflake/cli/api/config_provider.py

@@ -177,10 +177,14 @@ def get_connection_dict(self, connection_name: str) -> dict:
             )
 
     def get_all_connections(self, include_env_connections: bool = False) -> dict:
-        from snowflake.cli.api.config import get_all_connections
+        from snowflake.cli.api.config import ConnectionConfig, get_config_section
 
         # Legacy provider ignores the flag since it never had env connections
-        return get_all_connections()
+        connections = get_config_section("connections")
+        return {
+            name: ConnectionConfig.from_dict(self._transform_private_key_raw(config))
+            for name, config in connections.items()
+        }
 
 
 class AlternativeConfigProvider(ConfigProvider):

tests/config_ng/test_configuration.py

@@ -195,8 +195,8 @@ def test_complete_7_level_chain(config_ng_setup):
         # Level 3 provides warehouse
         assert conn["warehouse"] == "level3-wh"
 
-        # Level 2 provides password
-        assert conn["password"] == "level2-pass"
+        # Level 2 (cli_config) is skipped because connections.toml defines this connection
+        # password from cli_config is NOT present
 
         # Level 1 provides user
         assert conn["user"] == "level1-user"
@@ -444,7 +444,7 @@ def test_all_file_sources_precedence(config_ng_setup):
 
         expected = {
             "account": "from-connections",  # Level 3 wins
-            "user": "cli-user",  # Level 2 wins
+            "user": "snowsql-user",  # Level 1 only (cli_config skipped)
             "warehouse": "snowsql-warehouse",  # Level 1 only source
             "password": "connections-pass",  # Level 3 wins
         }
@@ -575,7 +575,7 @@ def test_all_files_plus_snowsql_env(config_ng_setup):
         expected = {
             "account": "env-account",  # Level 4 wins
             "user": "snowsql-user",  # Level 1 only
-            "warehouse": "cli-warehouse",  # Level 2 only
+            # warehouse from cli_config is skipped (connections.toml replaces cli_config)
             "database": "toml-db",  # Level 3 only
         }
         assert conn == expected
@@ -619,7 +619,7 @@ def test_all_files_plus_general_env(config_ng_setup):
         expected = {
             "account": "env-account",  # Level 6 wins
             "user": "snowsql-user",  # Level 1 only
-            "role": "cli-role",  # Level 2 only
+            # role from cli_config is skipped (connections.toml replaces cli_config)
             "warehouse": "env-warehouse",  # Level 6 wins
         }
         assert conn == expected
@@ -739,7 +739,7 @@ def test_all_files_plus_two_env_types(config_ng_setup):
         expected = {
             "account": "conn-specific",  # Level 5 wins
             "user": "snowsql-user",  # Level 1 only
-            "password": "cli-password",  # Level 2 only
+            # password from cli_config is skipped (connections.toml replaces cli_config)
             "warehouse": "conn-warehouse",  # Level 5 wins
         }
         assert conn == expected
@@ -904,7 +904,7 @@ def test_multiple_connections_different_source_patterns(config_ng_setup):
         conn1 = get_connection_dict("conn1")
         expected1 = {
             "account": "conn1-env",  # Connection-specific env wins
-            "user": "conn1-user",  # CLI config
+            # user from cli_config is skipped (connections.toml replaces cli_config)
             "warehouse": "conn1-warehouse",  # Connections TOML
             "schema": "common-schema",  # General env
         }

tests/test_config.py

@@ -28,7 +28,6 @@
     get_env_variable_name,
     set_config_value,
 )
-from snowflake.cli.api.config_provider import is_alternative_config_enabled
 from snowflake.cli.api.exceptions import MissingConfigurationError
 
 from tests.testing_utils.files_and_dirs import assert_file_permissions_are_strict
@@ -384,12 +383,7 @@ def assert_correct_connections_loaded():
         assert_correct_connections_loaded()
 
 
-# Legacy version - skip when config_ng is enabled
-@pytest.mark.skipif(
-    is_alternative_config_enabled(),
-    reason="Legacy behavior: connections.toml replaces all connections from config.toml",
-)
-def test_connections_toml_override_config_toml_legacy(
+def test_connections_toml_override_config_toml(
     test_snowcli_config, snowflake_home, config_manager
 ):
     connections_toml = snowflake_home / "connections.toml"
@@ -400,49 +394,14 @@ def test_connections_toml_override_config_toml_legacy(
     )
     config_init(test_snowcli_config)
 
-    # Legacy: Only connections from connections.toml are present
+    # Both legacy and config_ng: Only connections from connections.toml are present
+    # connections.toml REPLACES config.toml connections (not merge)
     assert get_default_connection_dict() == {"database": "overridden_database"}
     assert config_manager["connections"] == {
         "default": {"database": "overridden_database"}
     }
 
 
-# Config_ng version - skip when config_ng is NOT enabled
-@pytest.mark.skipif(
-    not is_alternative_config_enabled(),
-    reason="Config_ng behavior: connections.toml merges with config.toml per-key",
-)
-def test_connections_toml_override_config_toml_config_ng(
-    test_snowcli_config, snowflake_home, config_manager
-):
-    """Test config_ng behavior: connections.toml merges with config.toml per-key"""
-    connections_toml = snowflake_home / "connections.toml"
-    connections_toml.write_text(
-        """[default]
-    database = "overridden_database"
-    """
-    )
-    config_init(test_snowcli_config)
-
-    # Config_ng: Merged - database from connections.toml, other keys from config.toml
-    # The key difference from legacy: keys from config.toml are preserved
-    default_conn = get_default_connection_dict()
-
-    # Key from connections.toml (level 3) overrides
-    assert default_conn["database"] == "overridden_database"
-
-    # Keys from config.toml (level 2) are preserved
-    assert default_conn["schema"] == "test_public"
-    assert default_conn["role"] == "test_role"
-    assert default_conn["warehouse"] == "xs"
-    assert default_conn["password"] == "dummy_password"
-
-    # Verify other connections from config.toml are also accessible
-    full_conn = get_connection_dict("full")
-    assert full_conn["account"] == "dev_account"
-    assert full_conn["user"] == "dev_user"
-
-
 parametrize_chmod = pytest.mark.parametrize(
     "chmod",
     [