Merge pull request #170 from negz/creds

Fix get_credentials, and move it to request

Author: negz

crossplane/function/request.py

@@ -14,10 +14,20 @@
 
 """Utilities for working with RunFunctionRequests."""
 
+import dataclasses
+
 import crossplane.function.proto.v1.run_function_pb2 as fnv1
 from crossplane.function import resource
 
 
+@dataclasses.dataclass
+class Credentials:
+    """Credentials."""
+
+    type: str
+    data: dict
+
+
 def get_required_resources(req: fnv1.RunFunctionRequest, name: str) -> list[dict]:
     """Get required resources by name from the request.
 
@@ -73,3 +83,36 @@ def get_required_resource(req: fnv1.RunFunctionRequest, name: str) -> dict | Non
     """
     resources = get_required_resources(req, name)
     return resources[0] if resources else None
+
+
+def get_credentials(req: fnv1.RunFunctionRequest, name: str) -> Credentials:
+    """Get the supplied credentials from the request.
+
+    Args:
+        req: The RunFunctionRequest containing credentials.
+        name: The name of the credentials to get.
+
+    Returns:
+        The requested credentials with type and data.
+
+    If the credentials don't exist, returns empty credentials with type "data"
+    and empty data dictionary.
+    """
+    empty = Credentials(type="data", data={})
+
+    if not req or name not in req.credentials:
+        return empty
+
+    cred = req.credentials[name]
+
+    # Use WhichOneof to determine which field in the oneof is set
+    source_type = cred.WhichOneof("source")
+    if source_type == "credential_data":
+        # Convert bytes data to string data for backward compatibility
+        data = {}
+        for key, value in cred.credential_data.data.items():
+            data[key] = value.decode("utf-8")
+        return Credentials(type="credential_data", data=data)
+
+    # If no recognized source type is set, return empty
+    return empty

crossplane/function/resource.py

@@ -140,24 +140,3 @@ def get_condition(resource: structpb.Struct, typ: str) -> Condition:
         return condition
 
     return unknown
-
-
-@dataclasses.dataclass
-class Credentials:
-    """Credentials."""
-
-    type: str
-    data: dict
-
-
-def get_credentials(req: structpb.Struct, name: str) -> Credentials:
-    """Get the supplied credentials."""
-    empty = Credentials(type="data", data={})
-    if not req or "credentials" not in req:
-        return empty
-    if not req["credentials"] or name not in req["credentials"]:
-        return empty
-    return Credentials(
-        type=req["credentials"][name]["type"],
-        data=struct_to_dict(req["credentials"][name]["data"]),
-    )

tests/test_request.py

@@ -212,6 +212,58 @@ class TestCase:
             got = request.get_required_resource(case.req, case.name)
             self.assertEqual(case.want, got, case.reason)
 
+    def test_get_credentials(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            req: fnv1.RunFunctionRequest
+            name: str
+            want: request.Credentials
+
+        cases = [
+            TestCase(
+                reason="Should return empty credentials when no credentials exist.",
+                req=fnv1.RunFunctionRequest(),
+                name="test",
+                want=request.Credentials(type="data", data={}),
+            ),
+            TestCase(
+                reason="Should return empty credentials when specified name not found.",
+                req=fnv1.RunFunctionRequest(
+                    credentials={
+                        "other-cred": fnv1.Credentials(
+                            credential_data=fnv1.CredentialData(data={"key": b"value"})
+                        )
+                    }
+                ),
+                name="test",
+                want=request.Credentials(type="data", data={}),
+            ),
+            TestCase(
+                reason="Should return credentials when they exist.",
+                req=fnv1.RunFunctionRequest(
+                    credentials={
+                        "test": fnv1.Credentials(
+                            credential_data=fnv1.CredentialData(
+                                data={"username": b"admin", "password": b"secret"}
+                            )
+                        )
+                    }
+                ),
+                name="test",
+                want=request.Credentials(
+                    type="credential_data",
+                    data={"username": "admin", "password": "secret"},
+                ),
+            ),
+        ]
+
+        for case in cases:
+            got = request.get_credentials(case.req, case.name)
+            self.assertEqual(
+                dataclasses.asdict(case.want), dataclasses.asdict(got), case.reason
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_resource.py

@@ -205,49 +205,6 @@ class TestCase:
                 dataclasses.asdict(case.want), dataclasses.asdict(got), "-want, +got"
             )
 
-    def test_get_credentials(self) -> None:
-        @dataclasses.dataclass
-        class TestCase:
-            reason: str
-            req: structpb.Struct
-            name: str
-            want: resource.Credentials
-
-        cases = [
-            TestCase(
-                reason="Return the specified credentials if they exist.",
-                req=resource.dict_to_struct(
-                    {"credentials": {"test": {"type": "data", "data": {"foo": "bar"}}}}
-                ),
-                name="test",
-                want=resource.Credentials(type="data", data={"foo": "bar"}),
-            ),
-            TestCase(
-                reason="Return empty credentials if no credentials section exists.",
-                req=resource.dict_to_struct({}),
-                name="test",
-                want=resource.Credentials(type="data", data={}),
-            ),
-            TestCase(
-                reason="Return empty credentials if the specified name does not exist.",
-                req=resource.dict_to_struct(
-                    {
-                        "credentials": {
-                            "nottest": {"type": "data", "data": {"foo": "bar"}}
-                        }
-                    }
-                ),
-                name="test",
-                want=resource.Credentials(type="data", data={}),
-            ),
-        ]
-
-        for case in cases:
-            got = resource.get_credentials(case.req, case.name)
-            self.assertEqual(
-                dataclasses.asdict(case.want), dataclasses.asdict(got), "-want, +got"
-            )
-
     def test_dict_to_struct(self) -> None:
         @dataclasses.dataclass
         class TestCase: