diff --git a/src/datacustomcode/token_provider.py b/src/datacustomcode/token_provider.py index 981aebf..5327933 100644 --- a/src/datacustomcode/token_provider.py +++ b/src/datacustomcode/token_provider.py @@ -97,10 +97,19 @@ def __init__(self, sf_cli_org: str): def get_token(self) -> "AccessTokenResponse": """Get token from Salesforce SF CLI""" import json + import shutil import subprocess from datacustomcode.deploy import AccessTokenResponse + sf_path = shutil.which("sf") or shutil.which("sf.cmd") + if sf_path is None: + raise RuntimeError( + "The 'sf' command was not found. " + "Install Salesforce CLI: " + "https://developer.salesforce.com/tools/salesforcecli" + ) + def _run_sf_command(args: list[str], description: str) -> dict: try: result = subprocess.run( @@ -110,12 +119,6 @@ def _run_sf_command(args: list[str], description: str) -> dict: 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}'" @@ -142,7 +145,7 @@ def _run_sf_command(args: list[str], description: str) -> dict: # Get org info from sf org display display_data = _run_sf_command( - ["sf", "org", "display", "--target-org", self.sf_cli_org, "--json"], + [sf_path, "org", "display", "--target-org", self.sf_cli_org, "--json"], "sf org display", ) result_data = display_data.get("result", {}) @@ -159,7 +162,7 @@ def _run_sf_command(args: list[str], description: str) -> dict: try: token_data = _run_sf_command( [ - "sf", + sf_path, "org", "auth", "show-access-token", diff --git a/tests/io/reader/test_sf_cli.py b/tests/io/reader/test_sf_cli.py index eca3a56..b7f1b8e 100644 --- a/tests/io/reader/test_sf_cli.py +++ b/tests/io/reader/test_sf_cli.py @@ -73,6 +73,11 @@ class TestGetToken: def reader(self): return _make_reader() + @pytest.fixture(autouse=True) + def mock_sf_which(self): + with patch("shutil.which", return_value="sf"): + yield + def _run_result(self, stdout: str) -> MagicMock: result = MagicMock() result.stdout = stdout @@ -118,7 +123,7 @@ def test_returns_token_and_instance_url(self, reader): ) def test_file_not_found_raises_runtime_error(self, reader): - with patch("subprocess.run", side_effect=FileNotFoundError): + with patch("shutil.which", return_value=None): with pytest.raises(RuntimeError, match="'sf' command was not found"): reader._get_token() diff --git a/tests/test_token_provider.py b/tests/test_token_provider.py index 4fa4a02..b150a73 100644 --- a/tests/test_token_provider.py +++ b/tests/test_token_provider.py @@ -178,24 +178,25 @@ def test_successful_token_retrieval(self): } ) - with patch("subprocess.run") as mock_run: - mock_run.side_effect = [ - MagicMock(stdout=display_output), - MagicMock(stdout=token_output), - ] + with patch("shutil.which", return_value="sf"): + with patch("subprocess.run") as mock_run: + mock_run.side_effect = [ + MagicMock(stdout=display_output), + MagicMock(stdout=token_output), + ] - result = provider.get_token() + result = provider.get_token() - assert isinstance(result, AccessTokenResponse) - assert result.access_token == "cli_access_token" - assert result.instance_url == "https://cli.salesforce.com" + 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] + # 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_fallback_to_display_token_on_older_cli(self): """Test fallback to sf org display token when show-access-token unavailable.""" @@ -224,17 +225,18 @@ def side_effect(*args, **kwargs): mock.stdout = display_output return mock - with patch("subprocess.run", side_effect=side_effect): - result = provider.get_token() + with patch("shutil.which", return_value="sf"): + 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" + 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.""" + """Test that a missing sf binary raises RuntimeError.""" provider = SFCLITokenProvider("test_org") - with patch("subprocess.run", side_effect=FileNotFoundError()): + with patch("shutil.which", return_value=None): with pytest.raises(RuntimeError, match="'sf' command was not found"): provider.get_token()