SNOW-2306184: config refactor - improve connections env parsing

Author: sfc-gh-mraba

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

@@ -549,23 +549,35 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
         for env_name, env_value in os.environ.items():
             # Check for connection-specific pattern: SNOWFLAKE_CONNECTIONS_<NAME>_<KEY>
             if env_name.startswith("SNOWFLAKE_CONNECTIONS_"):
-                # Extract connection name and config key
+                # Extract remainder after the prefix
                 remainder = env_name[len("SNOWFLAKE_CONNECTIONS_") :]
-                parts = remainder.split("_", 1)
-                if len(parts) == 2:
-                    conn_name_upper, config_key_upper = parts
-                    conn_name = conn_name_upper.lower()
-                    config_key = config_key_upper.lower()
-
-                    if config_key in _ENV_CONFIG_KEYS:
-                        full_key = f"connections.{conn_name}.{config_key}"
-                        if key is None or full_key == key:
-                            values[full_key] = ConfigValue(
-                                key=full_key,
-                                value=env_value,
-                                source_name=self.source_name,
-                                raw_value=f"{env_name}={env_value}",
-                            )
+
+                # Find the longest matching key suffix from known config keys to
+                # correctly handle underscores both in connection names and keys
+                match: tuple[str, str] | None = None
+                for candidate in sorted(_ENV_CONFIG_KEYS, key=len, reverse=True):
+                    key_suffix = "_" + candidate.upper()
+                    if remainder.endswith(key_suffix):
+                        conn_name_upper = remainder[: -len(key_suffix)]
+                        if conn_name_upper:  # ensure non-empty connection name
+                            match = (conn_name_upper, candidate)
+                            break
+
+                if not match:
+                    # Unknown/unsupported key suffix; ignore
+                    continue
+
+                conn_name_upper, config_key = match
+                conn_name = conn_name_upper.lower()
+
+                full_key = f"connections.{conn_name}.{config_key}"
+                if key is None or full_key == key:
+                    values[full_key] = ConfigValue(
+                        key=full_key,
+                        value=env_value,
+                        source_name=self.source_name,
+                        raw_value=f"{env_name}={env_value}",
+                    )
 
         return values
 

src/snowflake/cli/api/config_provider.py

@@ -603,7 +603,7 @@ def _get_file_based_connections(self) -> dict:
         connections_prefix = "connections."
 
         assert self._resolver is not None
-        for source in self._resolver._sources:  # noqa: SLF001
+        for source in self._resolver.get_sources():
             if source.source_name not in file_source_names:
                 continue
 

tests/config_ng/test_env_parsing.py

@@ -0,0 +1,28 @@
+# Copyright (c) 2024 Snowflake Inc.
+
+"""Focused tests for environment variable parsing in config_ng."""
+
+
+def test_connection_specific_env_with_underscores(config_ng_setup):
+    """Connection names containing underscores should parse correctly.
+
+    Also validate keys that themselves contain underscores (e.g., PRIVATE_KEY_PATH).
+    """
+
+    env_vars = {
+        # Connection-specific variables for connection name with underscores
+        "SNOWFLAKE_CONNECTIONS_DEV_US_EAST_ACCOUNT": "from-specific",
+        "SNOWFLAKE_CONNECTIONS_DEV_US_EAST_PRIVATE_KEY_PATH": "/tmp/example_key.pem",
+        # General env remains available for other flat keys
+        "SNOWFLAKE_SCHEMA": "general-schema",
+    }
+
+    with config_ng_setup(env_vars=env_vars):
+        from snowflake.cli.api.config import get_connection_dict
+
+        conn = get_connection_dict("dev_us_east")
+
+        assert conn["account"] == "from-specific"
+        assert conn["private_key_path"] == "/tmp/example_key.pem"
+        # Ensure general env still contributes flat keys
+        assert conn["schema"] == "general-schema"