Let a function require the capabilities it can't work without

has_capability answers one capability at a time, so a function that
can't work correctly without some - because it relies on Crossplane
honoring a response field an older Crossplane would ignore - had to
write its own check and its own fatal result, one capability at a time.

missing_capabilities returns which of several capabilities Crossplane
doesn't advertise, and require_capabilities adds one fatal result naming
all of them, so an author fixes them in one go rather than a run per
capability. A Crossplane that advertises no capabilities predates
capability advertisement, so every capability is missing and the result
says so - the fix there is to upgrade, not to enable a feature.

    if not request.require_capabilities(req, rsp, fnv1.CAPABILITY_DEPENDENCIES):
        return rsp

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Steven Borrelli <steve@borrelli.org>

Author: stevendborrelli

crossplane/function/request.py

@@ -17,7 +17,7 @@
 import dataclasses
 
 import crossplane.function.proto.v1.run_function_pb2 as fnv1
-from crossplane.function import resource
+from crossplane.function import resource, response
 
 
 @dataclasses.dataclass
@@ -183,6 +183,79 @@ def has_capability(
     return cap in req.meta.capabilities
 
 
+def missing_capabilities(
+    req: fnv1.RunFunctionRequest,
+    *caps: fnv1.Capability.ValueType,
+) -> list[fnv1.Capability.ValueType]:
+    """Find the capabilities Crossplane doesn't advertise.
+
+    Args:
+        req: The RunFunctionRequest to check.
+        *caps: The capabilities to check for, e.g. fnv1.CAPABILITY_DEPENDENCIES.
+
+    Returns:
+        The capabilities in caps that aren't present in the request metadata,
+        in the order given and without duplicates. Empty if all are present.
+
+    If Crossplane doesn't advertise its capabilities at all, every capability
+    is missing. See advertises_capabilities.
+
+        missing = request.missing_capabilities(req, fnv1.CAPABILITY_DEPENDENCIES)
+    """
+    missing: list[fnv1.Capability.ValueType] = []
+    for cap in caps:
+        if not has_capability(req, cap) and cap not in missing:
+            missing.append(cap)
+    return missing
+
+
+def require_capabilities(
+    req: fnv1.RunFunctionRequest,
+    rsp: fnv1.RunFunctionResponse,
+    *caps: fnv1.Capability.ValueType,
+) -> bool:
+    """Require that Crossplane advertises the supplied capabilities.
+
+    Args:
+        req: The RunFunctionRequest to check.
+        rsp: The RunFunctionResponse to add a fatal result to.
+        *caps: The capabilities the function requires.
+
+    Returns:
+        True if Crossplane advertises every capability in caps. Otherwise
+        False, with one fatal result added to rsp naming every missing
+        capability.
+
+    Use this when a function can't work correctly without a capability -
+    for example because it relies on Crossplane honoring a response field
+    that an older Crossplane would ignore. A fatal result stops the pipeline,
+    which is easier to diagnose than a composition that silently behaves
+    differently.
+
+    A Crossplane that predates capability advertisement (v2.2) has none of
+    them as far as the function can tell, so every capability is missing and
+    the fatal result says so.
+
+        if not request.require_capabilities(req, rsp, fnv1.CAPABILITY_DEPENDENCIES):
+            return rsp
+    """
+    missing = missing_capabilities(req, *caps)
+    if not missing:
+        return True
+
+    msg = (
+        "This function requires Crossplane capabilities that Crossplane doesn't "
+        f"advertise: {', '.join(fnv1.Capability.Name(cap) for cap in missing)}."
+    )
+    if not advertises_capabilities(req):
+        msg += (
+            " Crossplane doesn't advertise any capabilities, so it predates"
+            " capability advertisement (v2.2)."
+        )
+    response.fatal(rsp, msg)
+    return False
+
+
 def get_required_schema(req: fnv1.RunFunctionRequest, name: str) -> dict | None:
     """Get a required OpenAPI schema by name from the request.
 

tests/test_request.py

@@ -342,6 +342,147 @@ class TestCase:
             got = request.has_capability(case.req, case.cap)
             self.assertEqual(case.want, got, case.reason)
 
+    def test_missing_capabilities(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            req: fnv1.RunFunctionRequest
+            caps: list[fnv1.Capability.ValueType]
+            want: list[fnv1.Capability.ValueType]
+
+        cases = [
+            TestCase(
+                reason="Should return nothing when no capabilities are required.",
+                req=fnv1.RunFunctionRequest(),
+                caps=[],
+                want=[],
+            ),
+            TestCase(
+                reason="Should return nothing when every capability is advertised.",
+                req=fnv1.RunFunctionRequest(
+                    meta=fnv1.RequestMeta(
+                        capabilities=[
+                            fnv1.CAPABILITY_CAPABILITIES,
+                            fnv1.CAPABILITY_REQUIRED_SCHEMAS,
+                            fnv1.CAPABILITY_DEPENDENCIES,
+                        ],
+                    ),
+                ),
+                caps=[fnv1.CAPABILITY_DEPENDENCIES, fnv1.CAPABILITY_REQUIRED_SCHEMAS],
+                want=[],
+            ),
+            TestCase(
+                reason="Should return missing capabilities in order, once each.",
+                req=fnv1.RunFunctionRequest(
+                    meta=fnv1.RequestMeta(
+                        capabilities=[
+                            fnv1.CAPABILITY_CAPABILITIES,
+                            fnv1.CAPABILITY_REQUIRED_SCHEMAS,
+                        ],
+                    ),
+                ),
+                caps=[
+                    fnv1.CAPABILITY_DEPENDENCIES,
+                    fnv1.CAPABILITY_REQUIRED_SCHEMAS,
+                    fnv1.CAPABILITY_CREDENTIALS,
+                    fnv1.CAPABILITY_DEPENDENCIES,
+                ],
+                want=[fnv1.CAPABILITY_DEPENDENCIES, fnv1.CAPABILITY_CREDENTIALS],
+            ),
+            TestCase(
+                reason="Should return every capability when none are advertised.",
+                req=fnv1.RunFunctionRequest(),
+                caps=[fnv1.CAPABILITY_DEPENDENCIES, fnv1.CAPABILITY_REQUIRED_SCHEMAS],
+                want=[fnv1.CAPABILITY_DEPENDENCIES, fnv1.CAPABILITY_REQUIRED_SCHEMAS],
+            ),
+        ]
+
+        for case in cases:
+            got = request.missing_capabilities(case.req, *case.caps)
+            self.assertEqual(case.want, got, case.reason)
+
+    def test_require_capabilities(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            req: fnv1.RunFunctionRequest
+            caps: list[fnv1.Capability.ValueType]
+            want: bool
+            want_results: list[fnv1.Result]
+
+        cases = [
+            TestCase(
+                reason="Should pass when no capabilities are required, whatever "
+                "Crossplane advertises.",
+                req=fnv1.RunFunctionRequest(),
+                caps=[],
+                want=True,
+                want_results=[],
+            ),
+            TestCase(
+                reason="Should pass, adding no result, when all are advertised.",
+                req=fnv1.RunFunctionRequest(
+                    meta=fnv1.RequestMeta(
+                        capabilities=[
+                            fnv1.CAPABILITY_CAPABILITIES,
+                            fnv1.CAPABILITY_DEPENDENCIES,
+                        ],
+                    ),
+                ),
+                caps=[fnv1.CAPABILITY_DEPENDENCIES],
+                want=True,
+                want_results=[],
+            ),
+            TestCase(
+                reason="Should fail with one fatal result naming every missing "
+                "capability.",
+                req=fnv1.RunFunctionRequest(
+                    meta=fnv1.RequestMeta(
+                        capabilities=[
+                            fnv1.CAPABILITY_CAPABILITIES,
+                            fnv1.CAPABILITY_REQUIRED_SCHEMAS,
+                        ],
+                    ),
+                ),
+                caps=[
+                    fnv1.CAPABILITY_DEPENDENCIES,
+                    fnv1.CAPABILITY_REQUIRED_SCHEMAS,
+                    fnv1.CAPABILITY_CREDENTIALS,
+                ],
+                want=False,
+                want_results=[
+                    fnv1.Result(
+                        severity=fnv1.SEVERITY_FATAL,
+                        message="This function requires Crossplane capabilities that "
+                        "Crossplane doesn't advertise: CAPABILITY_DEPENDENCIES, "
+                        "CAPABILITY_CREDENTIALS.",
+                    ),
+                ],
+            ),
+            TestCase(
+                reason="Should say Crossplane predates capability advertisement when "
+                "it advertises none.",
+                req=fnv1.RunFunctionRequest(),
+                caps=[fnv1.CAPABILITY_DEPENDENCIES],
+                want=False,
+                want_results=[
+                    fnv1.Result(
+                        severity=fnv1.SEVERITY_FATAL,
+                        message="This function requires Crossplane capabilities that "
+                        "Crossplane doesn't advertise: CAPABILITY_DEPENDENCIES. "
+                        "Crossplane doesn't advertise any capabilities, so it predates "
+                        "capability advertisement (v2.2).",
+                    ),
+                ],
+            ),
+        ]
+
+        for case in cases:
+            rsp = fnv1.RunFunctionResponse()
+            got = request.require_capabilities(case.req, rsp, *case.caps)
+            self.assertEqual(case.want, got, case.reason)
+            self.assertEqual(case.want_results, list(rsp.results), case.reason)
+
     def test_get_required_schema(self) -> None:
         @dataclasses.dataclass
         class TestCase: