From 30910d9c66ec94364cbafd465cb37ef32fba42f6 Mon Sep 17 00:00:00 2001 From: Mark DeLaVergne Date: Thu, 4 Jun 2026 20:59:10 -0400 Subject: [PATCH] Support older versions of SF CLI for deploy command --- src/datacustomcode/token_provider.py | 45 ++++++++++++++++------------ tests/io/reader/test_sf_cli.py | 2 +- tests/test_token_provider.py | 34 +++++++++++++++++++++ 3 files changed, 61 insertions(+), 20 deletions(-) diff --git a/src/datacustomcode/token_provider.py b/src/datacustomcode/token_provider.py index bbf230c..981aebf 100644 --- a/src/datacustomcode/token_provider.py +++ b/src/datacustomcode/token_provider.py @@ -140,37 +140,44 @@ def _run_sf_command(args: list[str], description: str) -> dict: ) return dict(data) - # Get instanceUrl from sf org display + # Get org info 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") + result_data = display_data.get("result", {}) + instance_url = result_data.get("instanceUrl") if not instance_url: raise RuntimeError( f"'sf org display' did not return an instance URL " f"for org '{self.sf_cli_org}'" ) - # 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") + # Try show-access-token first (SF CLI >= 2.136.6); fall back to the + # token from sf org display (older CLIs don't redact it). + access_token = None + try: + 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") + except RuntimeError: + # Command not available on older SF CLI versions + access_token = result_data.get("accessToken") + if not access_token: raise RuntimeError( - f"'sf org auth show-access-token' did not return an access token " - f"for org '{self.sf_cli_org}'" + f"Could not obtain an access token for org '{self.sf_cli_org}'. " + f"Upgrade SF CLI to 2.136.6+ or ensure the org is authenticated." ) return AccessTokenResponse(access_token=access_token, instance_url=instance_url) diff --git a/tests/io/reader/test_sf_cli.py b/tests/io/reader/test_sf_cli.py index 620b92b..eca3a56 100644 --- a/tests/io/reader/test_sf_cli.py +++ b/tests/io/reader/test_sf_cli.py @@ -175,7 +175,7 @@ def test_missing_access_token_raises_runtime_error(self, reader): 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"): + with pytest.raises(RuntimeError, match="Could not obtain an access token"): reader._get_token() def test_missing_instance_url_raises_runtime_error(self, reader): diff --git a/tests/test_token_provider.py b/tests/test_token_provider.py index ec43755..4fa4a02 100644 --- a/tests/test_token_provider.py +++ b/tests/test_token_provider.py @@ -197,6 +197,40 @@ def test_successful_token_retrieval(self): 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_fallback_to_display_token_on_older_cli(self): + """Test fallback to sf org display token when show-access-token unavailable.""" + import json + import subprocess + + provider = SFCLITokenProvider("test_org") + + display_output = json.dumps( + { + "status": 0, + "result": { + "accessToken": "display_token", + "instanceUrl": "https://cli.salesforce.com", + }, + } + ) + + def side_effect(*args, **kwargs): + cmd = args[0] + if "show-access-token" in cmd: + raise subprocess.CalledProcessError( + returncode=2, cmd="sf", stderr="not a sf command" + ) + mock = MagicMock() + mock.stdout = display_output + return mock + + with patch("subprocess.run", side_effect=side_effect): + result = provider.get_token() + + assert isinstance(result, AccessTokenResponse) + assert result.access_token == "display_token" + assert result.instance_url == "https://cli.salesforce.com" + def test_sf_command_not_found(self): """Test that FileNotFoundError is wrapped in RuntimeError.""" provider = SFCLITokenProvider("test_org")