SNOW-2306184: config refactor - old & new unit tests pass

Author: sfc-gh-mraba

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

@@ -53,13 +53,16 @@ class SnowSQLConfigFile(ValueSource):
     Later files override earlier files for the same keys.
     Returns configuration for ALL connections.
 
-    Config files searched (in order):
+    Config files searched (in order, when not in test mode):
     1. Bundled default config (if in package)
     2. /etc/snowsql.cnf (system-wide)
     3. /etc/snowflake/snowsql.cnf (alternative system)
     4. /usr/local/etc/snowsql.cnf (local system)
     5. ~/.snowsql.cnf (legacy user config)
     6. ~/.snowsql/config (current user config)
+
+    In test mode (when config_file_override is set), SnowSQL config files are skipped
+    to ensure test isolation.
     """
 
     # SnowSQL uses different key names - map them to CLI standard names
@@ -87,13 +90,31 @@ class SnowSQLConfigFile(ValueSource):
 
     def __init__(self):
         """Initialize SnowSQL config file source."""
-        self._config_files = [
-            Path("/etc/snowsql.cnf"),
-            Path("/etc/snowflake/snowsql.cnf"),
-            Path("/usr/local/etc/snowsql.cnf"),
-            Path.home() / ".snowsql.cnf",
-            Path.home() / ".snowsql" / "config",
-        ]
+        # Use SNOWFLAKE_HOME if set and directory exists, otherwise use standard paths
+        snowflake_home = os.environ.get("SNOWFLAKE_HOME")
+        if snowflake_home:
+            snowflake_home_path = Path(snowflake_home).expanduser()
+            if snowflake_home_path.exists():
+                # Use only the SnowSQL config file within SNOWFLAKE_HOME
+                self._config_files = [snowflake_home_path / "config"]
+            else:
+                # SNOWFLAKE_HOME set but doesn't exist, use standard paths
+                self._config_files = [
+                    Path("/etc/snowsql.cnf"),
+                    Path("/etc/snowflake/snowsql.cnf"),
+                    Path("/usr/local/etc/snowsql.cnf"),
+                    Path.home() / ".snowsql.cnf",
+                    Path.home() / ".snowsql" / "config",
+                ]
+        else:
+            # Standard paths when SNOWFLAKE_HOME not set
+            self._config_files = [
+                Path("/etc/snowsql.cnf"),
+                Path("/etc/snowflake/snowsql.cnf"),
+                Path("/usr/local/etc/snowsql.cnf"),
+                Path.home() / ".snowsql.cnf",
+                Path.home() / ".snowsql" / "config",
+            ]
 
     @property
     def source_name(self) -> str:
@@ -162,17 +183,46 @@ class CliConfigFile(ValueSource):
     Does NOT merge multiple files - first found wins.
     Returns configuration for ALL connections.
 
-    Search order:
+    Search order (when no override is set):
     1. ./config.toml (current directory)
     2. ~/.snowflake/config.toml (user config)
+
+    When config_file_override is set (e.g., in tests), only that file is used.
     """
 
     def __init__(self):
         """Initialize CLI config file source."""
-        self._search_paths = [
-            Path.cwd() / "config.toml",
-            Path.home() / ".snowflake" / "config.toml",
-        ]
+        # Check for config file override from CLI context first
+        try:
+            from snowflake.cli.api.cli_global_context import get_cli_context
+
+            cli_context = get_cli_context()
+            config_override = cli_context.config_file_override
+            if config_override:
+                self._search_paths = [Path(config_override)]
+                return
+        except Exception:
+            pass
+
+        # Use SNOWFLAKE_HOME if set and directory exists, otherwise use standard paths
+        snowflake_home = os.environ.get("SNOWFLAKE_HOME")
+        if snowflake_home:
+            snowflake_home_path = Path(snowflake_home).expanduser()
+            if snowflake_home_path.exists():
+                # Use only config.toml within SNOWFLAKE_HOME
+                self._search_paths = [snowflake_home_path / "config.toml"]
+            else:
+                # SNOWFLAKE_HOME set but doesn't exist, use standard paths
+                self._search_paths = [
+                    Path.cwd() / "config.toml",
+                    Path.home() / ".snowflake" / "config.toml",
+                ]
+        else:
+            # Standard paths when SNOWFLAKE_HOME not set
+            self._search_paths = [
+                Path.cwd() / "config.toml",
+                Path.home() / ".snowflake" / "config.toml",
+            ]
 
     @property
     def source_name(self) -> str:
@@ -234,7 +284,16 @@ class ConnectionsConfigFile(ValueSource):
 
     def __init__(self):
         """Initialize connections.toml source."""
-        self._file_path = Path.home() / ".snowflake" / "connections.toml"
+        # Use SNOWFLAKE_HOME if set and directory exists, otherwise use standard path
+        snowflake_home = os.environ.get("SNOWFLAKE_HOME")
+        if snowflake_home:
+            snowflake_home_path = Path(snowflake_home).expanduser()
+            if snowflake_home_path.exists():
+                self._file_path = snowflake_home_path / "connections.toml"
+            else:
+                self._file_path = Path.home() / ".snowflake" / "connections.toml"
+        else:
+            self._file_path = Path.home() / ".snowflake" / "connections.toml"
 
     @property
     def source_name(self) -> str:
@@ -244,6 +303,15 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
         """
         Read connections.toml if it exists.
         Returns keys in format: connections.{name}.{param} for ALL connections.
+
+        Supports both legacy formats:
+        1. Direct connection sections (legacy):
+           [default]
+           database = "value"
+
+        2. Nested under [connections] section:
+           [connections.default]
+           database = "value"
         """
         if not self._file_path.exists():
             return {}
@@ -253,12 +321,13 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
                 data = tomllib.load(f)
 
             result = {}
-            connections = data.get("connections", {})
 
-            for conn_name, conn_data in connections.items():
-                if isinstance(conn_data, dict):
-                    for param_key, param_value in conn_data.items():
-                        full_key = f"connections.{conn_name}.{param_key}"
+            # Check for direct connection sections (legacy format)
+            for section_name, section_data in data.items():
+                if isinstance(section_data, dict) and section_name != "connections":
+                    # This is a direct connection section like [default]
+                    for param_key, param_value in section_data.items():
+                        full_key = f"connections.{section_name}.{param_key}"
                         if key is None or full_key == key:
                             result[full_key] = ConfigValue(
                                 key=full_key,
@@ -267,6 +336,21 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
                                 raw_value=param_value,
                             )
 
+            # Check for nested [connections] section format
+            connections_section = data.get("connections", {})
+            if isinstance(connections_section, dict):
+                for conn_name, conn_data in connections_section.items():
+                    if isinstance(conn_data, dict):
+                        for param_key, param_value in conn_data.items():
+                            full_key = f"connections.{conn_name}.{param_key}"
+                            if key is None or full_key == key:
+                                result[full_key] = ConfigValue(
+                                    key=full_key,
+                                    value=param_value,
+                                    source_name=self.source_name,
+                                    raw_value=param_value,
+                                )
+
             return result
 
         except Exception as e:

src/snowflake/cli/api/config_provider.py

@@ -158,10 +158,17 @@ def read_config(self) -> None:
         config_manager.read_config()
 
     def get_connection_dict(self, connection_name: str) -> dict:
-        from snowflake.cli.api.config import get_connection_dict
+        from snowflake.cli.api.config import get_config_section
 
-        result = get_connection_dict(connection_name)
-        return self._transform_private_key_raw(result)
+        try:
+            result = get_config_section("connections", connection_name)
+            return self._transform_private_key_raw(result)
+        except KeyError:
+            from snowflake.cli.api.exceptions import MissingConfigurationError
+
+            raise MissingConfigurationError(
+                f"Connection {connection_name} is not configured"
+            )
 
     def get_all_connections(self) -> dict:
         from snowflake.cli.api.config import get_all_connections
@@ -375,15 +382,16 @@ def _get_connection_dict_internal(self, connection_name: str) -> Dict[str, Any]:
         """
         Get connection configuration by name.
 
-        Merges two types of keys:
-        1. Connection-specific: connections.{name}.{param} (from files)
-        2. Flat keys: {param} (from env/CLI, applies to active connection)
+        Behavior is controlled by SNOWFLAKE_CLI_CONNECTIONS_TOML_REPLACE environment variable:
+        - If set to "true" (default): connections.toml completely replaces connections
+          from config.toml (legacy behavior)
+        - If set to "false": connections.toml values are merged with config.toml values
 
         Args:
             connection_name: Name of the connection
 
         Returns:
-            Dictionary of connection parameters
+            Dictionary of connection parameters from file sources only
         """
         self._ensure_initialized()
 
@@ -392,19 +400,58 @@ def _get_connection_dict_internal(self, connection_name: str) -> Dict[str, Any]:
             self._config_cache = self._resolver.resolve()
 
         connection_dict: Dict[str, Any] = {}
-
-        # First, get connection-specific keys (from file sources)
         connection_prefix = f"connections.{connection_name}."
-        for key, value in self._config_cache.items():
-            if key.startswith(connection_prefix):
-                # Extract parameter name
-                param_name = key[len(connection_prefix) :]
-                connection_dict[param_name] = value
 
-        # Then, overlay flat keys (from env/CLI sources) - these have higher priority
-        for key, value in self._config_cache.items():
-            if "." not in key:  # Flat key like "account", "user"
-                connection_dict[key] = value
+        # Check if replacement behavior is enabled (default: true for backward compatibility)
+        import os
+
+        replace_behavior = os.environ.get(
+            "SNOWFLAKE_CLI_CONNECTIONS_TOML_REPLACE", "true"
+        ).lower() in ("true", "1", "yes", "on")
+
+        if replace_behavior:
+            # Legacy replacement behavior: if connections.toml has the connection,
+            # use ONLY values from connections.toml
+            has_connections_toml = False
+            if self._resolver is not None:
+                for key in self._config_cache.keys():
+                    if key.startswith(connection_prefix):
+                        # Check resolution history to see if this came from connections.toml
+                        history = self._resolver.get_resolution_history(key)
+                        if history and history.selected_entry:
+                            if (
+                                history.selected_entry.config_value.source_name
+                                == "connections_toml"
+                            ):
+                                has_connections_toml = True
+                                break
+
+            if has_connections_toml:
+                # Use ONLY connections.toml values (replacement behavior)
+                for key, value in self._config_cache.items():
+                    if key.startswith(connection_prefix):
+                        # Check if this specific value comes from connections.toml
+                        if self._resolver is not None:
+                            history = self._resolver.get_resolution_history(key)
+                            if history and history.selected_entry:
+                                if (
+                                    history.selected_entry.config_value.source_name
+                                    == "connections_toml"
+                                ):
+                                    param_name = key[len(connection_prefix) :]
+                                    connection_dict[param_name] = value
+            else:
+                # No connections.toml, use merged values from other sources
+                for key, value in self._config_cache.items():
+                    if key.startswith(connection_prefix):
+                        param_name = key[len(connection_prefix) :]
+                        connection_dict[param_name] = value
+        else:
+            # New merging behavior: merge all sources normally
+            for key, value in self._config_cache.items():
+                if key.startswith(connection_prefix):
+                    param_name = key[len(connection_prefix) :]
+                    connection_dict[param_name] = value
 
         if not connection_dict:
             from snowflake.cli.api.exceptions import MissingConfigurationError

src/snowflake/cli/api/connections.py

@@ -69,11 +69,18 @@ class ConnectionContext:
     oauth_enable_single_use_refresh_tokens: Optional[bool] = None
     client_store_temporary_credential: Optional[bool] = None
 
+    # Internal flag to track if config has been loaded
+    _config_loaded: bool = field(default=False, repr=False, init=False)
+
     VALIDATED_FIELD_NAMES = ["schema"]
 
     def present_values_as_dict(self) -> dict:
         """Dictionary representation of this ConnectionContext for values that are not None"""
-        return {k: v for (k, v) in asdict(self).items() if v is not None}
+        return {
+            k: v
+            for (k, v) in asdict(self).items()
+            if v is not None and not k.startswith("_")
+        }
 
     def clone(self) -> ConnectionContext:
         return replace(self)
@@ -112,6 +119,7 @@ def update_from_config(self) -> ConnectionContext:
             del connection_config["private_key_path"]
 
         self.merge_with_config(**connection_config)
+        self._config_loaded = True
         return self
 
     def __repr__(self) -> str:
@@ -138,16 +146,12 @@ def validate_schema(self, value: Optional[str]):
     def validate_and_complete(self):
         """
         Ensure we can create a connection from this context.
-        Loads connection parameters from config if not already set.
+        Sets default connection name if needed, but does not load configuration.
+        Configuration is loaded lazily in build_connection().
         """
         if not self.temporary_connection and not self.connection_name:
             self.connection_name = get_default_connection_name()
 
-        # Load connection parameters from config if we have a connection_name
-        # and haven't loaded them yet (e.g., user is still None)
-        if self.connection_name and not self.user:
-            self.update_from_config()
-
     def build_connection(self):
         from snowflake.cli._app.snow_connector import connect_to_snowflake
 
@@ -160,9 +164,36 @@ def build_connection(self):
                 module="snowflake.connector.config_manager",
             )
 
-        # Get connection parameters and pass them directly to connect_to_snowflake
-        # This restores the original behavior before the change that enforced temporary_connection
-        conn_params = self.present_values_as_dict()
+        if self.temporary_connection:
+            # For temporary connections, pass all parameters
+            # connect_to_snowflake will use these directly without loading config
+            conn_params = self.present_values_as_dict()
+        else:
+            # For named connections, pass connection_name and all override parameters
+            # connect_to_snowflake will load the connection config internally and apply overrides
+            all_params = self.present_values_as_dict()
+            control_params = {
+                "connection_name",
+                "enable_diag",
+                "diag_log_path",
+                "diag_allowlist_path",
+                "temporary_connection",
+                "mfa_passcode",
+            }
+
+            # Separate control parameters from connection overrides
+            conn_params = {}
+            overrides = {}
+
+            for k, v in all_params.items():
+                if k in control_params:
+                    conn_params[k] = v
+                else:
+                    # These are connection parameters that should override config values
+                    overrides[k] = v
+
+            # Merge overrides into conn_params
+            conn_params.update(overrides)
 
         return connect_to_snowflake(**conn_params)
 

tests/conftest.py

@@ -106,6 +106,11 @@ def os_agnostic_snapshot(snapshot):
 def reset_global_context_and_setup_config_and_logging_levels(
     request, test_snowcli_config
 ):
+    # Reset config provider singleton to prevent test interference
+    from snowflake.cli.api.config_provider import reset_config_provider
+
+    reset_config_provider()
+
     with fork_cli_context():
         connection_cache = OpenConnectionCache()
         cli_context_manager = get_cli_context_manager()