Use new SF CLI command to obtain token for --sf-cli-org usage

Author: markdlv-sf

src/datacustomcode/token_provider.py

@@ -101,48 +101,70 @@ def get_token(self) -> "AccessTokenResponse":
 
         from datacustomcode.deploy import AccessTokenResponse
 
-        try:
-            result = subprocess.run(
-                ["sf", "org", "display", "--target-org", self.sf_cli_org, "--json"],
-                capture_output=True,
-                text=True,
-                check=True,
-                timeout=30,
-            )
-        except FileNotFoundError as exc:
-            raise RuntimeError(
-                "The 'sf' command was not found. "
-                "Install Salesforce CLI: https://developer.salesforce.com/tools/salesforcecli"
-            ) from exc
-        except subprocess.TimeoutExpired as exc:
-            raise RuntimeError(
-                f"'sf org display' timed out for org '{self.sf_cli_org}'"
-            ) from exc
-        except subprocess.CalledProcessError as exc:
-            raise RuntimeError(
-                f"'sf org display' failed for org '{self.sf_cli_org}': {exc.stderr}"
-            ) from exc
+        def _run_sf_command(args: list[str], description: str) -> dict:
+            try:
+                result = subprocess.run(
+                    args,
+                    capture_output=True,
+                    text=True,
+                    check=True,
+                    timeout=30,
+                )
+            except FileNotFoundError as exc:
+                raise RuntimeError(
+                    "The 'sf' command was not found. "
+                    "Install Salesforce CLI: "
+                    "https://developer.salesforce.com/tools/salesforcecli"
+                ) from exc
+            except subprocess.TimeoutExpired as exc:
+                raise RuntimeError(
+                    f"'{description}' timed out for org '{self.sf_cli_org}'"
+                ) from exc
+            except subprocess.CalledProcessError as exc:
+                raise RuntimeError(
+                    f"'{description}' failed for org '{self.sf_cli_org}': "
+                    f"{exc.stderr}"
+                ) from exc
 
-        try:
-            data = json.loads(result.stdout)
-        except json.JSONDecodeError as exc:
-            raise RuntimeError(
-                f"Failed to parse JSON from 'sf org display': {result.stdout}"
-            ) from exc
+            try:
+                data = json.loads(result.stdout)
+            except json.JSONDecodeError as exc:
+                raise RuntimeError(
+                    f"Failed to parse JSON from '{description}': {result.stdout}"
+                ) from exc
 
-        if data.get("status") != 0:
+            if data.get("status") != 0:
+                raise RuntimeError(
+                    f"SF CLI error for org '{self.sf_cli_org}': "
+                    f"{data.get('message', 'unknown error')}"
+                )
+            return data
+
+        # Get instanceUrl from sf org display
+        display_data = _run_sf_command(
+            ["sf", "org", "display", "--target-org", self.sf_cli_org, "--json"],
+            "sf org display",
+        )
+        instance_url = display_data.get("result", {}).get("instanceUrl")
+        if not instance_url:
             raise RuntimeError(
-                f"SF CLI error for org '{self.sf_cli_org}': "
-                f"{data.get('message', 'unknown error')}"
+                f"'sf org display' did not return an instance URL "
+                f"for org '{self.sf_cli_org}'"
             )
 
-        result_data = data.get("result", {})
-        access_token = result_data.get("accessToken")
-        instance_url = result_data.get("instanceUrl")
-
-        if not access_token or not instance_url:
+        # Get access token via show-access-token (newer SF CLI versions
+        # redact the token in sf org display)
+        token_data = _run_sf_command(
+            [
+                "sf", "org", "auth", "show-access-token",
+                "--target-org", self.sf_cli_org, "--json",
+            ],
+            "sf org auth show-access-token",
+        )
+        access_token = token_data.get("result", {}).get("accessToken")
+        if not access_token:
             raise RuntimeError(
-                f"'sf org display' did not return an access token or instance URL "
+                f"'sf org auth show-access-token' did not return an access token "
                 f"for org '{self.sf_cli_org}'"
             )
 

tests/test_token_provider.py

@@ -160,25 +160,43 @@ def test_successful_token_retrieval(self):
 
         provider = SFCLITokenProvider("test_org")
 
-        cli_output = json.dumps(
+        display_output = json.dumps(
             {
                 "status": 0,
                 "result": {
-                    "accessToken": "cli_access_token",
+                    "accessToken": "[REDACTED]",
                     "instanceUrl": "https://cli.salesforce.com",
                 },
             }
         )
+        token_output = json.dumps(
+            {
+                "status": 0,
+                "result": {
+                    "accessToken": "cli_access_token",
+                },
+            }
+        )
 
         with patch("subprocess.run") as mock_run:
-            mock_run.return_value = MagicMock(stdout=cli_output)
+            mock_run.side_effect = [
+                MagicMock(stdout=display_output),
+                MagicMock(stdout=token_output),
+            ]
 
             result = provider.get_token()
 
             assert isinstance(result, AccessTokenResponse)
             assert result.access_token == "cli_access_token"
             assert result.instance_url == "https://cli.salesforce.com"
 
+            # Verify both commands were called
+            assert mock_run.call_count == 2
+            display_call = mock_run.call_args_list[0]
+            token_call = mock_run.call_args_list[1]
+            assert "org" in display_call[0][0] and "display" in display_call[0][0]
+            assert "show-access-token" in token_call[0][0]
+
     def test_sf_command_not_found(self):
         """Test that FileNotFoundError is wrapped in RuntimeError."""
         provider = SFCLITokenProvider("test_org")