Add convenience helpers for conditions, status, and resource names

crossplane/function/resource.py

@@ -16,6 +16,7 @@
 
 import dataclasses
 import datetime
+import hashlib
 
 import pydantic
 from google.protobuf import json_format
@@ -59,6 +60,25 @@ def update(r: fnv1.Resource, source: dict | structpb.Struct | pydantic.BaseModel
             raise TypeError(msg)
 
 
+def update_status(
+    r: fnv1.Resource,
+    status: dict | pydantic.BaseModel,
+) -> None:
+    """Update a resource's status.
+
+    Args:
+        r: A composite or composed resource to update.
+        status: The status to set, as a dictionary or Pydantic model.
+
+    Sets ``r.resource.status`` from the supplied status. When the status
+    is a Pydantic model, fields set to their default value are excluded,
+    matching the behavior of :func:`update`.
+    """
+    if isinstance(status, pydantic.BaseModel):
+        status = status.model_dump(exclude_defaults=True, warnings=False)
+    update(r, {"status": status})
+
+
 def dict_to_struct(d: dict) -> structpb.Struct:
     """Create a Struct well-known type from the supplied dict.
 
@@ -140,3 +160,37 @@ def get_condition(resource: structpb.Struct, typ: str) -> Condition:
         return condition
 
     return unknown
+
+
+_DNS_LABEL_MAX = 63
+_HASH_LEN = 5
+
+
+def child_name(*parts: str, sep: str = "-") -> str:
+    """Build a deterministic, DNS-label-safe name for a child resource.
+
+    Args:
+        *parts: Name components to join (e.g. parent name, suffix).
+        sep: Separator between parts. Defaults to "-".
+
+    Returns:
+        A name that is at most 63 characters long.
+
+    Composition functions often derive child resource names from a parent
+    name and a discriminator. The resulting name must be a valid DNS label
+    (at most 63 characters). This function joins the parts, appends a
+    deterministic 5-character hash suffix for uniqueness, and truncates
+    the prefix to fit within the limit.
+
+    The hash suffix is always appended, even for short names, so that
+    names are visually consistent regardless of length::
+
+        child_name("my-xr", "bucket")       # "my-xr-bucket-a1b2c"
+        child_name("my-very-long-xr-name",
+                   "with-a-very-long-suffix") # truncated to 63 chars
+    """
+    full = sep.join(parts)
+    h = hashlib.sha256(full.encode()).hexdigest()[:_HASH_LEN]
+    max_prefix = _DNS_LABEL_MAX - _HASH_LEN - 1
+    prefix = full[:max_prefix].rstrip(sep)
+    return f"{prefix}{sep}{h}"

crossplane/function/response.py

@@ -81,6 +81,44 @@ def fatal(rsp: fnv1.RunFunctionResponse, message: str) -> None:
     )
 
 
+_STATUS_MAP = {
+    "True": fnv1.STATUS_CONDITION_TRUE,
+    "False": fnv1.STATUS_CONDITION_FALSE,
+    "Unknown": fnv1.STATUS_CONDITION_UNKNOWN,
+}
+
+
+def set_conditions(
+    rsp: fnv1.RunFunctionResponse,
+    *conditions: resource.Condition,
+) -> None:
+    """Set one or more conditions on the composite resource (XR).
+
+    Args:
+        rsp: The RunFunctionResponse to update.
+        *conditions: The conditions to set.
+
+    Each condition is appended to ``rsp.conditions``. Crossplane uses the
+    conditions returned by a function to set custom status conditions on
+    the composite resource.
+
+    The ``last_transition_time`` field of each condition is ignored.
+    Crossplane sets the transition time itself.
+
+    Do not set the ``Ready`` condition type. Crossplane manages it based
+    on resource readiness.
+    """
+    for condition in conditions:
+        c = fnv1.Condition(
+            type=condition.typ,
+            status=_STATUS_MAP.get(condition.status, fnv1.STATUS_CONDITION_UNKNOWN),
+            reason=condition.reason or "",
+        )
+        if condition.message:
+            c.message = condition.message
+        rsp.conditions.append(c)
+
+
 def set_output(rsp: fnv1.RunFunctionResponse, output: dict | structpb.Struct) -> None:
     """Set the output field in a RunFunctionResponse for operation functions.
 

tests/test_resource.py

@@ -29,6 +29,56 @@ class TestResource(unittest.TestCase):
     def setUp(self) -> None:
         logging.configure(level=logging.Level.DISABLED)
 
+    def test_update_status(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            r: fnv1.Resource
+            status: dict | pydantic.BaseModel
+            want: dict
+
+        cases = [
+            TestCase(
+                reason="Setting status from a dict should work.",
+                r=fnv1.Resource(
+                    resource=resource.dict_to_struct(
+                        {"apiVersion": "example.org", "kind": "XR"}
+                    ),
+                ),
+                status={"ready": True},
+                want={
+                    "apiVersion": "example.org",
+                    "kind": "XR",
+                    "status": {"ready": True},
+                },
+            ),
+            TestCase(
+                reason="Setting status from a Pydantic model should work.",
+                r=fnv1.Resource(
+                    resource=resource.dict_to_struct(
+                        {"apiVersion": "example.org", "kind": "XR"}
+                    ),
+                ),
+                status=v1beta2.ForProvider(region="us-west-2"),
+                want={
+                    "apiVersion": "example.org",
+                    "kind": "XR",
+                    "status": {"region": "us-west-2"},
+                },
+            ),
+            TestCase(
+                reason="Setting status on an empty resource should work.",
+                r=fnv1.Resource(),
+                status={"replicas": 3},
+                want={"status": {"replicas": 3}},
+            ),
+        ]
+
+        for case in cases:
+            resource.update_status(case.r, case.status)
+            got = resource.struct_to_dict(case.r.resource)
+            self.assertEqual(case.want, got, case.reason)
+
     def test_add(self) -> None:
         @dataclasses.dataclass
         class TestCase:
@@ -324,6 +374,59 @@ class TestCase:
             got = resource.struct_to_dict(case.s)
             self.assertEqual(case.want, got, "-want, +got")
 
+    def test_child_name(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            parts: list[str]
+            want: str
+
+        cases = [
+            TestCase(
+                reason="A short name should be joined with a hash suffix.",
+                parts=["my-xr", "bucket"],
+                want="my-xr-bucket-05ecb",
+            ),
+            TestCase(
+                reason="A single part should get a hash suffix.",
+                parts=["my-xr"],
+                want="my-xr-9d53f",
+            ),
+            TestCase(
+                reason="A long name should be truncated to fit within 63 characters.",
+                parts=["a" * 40, "b" * 40],
+                want="a" * 40 + "-" + "b" * 16 + "-" + "f5e42",
+            ),
+            TestCase(
+                reason="A name that would end with a trailing separator "
+                "after truncation should have the separator stripped.",
+                parts=["a" * 56 + "-", "x"],
+                # Without stripping, this would be "aaa..a--<hash>".
+                # The trailing separator from the truncation is stripped.
+                want="a" * 56 + "-" + "995eb",
+            ),
+            TestCase(
+                reason="The same inputs should always produce the same name.",
+                parts=["parent", "child"],
+                want="parent-child-2f0c9",
+            ),
+        ]
+
+        for case in cases:
+            got = resource.child_name(*case.parts)
+            self.assertEqual(case.want, got, case.reason)
+            self.assertLessEqual(len(got), 63, case.reason)
+
+    def test_child_name_deterministic(self) -> None:
+        a = resource.child_name("parent", "child")
+        b = resource.child_name("parent", "child")
+        self.assertEqual(a, b)
+
+    def test_child_name_unique(self) -> None:
+        a = resource.child_name("parent", "child-a")
+        b = resource.child_name("parent", "child-b")
+        self.assertNotEqual(a, b)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_response.py

@@ -71,6 +71,72 @@ class TestCase:
                 "-want, +got",
             )
 
+    def test_set_conditions(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            conditions: list[resource.Condition]
+            want_types: list[str]
+            want_statuses: list[fnv1.Status.ValueType]
+            want_reasons: list[str]
+            want_messages: list[str]
+
+        cases = [
+            TestCase(
+                reason="A single True condition should work.",
+                conditions=[
+                    resource.Condition(
+                        typ="DatabaseReady",
+                        status="True",
+                        reason="Available",
+                        message="The database is ready",
+                    ),
+                ],
+                want_types=["DatabaseReady"],
+                want_statuses=[fnv1.STATUS_CONDITION_TRUE],
+                want_reasons=["Available"],
+                want_messages=["The database is ready"],
+            ),
+            TestCase(
+                reason="Multiple conditions should all be appended.",
+                conditions=[
+                    resource.Condition(
+                        typ="DatabaseReady",
+                        status="True",
+                        reason="Available",
+                    ),
+                    resource.Condition(
+                        typ="CacheReady",
+                        status="False",
+                        reason="Creating",
+                    ),
+                    resource.Condition(
+                        typ="NetworkReady",
+                        status="Unknown",
+                    ),
+                ],
+                want_types=["DatabaseReady", "CacheReady", "NetworkReady"],
+                want_statuses=[
+                    fnv1.STATUS_CONDITION_TRUE,
+                    fnv1.STATUS_CONDITION_FALSE,
+                    fnv1.STATUS_CONDITION_UNKNOWN,
+                ],
+                want_reasons=["Available", "Creating", ""],
+                want_messages=["", "", ""],
+            ),
+        ]
+
+        for case in cases:
+            rsp = fnv1.RunFunctionResponse()
+            response.set_conditions(rsp, *case.conditions)
+
+            self.assertEqual(len(case.conditions), len(rsp.conditions), case.reason)
+            for i, got in enumerate(rsp.conditions):
+                self.assertEqual(case.want_types[i], got.type, case.reason)
+                self.assertEqual(case.want_statuses[i], got.status, case.reason)
+                self.assertEqual(case.want_reasons[i], got.reason, case.reason)
+                self.assertEqual(case.want_messages[i], got.message, case.reason)
+
     def test_set_output(self) -> None:
         @dataclasses.dataclass
         class TestCase: