reject absolute within path arg

Author: Dylan Huang

eval_protocol/fireworks_api_client.py

@@ -10,116 +10,144 @@
 
 class FireworksAPIClient:
     """Client for making authenticated requests to Fireworks API with proper headers.
-    
+
     This client automatically includes:
     - Authorization header (Bearer token)
     - User-Agent header for tracking eval-protocol CLI usage
     """
-    
+
     def __init__(self, api_key: Optional[str] = None, api_base: Optional[str] = None):
         """Initialize the Fireworks API client.
-        
+
         Args:
             api_key: Fireworks API key. If None, will be read from environment.
             api_base: API base URL. If None, defaults to https://api.fireworks.ai
         """
         self.api_key = api_key
         self.api_base = api_base or os.environ.get("FIREWORKS_API_BASE", "https://api.fireworks.ai")
         self._session = requests.Session()
-        
-    def _get_headers(self, content_type: Optional[str] = "application/json", 
-                     additional_headers: Optional[Dict[str, str]] = None) -> Dict[str, str]:
+
+    def _validate_path_is_relative(self, path: str) -> None:
+        """Validate that the path is relative, not an absolute URL.
+
+        Args:
+            path: The path to validate
+
+        Raises:
+            ValueError: If path appears to be an absolute URL (starts with http:// or https://)
+        """
+        if path.startswith(("http://", "https://")):
+            raise ValueError(
+                f"Absolute URL detected: '{path}'. FireworksAPIClient methods expect relative paths only. "
+                f"Use a relative path like 'v1/path' instead of '{path}'. "
+                f"The client will automatically prepend the api_base: '{self.api_base}'"
+            )
+
+    def _get_headers(
+        self, content_type: Optional[str] = "application/json", additional_headers: Optional[Dict[str, str]] = None
+    ) -> Dict[str, str]:
         """Build headers for API requests.
-        
+
         Args:
             content_type: Content-Type header value. If None, Content-Type won't be set.
             additional_headers: Additional headers to merge in.
-            
+
         Returns:
             Dictionary of headers including authorization and user-agent.
         """
         headers = {
             "User-Agent": get_user_agent(),
         }
-        
+
         if self.api_key:
             headers["Authorization"] = f"Bearer {self.api_key}"
-            
+
         if content_type:
             headers["Content-Type"] = content_type
-            
+
         if additional_headers:
             headers.update(additional_headers)
-            
+
         return headers
-    
-    def get(self, path: str, params: Optional[Dict[str, Any]] = None, 
-            timeout: int = 30, **kwargs) -> requests.Response:
+
+    def get(
+        self, path: str, params: Optional[Dict[str, Any]] = None, timeout: int = 30, **kwargs
+    ) -> requests.Response:
         """Make a GET request to the Fireworks API.
-        
+
         Args:
             path: API path (relative to api_base)
             params: Query parameters
             timeout: Request timeout in seconds
             **kwargs: Additional arguments passed to requests.get
-            
+
         Returns:
             Response object
         """
+        self._validate_path_is_relative(path)
         url = f"{self.api_base.rstrip('/')}/{path.lstrip('/')}"
         headers = self._get_headers(content_type=None)
         if "headers" in kwargs:
             headers.update(kwargs.pop("headers"))
         return self._session.get(url, params=params, headers=headers, timeout=timeout, **kwargs)
-    
-    def post(self, path: str, json: Optional[Dict[str, Any]] = None,
-             data: Optional[Any] = None, files: Optional[Dict[str, Any]] = None,
-             timeout: int = 60, **kwargs) -> requests.Response:
+
+    def post(
+        self,
+        path: str,
+        json: Optional[Dict[str, Any]] = None,
+        data: Optional[Any] = None,
+        files: Optional[Dict[str, Any]] = None,
+        timeout: int = 60,
+        **kwargs,
+    ) -> requests.Response:
         """Make a POST request to the Fireworks API.
-        
+
         Args:
             path: API path (relative to api_base)
             json: JSON payload
             data: Form data payload
             files: Files to upload
             timeout: Request timeout in seconds
             **kwargs: Additional arguments passed to requests.post
-            
+
         Returns:
             Response object
         """
+        self._validate_path_is_relative(path)
         url = f"{self.api_base.rstrip('/')}/{path.lstrip('/')}"
-        
+
         # For file uploads, don't set Content-Type (let requests handle multipart/form-data)
         content_type = None if files else "application/json"
         headers = self._get_headers(content_type=content_type)
-        
+
         if "headers" in kwargs:
             headers.update(kwargs.pop("headers"))
-            
-        return self._session.post(url, json=json, data=data, files=files, 
-                                 headers=headers, timeout=timeout, **kwargs)
-    
-    def put(self, path: str, json: Optional[Dict[str, Any]] = None,
-            timeout: int = 60, **kwargs) -> requests.Response:
+
+        return self._session.post(url, json=json, data=data, files=files, headers=headers, timeout=timeout, **kwargs)
+
+    def put(self, path: str, json: Optional[Dict[str, Any]] = None, timeout: int = 60, **kwargs) -> requests.Response:
         """Make a PUT request to the Fireworks API."""
+        self._validate_path_is_relative(path)
         url = f"{self.api_base.rstrip('/')}/{path.lstrip('/')}"
         headers = self._get_headers()
         if "headers" in kwargs:
             headers.update(kwargs.pop("headers"))
         return self._session.put(url, json=json, headers=headers, timeout=timeout, **kwargs)
-    
-    def patch(self, path: str, json: Optional[Dict[str, Any]] = None,
-              timeout: int = 60, **kwargs) -> requests.Response:
+
+    def patch(
+        self, path: str, json: Optional[Dict[str, Any]] = None, timeout: int = 60, **kwargs
+    ) -> requests.Response:
         """Make a PATCH request to the Fireworks API."""
+        self._validate_path_is_relative(path)
         url = f"{self.api_base.rstrip('/')}/{path.lstrip('/')}"
         headers = self._get_headers()
         if "headers" in kwargs:
             headers.update(kwargs.pop("headers"))
         return self._session.patch(url, json=json, headers=headers, timeout=timeout, **kwargs)
-    
+
     def delete(self, path: str, timeout: int = 30, **kwargs) -> requests.Response:
         """Make a DELETE request to the Fireworks API."""
+        self._validate_path_is_relative(path)
         url = f"{self.api_base.rstrip('/')}/{path.lstrip('/')}"
         headers = self._get_headers(content_type=None)
         if "headers" in kwargs:

tests/test_fireworks_api_client.py

@@ -415,23 +415,19 @@ def test_paths_containing_v1_pattern(self):
 
                 mock_post.reset_mock()
 
-    def test_full_url_passed_by_mistake_detected(self):
-        """Test that accidentally passing a full URL instead of relative path is detected.
+    def test_full_url_passed_by_mistake_raises_error(self):
+        """Test that accidentally passing a full URL instead of relative path raises ValueError.
 
-        This test documents the bug pattern: if a full URL like '{api_base}/v1/path'
-        is passed instead of a relative path like 'v1/path', it will result in a
-        malformed URL like '{api_base}/{api_base}/v1/path'.
-
-        This test verifies that our code correctly handles relative paths (which prevents
-        the bug), and documents what would happen if the bug occurred.
+        This test verifies that our code correctly catches the bug early by raising an error
+        when an absolute URL is passed instead of a relative path.
         """
         api_base = "https://api.fireworks.ai"
         client = FireworksAPIClient(api_key="test_key", api_base=api_base)
 
+        # CORRECT: Relative path (what we should use) - should work fine
         mock_response = MagicMock()
         mock_response.status_code = 200
 
-        # CORRECT: Relative path (what we should use)
         with patch.object(client._session, "post", return_value=mock_response) as mock_post:
             correct_relative_path = "v1/test-evaluator:getUploadEndpoint"
             client.post(correct_relative_path, json={})
@@ -441,27 +437,45 @@ def test_full_url_passed_by_mistake_detected(self):
             expected_correct_url = f"{api_base}/{correct_relative_path}"
             assert correct_url == expected_correct_url
 
-        # INCORRECT: Full URL (this would cause the bug - but we're not actually testing this,
-        # just documenting that our current implementation would create a malformed URL)
-        # If someone accidentally did: client.post(f"{api_base}/v1/path", ...)
-        # The result would be: f"{api_base}/{api_base}/v1/path" which is wrong.
-        # Our tests above ensure we use relative paths, preventing this bug.
-        mock_post.reset_mock()
-        with patch.object(client._session, "post", return_value=mock_response) as mock_post:
-            # Simulating what WOULD happen if buggy code passed full URL
-            buggy_full_url = f"{api_base}/v1/test-evaluator:getUploadEndpoint"
-            client.post(buggy_full_url, json={})
+        # INCORRECT: Full URL should raise ValueError
+        full_url_with_http = "https://api.fireworks.ai/v1/test-evaluator:getUploadEndpoint"
+        with pytest.raises(ValueError, match="Absolute URL detected"):
+            client.post(full_url_with_http, json={})
+
+        full_url_with_http_scheme = "http://api.fireworks.ai/v1/test-evaluator:getUploadEndpoint"
+        with pytest.raises(ValueError, match="Absolute URL detected"):
+            client.post(full_url_with_http_scheme, json={})
+
+        # Test that error message is helpful
+        with pytest.raises(ValueError) as exc_info:
+            client.post(full_url_with_http, json={})
+        error_msg = str(exc_info.value)
+        assert "Absolute URL detected" in error_msg
+        assert full_url_with_http in error_msg
+        assert "relative paths only" in error_msg
+        assert api_base in error_msg  # Should mention api_base in the help message
+
+    def test_all_methods_reject_absolute_urls(self):
+        """Test that all HTTP methods reject absolute URLs."""
+        api_base = "https://api.fireworks.ai"
+        client = FireworksAPIClient(api_key="test_key", api_base=api_base)
 
-            call_args = mock_post.call_args
-            buggy_url = call_args[0][0]
-            # This shows what the buggy URL would look like
-            buggy_expected = f"{api_base}/{buggy_full_url}"
-
-            # This assertion documents the bug pattern - the URL would be malformed
-            assert buggy_url == buggy_expected
-            assert buggy_url.startswith(f"{api_base}/{api_base}"), (
-                "This documents the bug: passing full URL creates double-prefix. Always use relative paths!"
-            )
+        absolute_url = f"{api_base}/v1/test/path"
+
+        methods = [
+            ("get", lambda url: client.get(url)),
+            ("post", lambda url: client.post(url, json={})),
+            ("put", lambda url: client.put(url, json={})),
+            ("patch", lambda url: client.patch(url, json={})),
+            ("delete", lambda url: client.delete(url)),
+        ]
+
+        for method_name, method_call in methods:
+            with pytest.raises(ValueError, match="Absolute URL detected") as exc_info:
+                method_call(absolute_url)
+            error_msg = str(exc_info.value)
+            assert "Absolute URL detected" in error_msg, f"{method_name.upper()} should reject absolute URL"
+            assert absolute_url in error_msg
 
 
 if __name__ == "__main__":