diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py index 9fbf12d..9ceba3d 100644 --- a/backend/fluksio/flow/nodes/http.py +++ b/backend/fluksio/flow/nodes/http.py @@ -301,7 +301,7 @@ class HttpNode(Node): :type params: dict :param kwargs: Pipeline data to send (from required messages). :type kwargs: Any - :returns: Response data if the endpoint returns JSON, None otherwise. + :returns: The reply's values for the declared output ports, or None. :rtype: dict | None """ client = shared_client() @@ -330,8 +330,13 @@ class HttpNode(Node): except Exception: return None # Outputs are keyed by port, so only an object can be one. A bare - # scalar or list is a valid reply, just not something to publish. - return body if isinstance(body, dict) else None + # scalar or list is a valid reply, just not something to publish, + # and neither is a key no port declares: the reply is the + # endpoint's, and a `status` beside the values is common. + if not isinstance(body, dict): + return None + ports = {spec.port for spec in self.output_ports} + return {key: value for key, value in body.items() if key in ports} or None except httpx.HTTPStatusError as e: logger.error( diff --git a/backend/tests/flow/test_senders.py b/backend/tests/flow/test_senders.py index 1cbfbb8..e59cd6c 100644 --- a/backend/tests/flow/test_senders.py +++ b/backend/tests/flow/test_senders.py @@ -2,8 +2,10 @@ import asyncio +import httpx + from fluksio.flow.messages import DType, MessageSpec -from fluksio.flow.nodes import MqttNode +from fluksio.flow.nodes import HttpNode, MqttNode from fluksio.flow.nodes.http import close_shared_client, shared_client @@ -20,6 +22,28 @@ def test_http_senders_share_one_pooled_client(): close_shared_client() +def test_an_http_reply_keeps_only_the_declared_ports(monkeypatch): + """A `status` beside the value is the endpoint's business, not an error.""" + reply = {"temp": 21.5, "status": "ok"} + client = httpx.Client( + transport=httpx.MockTransport(lambda _: httpx.Response(200, json=reply)) + ) + monkeypatch.setattr("fluksio.flow.nodes.http.shared_client", lambda: client) + + def sender(*provides: MessageSpec) -> HttpNode: + node = HttpNode( + requires=[MessageSpec(name="setpoint", dtype=DType.FLOAT)], + provides=provides, + params={"url": "http://thermostat.test/set"}, + ) + node.assign_flow("heating", "thermostat") + return node + + temp = MessageSpec(name="temp", dtype=DType.FLOAT) + assert sender(temp).execute({"heating.setpoint": 21.0}) == {"heating.temp": 21.5} + assert sender().execute({"heating.setpoint": 21.0}) is None + + def _sender() -> MqttNode: node = MqttNode( requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)], diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 0157174..ea57a43 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -63,7 +63,7 @@ The ports pick the direction: outputs only make a **listener**, inputs a **sender**. A listener is a **webhook**: the engine mounts a route at `/hooks/{flow}/{url}/{secret}` while the flow runs. A sender requests `url` and publishes a JSON object response on its outputs, one key per output port. -A key no port declares fails the call, and a response that is not an object +Keys no port declares are dropped, and a response that is not an object publishes nothing. | Setting | Default | Notes |