test: add tests for external databricks helpers

Author: dbkegley

examples/connect/fastapi/app.py

@@ -9,7 +9,6 @@
 from fastapi.responses import JSONResponse
 from posit.connect.external.databricks import PositCredentialsStrategy
 
-
 DATABRICKS_HOST = os.getenv("DATABRICKS_HOST")
 DATABRICKS_HOST_URL = f"https://{DATABRICKS_HOST}"
 SQL_HTTP_PATH = os.getenv("DATABRICKS_PATH")

src/posit/connect/external/databricks.py

@@ -19,6 +19,7 @@ class CredentialsStrategy(abc.ABC):
     https://github.com/databricks/databricks-sql-python/blob/v3.3.0/src/databricks/sql/auth/authenticators.py#L19-L33
     https://github.com/databricks/databricks-sdk-py/blob/v0.29.0/databricks/sdk/credentials_provider.py#L44-L54
     """
+
     @abc.abstractmethod
     def auth_type(self) -> str:
         raise NotImplementedError

tests/posit/connect/external/test_databricks.py

@@ -0,0 +1,72 @@
+from typing import Dict
+from unittest.mock import patch
+
+import responses
+from posit.connect import Client
+from posit.connect.external.databricks import (
+    CredentialsProvider,
+    PositCredentialsProvider,
+    PositCredentialsStrategy,
+)
+
+
+class mock_strategy:
+    def auth_type(self) -> str:
+        return "local"
+    def __call__(self) -> CredentialsProvider:
+        def inner() -> Dict[str,str]:
+            return {"Authorization": "Bearer static-pat-token"}
+        return inner
+
+
+def register_mocks():
+    responses.post(
+        "https://connect.example/__api__/v1/oauth/integrations/credentials",
+        match=[
+            responses.matchers.urlencoded_params_matcher(
+                {
+                    "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+                    "subject_token_type": "urn:posit:connect:user-session-token",
+                    "subject_token": "cit",
+                }
+            )
+        ],
+        json={
+            "access_token": "dynamic-viewer-access-token",
+            "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "token_type": "Bearer",
+        },
+    )
+
+
+class TestPositCredentialsHelpers:
+    @responses.activate
+    def test_posit_credentials_provider(self):
+        register_mocks()
+
+        client = Client(api_key="12345", url="https://connect.example/")
+        cp = PositCredentialsProvider(posit_oauth=client.oauth, user_session_token="cit")
+        assert cp() == {"Authorization": f"Bearer dynamic-viewer-access-token"}
+
+    @responses.activate
+    @patch.dict("os.environ", {"RSTUDIO_PRODUCT": "CONNECT"})
+    def test_posit_credentials_strategy(self):
+        register_mocks()
+
+        client = Client(api_key="12345", url="https://connect.example/")
+        cs = PositCredentialsStrategy(local_strategy=mock_strategy(),
+                                      user_session_token="cit",
+                                      client=client)
+        cp = cs()
+        assert cs.auth_type() == "posit-oauth-integration"
+        assert cp() == {"Authorization": "Bearer dynamic-viewer-access-token"}
+
+    def test_posit_credentials_strategy_fallback(self):
+        # local_strategy is used when the content is running locally
+        client = Client(api_key="12345", url="https://connect.example/")
+        cs = PositCredentialsStrategy(local_strategy=mock_strategy(),
+                                      user_session_token="cit",
+                                      client=client)
+        cp = cs()
+        assert cs.auth_type() == "local"
+        assert cp() == {"Authorization": "Bearer static-pat-token"}