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

src/datacustomcode/token_provider.py

@@ -101,48 +101,75 @@ 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 dict(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/io/reader/test_sf_cli.py

@@ -79,23 +79,43 @@ def _run_result(self, stdout: str) -> MagicMock:
         return result
 
     def test_returns_token_and_instance_url(self, reader):
+        display_result = self._run_result(
+            _sf_display_output("redacted", "https://org.salesforce.com")
+        )
+        token_result = self._run_result(
+            json.dumps({"status": 0, "result": {"accessToken": "mytoken"}})
+        )
         with patch(
             "subprocess.run",
-            return_value=self._run_result(
-                _sf_display_output("mytoken", "https://org.salesforce.com")
-            ),
+            side_effect=[display_result, token_result],
         ) as mock_run:
             token, url = reader._get_token()
 
         assert token == "mytoken"
         assert url == "https://org.salesforce.com"
-        mock_run.assert_called_once_with(
+        assert mock_run.call_count == 2
+        mock_run.assert_any_call(
             ["sf", "org", "display", "--target-org", "dev1", "--json"],
             capture_output=True,
             text=True,
             check=True,
             timeout=30,
         )
+        mock_run.assert_any_call(
+            [
+                "sf",
+                "org",
+                "auth",
+                "show-access-token",
+                "--target-org",
+                "dev1",
+                "--json",
+            ],
+            capture_output=True,
+            text=True,
+            check=True,
+            timeout=30,
+        )
 
     def test_file_not_found_raises_runtime_error(self, reader):
         with patch("subprocess.run", side_effect=FileNotFoundError):
@@ -148,21 +168,21 @@ def test_nonzero_status_without_message_uses_unknown_error(self, reader):
                 reader._get_token()
 
     def test_missing_access_token_raises_runtime_error(self, reader):
-        payload = json.dumps(
+        display_result = MagicMock()
+        display_result.stdout = json.dumps(
             {"status": 0, "result": {"instanceUrl": "https://x.salesforce.com"}}
         )
-        result = MagicMock()
-        result.stdout = payload
-        with patch("subprocess.run", return_value=result):
-            with pytest.raises(RuntimeError, match="access token or instance URL"):
+        token_result = MagicMock()
+        token_result.stdout = json.dumps({"status": 0, "result": {}})
+        with patch("subprocess.run", side_effect=[display_result, token_result]):
+            with pytest.raises(RuntimeError, match="did not return an access token"):
                 reader._get_token()
 
     def test_missing_instance_url_raises_runtime_error(self, reader):
-        payload = json.dumps({"status": 0, "result": {"accessToken": "tok"}})
-        result = MagicMock()
-        result.stdout = payload
-        with patch("subprocess.run", return_value=result):
-            with pytest.raises(RuntimeError, match="access token or instance URL"):
+        display_result = MagicMock()
+        display_result.stdout = json.dumps({"status": 0, "result": {}})
+        with patch("subprocess.run", return_value=display_result):
+            with pytest.raises(RuntimeError, match="did not return an instance URL"):
                 reader._get_token()
 
 

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")