chore: Add unit test coverage

Author: anubhav756

packages/toolbox-core/tests/test_tool.py

@@ -22,11 +22,13 @@
 from aiohttp import ClientSession
 from aioresponses import aioresponses
 from pydantic import ValidationError
+from warnings import catch_warnings, simplefilter
 
 from toolbox_core.protocol import ParameterSchema
 from toolbox_core.tool import ToolboxTool, create_func_docstring, resolve_value
 
 TEST_BASE_URL = "http://toolbox.example.com"
+HTTPS_BASE_URL = "https://toolbox.example.com"
 TEST_TOOL_NAME = "sample_tool"
 
 
@@ -195,7 +197,7 @@ async def test_tool_creation_callable_and_run(
     Tests creating a ToolboxTool, checks callability, and simulates a run.
     """
     tool_name = TEST_TOOL_NAME
-    base_url = TEST_BASE_URL
+    base_url = HTTPS_BASE_URL
     invoke_url = f"{base_url}/api/tool/{tool_name}/invoke"
 
     input_args = {"message": "hello world", "count": 5}
@@ -246,7 +248,7 @@ async def test_tool_run_with_pydantic_validation_error(
     due to Pydantic validation *before* making an HTTP request.
     """
     tool_name = TEST_TOOL_NAME
-    base_url = TEST_BASE_URL
+    base_url = HTTPS_BASE_URL
     invoke_url = f"{base_url}/api/tool/{tool_name}/invoke"
 
     with aioresponses() as m:
@@ -340,18 +342,23 @@ async def test_resolve_value_async_callable():
 
 def test_tool_init_basic(http_session, sample_tool_params, sample_tool_description):
     """Tests basic tool initialization without headers or auth."""
-    tool_instance = ToolboxTool(
-        session=http_session,
-        base_url=TEST_BASE_URL,
-        name=TEST_TOOL_NAME,
-        description=sample_tool_description,
-        params=sample_tool_params,
-        required_authn_params={},
-        required_authz_tokens=[],
-        auth_service_token_getters={},
-        bound_params={},
-        client_headers={},
-    )
+    with catch_warnings(record=True) as record:
+        simplefilter("always")
+
+        tool_instance = ToolboxTool(
+            session=http_session,
+            base_url=HTTPS_BASE_URL,
+            name=TEST_TOOL_NAME,
+            description=sample_tool_description,
+            params=sample_tool_params,
+            required_authn_params={},
+            required_authz_tokens=[],
+            auth_service_token_getters={},
+            bound_params={},
+            client_headers={},
+        )
+    assert len(record) == 0, f"ToolboxTool instantiation unexpectedly warned: {[f'{w.category.__name__}: {w.message}' for w in record]}"
+
     assert tool_instance.__name__ == TEST_TOOL_NAME
     assert inspect.iscoroutinefunction(tool_instance.__call__)
     assert "message" in tool_instance.__signature__.parameters
@@ -367,7 +374,7 @@ def test_tool_init_with_client_headers(
     """Tests tool initialization *with* client headers."""
     tool_instance = ToolboxTool(
         session=http_session,
-        base_url=TEST_BASE_URL,
+        base_url=HTTPS_BASE_URL,
         name=TEST_TOOL_NAME,
         description=sample_tool_description,
         params=sample_tool_params,
@@ -395,7 +402,7 @@ def test_tool_init_header_auth_conflict(
     ):
         ToolboxTool(
             session=http_session,
-            base_url=TEST_BASE_URL,
+            base_url=HTTPS_BASE_URL,
             name="auth_conflict_tool",
             description=sample_tool_description,
             params=sample_tool_auth_params,
@@ -418,7 +425,7 @@ def test_tool_add_auth_token_getters_conflict_with_existing_client_header(
     """
     tool_instance = ToolboxTool(
         session=http_session,
-        base_url=TEST_BASE_URL,
+        base_url=HTTPS_BASE_URL,
         name="tool_with_client_header",
         description=sample_tool_description,
         params=sample_tool_params,
@@ -454,7 +461,7 @@ def test_add_auth_token_getters_unused_token(
     """
     tool_instance = ToolboxTool(
         session=http_session,
-        base_url=TEST_BASE_URL,
+        base_url=HTTPS_BASE_URL,
         name=TEST_TOOL_NAME,
         description=sample_tool_description,
         params=sample_tool_params,
@@ -469,3 +476,126 @@ def test_add_auth_token_getters_unused_token(
 
     with pytest.raises(ValueError, match=expected_error_message):
         tool_instance.add_auth_token_getters(unused_auth_getters)
+
+
+# --- Test for the HTTP Warning ---
+@pytest.mark.parametrize(
+    "trigger_condition_params",
+    [
+        {"client_headers": {"X-Some-Header": "value"}},
+        {"required_authn_params": {"param1": ["auth-service1"]}},
+        {"required_authz_tokens": ["auth-service2"]},
+        {
+            "client_headers": {"X-Some-Header": "value"},
+            "required_authn_params": {"param1": ["auth-service1"]},
+        },
+        {
+            "client_headers": {"X-Some-Header": "value"},
+            "required_authz_tokens": ["auth-service2"],
+        },
+        {
+            "required_authn_params": {"param1": ["auth-service1"]},
+            "required_authz_tokens": ["auth-service2"],
+        },
+        {
+            "client_headers": {"X-Some-Header": "value"},
+            "required_authn_params": {"param1": ["auth-service1"]},
+            "required_authz_tokens": ["auth-service2"],
+        },
+    ],
+    ids=[
+        "client_headers_only",
+        "authn_params_only",
+        "authz_tokens_only",
+        "headers_and_authn",
+        "headers_and_authz",
+        "authn_and_authz",
+        "all_three_conditions",
+    ],
+)
+def test_tool_init_http_warning_when_sensitive_info_over_http(
+    http_session: ClientSession,
+    sample_tool_params: list[ParameterSchema],
+    sample_tool_description: str,
+    trigger_condition_params: dict,
+):
+    """
+    Tests that a UserWarning is issued if client headers, auth params, or
+    auth tokens are present and the base_url is HTTP.
+    """
+    expected_warning_message = (
+        "Sending ID token over HTTP. User data may be exposed. "
+        "Use HTTPS for secure communication."
+    )
+
+    init_kwargs = {
+        "session": http_session,
+        "base_url": TEST_BASE_URL,
+        "name": "http_warning_tool",
+        "description": sample_tool_description,
+        "params": sample_tool_params,
+        "required_authn_params": {},
+        "required_authz_tokens": [],
+        "auth_service_token_getters": {},
+        "bound_params": {},
+        "client_headers": {},
+    }
+    # Apply the specific conditions for this parametrized test
+    init_kwargs.update(trigger_condition_params)
+
+    with pytest.warns(UserWarning, match=expected_warning_message):
+        ToolboxTool(**init_kwargs)
+
+
+def test_tool_init_no_http_warning_if_https(
+    http_session: ClientSession,
+    sample_tool_params: list[ParameterSchema],
+    sample_tool_description: str,
+    static_client_header: dict,
+):
+    """
+    Tests that NO UserWarning is issued if client headers are present but
+    the base_url is HTTPS.
+    """
+    with catch_warnings(record=True) as record:
+        simplefilter("always")
+
+        ToolboxTool(
+            session=http_session,
+            base_url=HTTPS_BASE_URL,
+            name="https_tool",
+            description=sample_tool_description,
+            params=sample_tool_params,
+            required_authn_params={},
+            required_authz_tokens=[],
+            auth_service_token_getters={},
+            bound_params={},
+            client_headers=static_client_header,
+        )
+    assert len(record) == 0, f"Expected no warnings, but got: {[f'{w.category.__name__}: {w.message}' for w in record]}"
+
+def test_tool_init_no_http_warning_if_no_sensitive_info_on_http(
+    http_session: ClientSession,
+    sample_tool_params: list[ParameterSchema],
+    sample_tool_description: str,
+):
+    """
+    Tests that NO UserWarning is issued if the URL is HTTP but there are
+    no client headers, auth params, or auth tokens.
+    """
+    with catch_warnings(record=True) as record:
+        simplefilter("always")
+
+        ToolboxTool(
+            session=http_session,
+            base_url=TEST_BASE_URL,
+            name="http_tool_no_sensitive",
+            description=sample_tool_description,
+            params=sample_tool_params,
+            required_authn_params={},
+            required_authz_tokens=[],
+            auth_service_token_getters={},
+            bound_params={},
+            client_headers={},
+        )
+    assert len(record) == 0, f"Expected no warnings, but got: {[f'{w.category.__name__}: {w.message}' for w in record]}"