fix: use CredentialStoreAdapter in sap_tool instead of raw os.getenv()

tools/src/aden_tools/tools/sap_tool/sap_tool.py

@@ -8,25 +8,13 @@
 
 import base64
 import os
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import httpx
 from fastmcp import FastMCP
 
-
-def _get_config() -> tuple[str, dict] | dict:
-    """Return (base_url, headers) or error dict."""
-    base_url = os.getenv("SAP_BASE_URL", "").rstrip("/")
-    username = os.getenv("SAP_USERNAME", "")
-    password = os.getenv("SAP_PASSWORD", "")
-    if not base_url or not username or not password:
-        return {
-            "error": "SAP_BASE_URL, SAP_USERNAME, and SAP_PASSWORD are required",
-            "help": "Set SAP_BASE_URL, SAP_USERNAME, and SAP_PASSWORD environment variables",
-        }
-    creds = base64.b64encode(f"{username}:{password}".encode()).decode()
-    headers = {"Authorization": f"Basic {creds}", "Accept": "application/json"}
-    return base_url, headers
+if TYPE_CHECKING:
+    from aden_tools.credentials import CredentialStoreAdapter
 
 
 def _get(url: str, headers: dict, params: dict | None = None) -> dict:
@@ -45,9 +33,41 @@ def _odata_list(data: dict) -> tuple[list, int | None]:
     return results, count
 
 
-def register_tools(mcp: FastMCP, credentials: Any = None) -> None:
+def register_tools(
+    mcp: FastMCP,
+    credentials: CredentialStoreAdapter | None = None,
+) -> None:
     """Register SAP S/4HANA tools."""
 
+    def _get_config() -> tuple[str, dict] | dict[str, str]:
+        """Return (base_url, headers) or error dict."""
+        if credentials is not None:
+            base_url = credentials.get("sap_base_url")
+            username = credentials.get("sap_username")
+            password = credentials.get("sap_password")
+        else:
+            base_url = os.getenv("SAP_BASE_URL")
+            username = os.getenv("SAP_USERNAME")
+            password = os.getenv("SAP_PASSWORD")
+
+        if not base_url or not username or not password:
+            return {
+                "error": "SAP credentials not configured",
+                "help": (
+                    "Set SAP_BASE_URL, SAP_USERNAME, and SAP_PASSWORD "
+                    "environment variables or configure via credential store"
+                ),
+            }
+        base_url = base_url.rstrip("/")
+        encoded = base64.b64encode(
+            f"{username}:{password}".encode()
+        ).decode()
+        headers = {
+            "Authorization": f"Basic {encoded}",
+            "Accept": "application/json",
+        }
+        return base_url, headers
+
     @mcp.tool()
     def sap_list_purchase_orders(
         top: int = 50,

tools/tests/tools/test_sap_tool.py

@@ -22,13 +22,30 @@ def _mock_resp(data, status_code=200):
     return resp
 
 
+def _mock_credentials():
+    creds = MagicMock()
+    creds.get.side_effect = lambda key: {
+        "sap_base_url": "https://cred-store.s4hana.ondemand.com",
+        "sap_username": "CRED_USER",
+        "sap_password": "cred-password",
+    }.get(key)
+    return creds
+
+
 @pytest.fixture
 def tool_fns(mcp: FastMCP):
     register_tools(mcp, credentials=None)
     tools = mcp._tool_manager._tools
     return {name: tools[name].fn for name in tools}
 
 
+@pytest.fixture
+def tool_fns_with_creds(mcp: FastMCP):
+    register_tools(mcp, credentials=_mock_credentials())
+    tools = mcp._tool_manager._tools
+    return {name: tools[name].fn for name in tools}
+
+
 class TestSAPListPurchaseOrders:
     def test_missing_credentials(self, tool_fns):
         with patch.dict("os.environ", {}, clear=True):
@@ -176,3 +193,66 @@ def test_successful_list(self, tool_fns):
         assert result["count"] == 1
         assert result["sales_orders"][0]["sales_order"] == "1"
         assert result["sales_orders"][0]["net_amount"] == "25000.00"
+
+
+class TestCredentialStoreAdapter:
+    """Verify credentials are resolved via CredentialStoreAdapter."""
+
+    def test_credential_store_used(self, tool_fns_with_creds):
+        data = {
+            "d": {
+                "__count": "1",
+                "results": [
+                    {
+                        "PurchaseOrder": "4500000001",
+                        "PurchaseOrderType": "NB",
+                        "CompanyCode": "1010",
+                        "Supplier": "17300001",
+                        "CreationDate": "/Date(1672531200000)/",
+                        "PurchaseOrderNetAmount": "15000.00",
+                        "DocumentCurrency": "USD",
+                    }
+                ],
+            }
+        }
+        with patch(
+            "aden_tools.tools.sap_tool.sap_tool.httpx.get",
+            return_value=_mock_resp(data),
+        ) as mock_get:
+            result = tool_fns_with_creds["sap_list_purchase_orders"]()
+
+        assert result["count"] == 1
+        call_url = mock_get.call_args.args[0]
+        assert "cred-store.s4hana.ondemand.com" in call_url
+
+    def test_credential_store_missing_values(self):
+        creds = MagicMock()
+        creds.get.return_value = None
+
+        mcp = FastMCP("test")
+        register_tools(mcp, credentials=creds)
+        tools = mcp._tool_manager._tools
+        fn = tools["sap_list_purchase_orders"].fn
+
+        result = fn()
+        assert "error" in result
+
+    def test_env_fallback_when_no_adapter(self, tool_fns):
+        data = {
+            "d": {
+                "__count": "0",
+                "results": [],
+            }
+        }
+        with (
+            patch.dict("os.environ", ENV),
+            patch(
+                "aden_tools.tools.sap_tool.sap_tool.httpx.get",
+                return_value=_mock_resp(data),
+            ) as mock_get,
+        ):
+            result = tool_fns["sap_list_purchase_orders"]()
+
+        assert result["count"] == 0
+        call_url = mock_get.call_args.args[0]
+        assert "my-tenant-api.s4hana.ondemand.com" in call_url