SNOW-2306184: config refactor - improve resolution raport

sfc-gh-mraba · sfc-gh-mraba · commit 9add48f45631 · 2025-10-15T16:31:09.000+02:00
diff --git a/src/snowflake/cli/_plugins/helpers/commands.py b/src/snowflake/cli/_plugins/helpers/commands.py
@@ -375,10 +375,12 @@ def show_config_sources(
     Set SNOWFLAKE_CLI_CONFIG_V2_ENABLED=true to enable it.
     """
     from snowflake.cli.api.config_ng import (
-        explain_configuration,
         export_resolution_history,
         is_resolution_logging_available,
     )
+    from snowflake.cli.api.config_ng.resolution_logger import (
+        get_configuration_explanation_results,
+    )
 
     if not is_resolution_logging_available():
         return MessageResult(
@@ -401,18 +403,4 @@ def show_config_sources(
             f"and can be attached to support tickets."
         )
 
-    # Show resolution information
-    explain_configuration(key=key, verbose=show_details)
-
-    if key:
-        return MessageResult(
-            f"\n✅ Showing resolution for key: {key}\n"
-            f"Use --show-details to see the complete resolution chain."
-        )
-    else:
-        return MessageResult(
-            "\n✅ Configuration resolution summary displayed above.\n"
-            "Use a specific key (e.g., 'snow helpers show-config-sources account') "
-            "to see detailed resolution for that key.\n"
-            "Use --show-details to see complete resolution chains for all keys."
-        )
+    return get_configuration_explanation_results(key=key, verbose=show_details)
diff --git a/src/snowflake/cli/api/config_ng/resolution_logger.py b/src/snowflake/cli/api/config_ng/resolution_logger.py
@@ -31,6 +31,12 @@
     get_config_provider_singleton,
 )
 from snowflake.cli.api.console import cli_console
+from snowflake.cli.api.output.types import (
+    CollectionResult,
+    CommandResult,
+    MessageResult,
+    MultipleResults,
+)
 
 if TYPE_CHECKING:
     from snowflake.cli.api.config_ng.resolver import ConfigurationResolver
@@ -304,3 +310,35 @@ def explain_configuration(key: Optional[str] = None, verbose: bool = False) -> N
 
             if verbose:
                 resolver.print_all_chains()
+
+
+def get_configuration_explanation_results(
+    key: Optional[str] = None, verbose: bool = False
+) -> CommandResult:
+    """
+    Build CommandResult(s) representing a fixed-column sources table and optional
+    masked history message, suitable for Snow's output formats.
+
+    Returns:
+        - CollectionResult for the table (always)
+        - If verbose is True, MultipleResults with the table and a MessageResult
+          containing the masked resolution history (for the key or all keys)
+    """
+    from snowflake.cli.api.config_provider import get_config_provider_singleton
+
+    provider = get_config_provider_singleton()
+    provider.read_config()
+
+    resolver = get_resolver()
+    if resolver is None:
+        return MessageResult(
+            "Configuration resolution logging is not available. "
+            f"Set {ALTERNATIVE_CONFIG_ENV_VAR}=true to enable it."
+        )
+
+    table_result: CollectionResult = resolver.build_sources_table(key)
+    if not verbose:
+        return table_result
+
+    history_message: MessageResult = resolver.format_history_message(key)
+    return MultipleResults([table_result, history_message])
diff --git a/src/snowflake/cli/api/config_ng/resolver.py b/src/snowflake/cli/api/config_ng/resolver.py
@@ -27,14 +27,15 @@
 from collections import defaultdict
 from datetime import datetime
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Tuple
 
 from snowflake.cli.api.config_ng.core import (
     ConfigValue,
     ResolutionEntry,
     ResolutionHistory,
 )
 from snowflake.cli.api.console import cli_console
+from snowflake.cli.api.output.types import CollectionResult, MessageResult
 
 if TYPE_CHECKING:
     from snowflake.cli.api.config_ng.core import ValueSource
@@ -62,6 +63,40 @@
     "token_file_path",
 }
 
+# Fixed table columns ordered from most important (left) to least (right)
+SourceColumn = Literal[
+    "params",
+    "global_envs",
+    "connections_env",
+    "snowsql_env",
+    "connections.toml",
+    "config.toml",
+    "snowsql",
+]
+
+TABLE_COLUMNS: Tuple[str, ...] = (
+    "key",
+    "value",
+    "params",
+    "global_envs",
+    "connections_env",
+    "snowsql_env",
+    "connections.toml",
+    "config.toml",
+    "snowsql",
+)
+
+# Mapping of internal source names to fixed table columns
+SOURCE_TO_COLUMN: Dict[str, SourceColumn] = {
+    "cli_arguments": "params",
+    "cli_env": "global_envs",
+    "connection_specific_env": "connections_env",
+    "snowsql_env": "snowsql_env",
+    "connections_toml": "connections.toml",
+    "cli_config_toml": "config.toml",
+    "snowsql_config": "snowsql",
+}
+
 
 def _should_mask_value(key: str) -> bool:
     """
@@ -508,6 +543,104 @@ def get_history_summary(self) -> dict:
         """
         return self._history_tracker.get_summary()
 
+    def build_sources_table(self, key: Optional[str] = None) -> CollectionResult:
+        """
+        Build a tabular view of configuration sources per key.
+
+        Columns (left to right): key, value, params, env, connections.toml, cli_config.toml, snowsql.
+        - value: masked final selected value for the key
+        - presence columns: "+" if a given source provided a value for the key, empty otherwise
+        """
+        # Ensure history is populated
+        if key is None and not self._history_tracker.get_all_histories():
+            # Resolve all keys to populate history
+            self.resolve()
+        elif key is not None and self._history_tracker.get_history(key) is None:
+            # Resolve only the specific key
+            self.resolve(key=key)
+
+        histories = (
+            {key: self._history_tracker.get_history(key)}
+            if key is not None
+            else self._history_tracker.get_all_histories()
+        )
+
+        def _row_items():
+            for k, history in histories.items():
+                if history is None:
+                    continue
+                # Initialize row with fixed columns
+                row: Dict[str, Any] = {c: "" for c in TABLE_COLUMNS}
+                row["key"] = k
+
+                # Final value (masked)
+                masked_final = _mask_sensitive_value(k, history.final_value)
+                row["value"] = masked_final
+
+                # Mark presence per source
+                for entry in history.entries:
+                    source_column = SOURCE_TO_COLUMN.get(entry.config_value.source_name)
+                    if source_column is not None:
+                        row[source_column] = "+"
+
+                # Ensure result preserves the column order
+                ordered_row = {column: row[column] for column in TABLE_COLUMNS}
+                yield ordered_row
+
+        return CollectionResult(_row_items())
+
+    def format_history_message(self, key: Optional[str] = None) -> MessageResult:
+        """
+        Build a masked, human-readable history of merging as a single message.
+        If key is None, returns concatenated histories for all keys.
+        """
+        histories = (
+            {key: self.get_resolution_history(key)}
+            if key is not None
+            else self.get_all_histories()
+        )
+
+        if not histories:
+            return MessageResult("No resolution history available")
+
+        lines: List[str] = []
+        for k in sorted(histories.keys()):
+            history = histories[k]
+            if history is None:
+                continue
+            lines.append(f"{k} resolution chain ({len(history.entries)} sources):")
+            for i, entry in enumerate(history.entries, 1):
+                cv = entry.config_value
+                status_text = (
+                    "(SELECTED)"
+                    if entry.was_used
+                    else (
+                        f"(overridden by {entry.overridden_by})"
+                        if entry.overridden_by
+                        else "(not used)"
+                    )
+                )
+
+                masked_value = _mask_sensitive_value(cv.key, cv.value)
+                masked_raw = (
+                    _mask_sensitive_value(cv.key, cv.raw_value)
+                    if cv.raw_value is not None
+                    else None
+                )
+                value_display = f'"{masked_value}"'
+                if masked_raw is not None and cv.raw_value != cv.value:
+                    value_display = f'"{masked_raw}" → {masked_value}'
+
+                lines.append(f"  {i}. {cv.source_name}: {value_display} {status_text}")
+
+            if history.default_used:
+                masked_default = _mask_sensitive_value(k, history.final_value)
+                lines.append(f"  Default value used: {masked_default}")
+
+            lines.append("")
+
+        return MessageResult("\n".join(lines).rstrip())
+
     def format_resolution_chain(self, key: str) -> str:
         """
         Format the resolution chain for a key (debugging helper).
diff --git a/tests/helpers/test_show_config_sources.py b/tests/helpers/test_show_config_sources.py
@@ -71,65 +71,106 @@ def test_command_unavailable_message_when_logging_not_available(
 
     @mock.patch.dict(os.environ, {ALTERNATIVE_CONFIG_ENV_VAR: "1"}, clear=True)
     @mock.patch("snowflake.cli.api.config_ng.is_resolution_logging_available")
-    @mock.patch("snowflake.cli.api.config_ng.explain_configuration")
+    @mock.patch(
+        "snowflake.cli.api.config_ng.resolution_logger.get_configuration_explanation_results"
+    )
     def test_command_shows_summary_without_arguments(
-        self, mock_explain, mock_is_available, runner
+        self, mock_get_results, mock_is_available, runner
     ):
         """Command should show configuration summary when called without arguments."""
+        from snowflake.cli.api.output.types import CollectionResult
+
         mock_is_available.return_value = True
+        mock_get_results.return_value = CollectionResult([])
         result = runner.invoke(["helpers", COMMAND])
         assert result.exit_code == 0
-        mock_explain.assert_called_once_with(key=None, verbose=False)
-        assert "Configuration resolution summary displayed above" in result.output
+        mock_get_results.assert_called_once_with(key=None, verbose=False)
 
     @mock.patch.dict(os.environ, {ALTERNATIVE_CONFIG_ENV_VAR: "1"}, clear=True)
     @mock.patch("snowflake.cli.api.config_ng.is_resolution_logging_available")
-    @mock.patch("snowflake.cli.api.config_ng.explain_configuration")
-    def test_command_shows_specific_key(self, mock_explain, mock_is_available, runner):
+    @mock.patch(
+        "snowflake.cli.api.config_ng.resolution_logger.get_configuration_explanation_results"
+    )
+    def test_command_shows_specific_key(
+        self, mock_get_results, mock_is_available, runner
+    ):
         """Command should show resolution for specific key when provided."""
+        from snowflake.cli.api.output.types import CollectionResult
+
         mock_is_available.return_value = True
+        mock_get_results.return_value = CollectionResult([])
         result = runner.invoke(["helpers", COMMAND, "account"])
         assert result.exit_code == 0
-        mock_explain.assert_called_once_with(key="account", verbose=False)
-        assert "Showing resolution for key: account" in result.output
+        mock_get_results.assert_called_once_with(key="account", verbose=False)
 
     @mock.patch.dict(os.environ, {ALTERNATIVE_CONFIG_ENV_VAR: "1"}, clear=True)
     @mock.patch("snowflake.cli.api.config_ng.is_resolution_logging_available")
-    @mock.patch("snowflake.cli.api.config_ng.explain_configuration")
+    @mock.patch(
+        "snowflake.cli.api.config_ng.resolution_logger.get_configuration_explanation_results"
+    )
     def test_command_shows_details_with_flag(
-        self, mock_explain, mock_is_available, runner
+        self, mock_get_results, mock_is_available, runner
     ):
         """Command should show detailed resolution when --show-details flag is used."""
+        from snowflake.cli.api.output.types import (
+            CollectionResult,
+            MessageResult,
+            MultipleResults,
+        )
+
         mock_is_available.return_value = True
+        mock_get_results.return_value = MultipleResults(
+            [CollectionResult([]), MessageResult("test history")]
+        )
         result = runner.invoke(["helpers", COMMAND, "--show-details"])
         assert result.exit_code == 0
-        mock_explain.assert_called_once_with(key=None, verbose=True)
-        assert "Configuration resolution summary displayed above" in result.output
+        mock_get_results.assert_called_once_with(key=None, verbose=True)
 
     @mock.patch.dict(os.environ, {ALTERNATIVE_CONFIG_ENV_VAR: "1"}, clear=True)
     @mock.patch("snowflake.cli.api.config_ng.is_resolution_logging_available")
-    @mock.patch("snowflake.cli.api.config_ng.explain_configuration")
+    @mock.patch(
+        "snowflake.cli.api.config_ng.resolution_logger.get_configuration_explanation_results"
+    )
     def test_command_shows_details_with_short_flag(
-        self, mock_explain, mock_is_available, runner
+        self, mock_get_results, mock_is_available, runner
     ):
         """Command should show detailed resolution when -d flag is used."""
+        from snowflake.cli.api.output.types import (
+            CollectionResult,
+            MessageResult,
+            MultipleResults,
+        )
+
         mock_is_available.return_value = True
+        mock_get_results.return_value = MultipleResults(
+            [CollectionResult([]), MessageResult("test history")]
+        )
         result = runner.invoke(["helpers", COMMAND, "-d"])
         assert result.exit_code == 0
-        mock_explain.assert_called_once_with(key=None, verbose=True)
+        mock_get_results.assert_called_once_with(key=None, verbose=True)
 
     @mock.patch.dict(os.environ, {ALTERNATIVE_CONFIG_ENV_VAR: "1"}, clear=True)
     @mock.patch("snowflake.cli.api.config_ng.is_resolution_logging_available")
-    @mock.patch("snowflake.cli.api.config_ng.explain_configuration")
+    @mock.patch(
+        "snowflake.cli.api.config_ng.resolution_logger.get_configuration_explanation_results"
+    )
     def test_command_shows_key_with_details(
-        self, mock_explain, mock_is_available, runner
+        self, mock_get_results, mock_is_available, runner
     ):
         """Command should show detailed resolution for specific key."""
+        from snowflake.cli.api.output.types import (
+            CollectionResult,
+            MessageResult,
+            MultipleResults,
+        )
+
         mock_is_available.return_value = True
+        mock_get_results.return_value = MultipleResults(
+            [CollectionResult([]), MessageResult("test history")]
+        )
         result = runner.invoke(["helpers", COMMAND, "user", "--show-details"])
         assert result.exit_code == 0
-        mock_explain.assert_called_once_with(key="user", verbose=True)
-        assert "Showing resolution for key: user" in result.output
+        mock_get_results.assert_called_once_with(key="user", verbose=True)
 
     @mock.patch.dict(os.environ, {ALTERNATIVE_CONFIG_ENV_VAR: "1"}, clear=True)
     @mock.patch("snowflake.cli.api.config_ng.is_resolution_logging_available")
