Fix Windows FileNotFoundError when invoking sf CLI

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

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

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