Merge pull request #1526 from kernelkit/test-action

test: infamy: add optional argument to rpc/actions

test/infamy/netconf.py

@@ -367,14 +367,19 @@ class Device(Transport):
         lyd = mod.parse_data_dict(call, rpc=True)
         return self.call(lyd.print_mem("xml", with_siblings=True, pretty=False))
 
-    def call_action(self, xpath):
-        """Call NETCONF action (contextualized RPC), XML version"""
+    def call_action(self, xpath, input_data=None):
+        """Call NETCONF action (contextualized RPC), XML version.
+
+        If `input_data` is supplied, it's set at the action's xpath as
+        the action's input leaves.  Defaults to an empty input for
+        actions that take no parameters.
+        """
         action={}
         pattern = r"^/(?P<module>[^:]+):(?P<path>[^/]+)"
         match = re.search(pattern, xpath)
         module = match.group('module')
         modpath = f"/{match.group('module')}:{match.group('path')}"
-        libyang.xpath_set(action, xpath, {})
+        libyang.xpath_set(action, xpath, input_data or {})
         mod = self.ly.get_module(module)
         lyd = mod.parse_data_dict(action, rpc=True)
         xml = "<action xmlns=\"urn:ietf:params:xml:ns:yang:1\">" + lyd.print_mem("xml", with_siblings=True, pretty=False) + "</action>"

test/infamy/restconf.py

@@ -452,12 +452,26 @@ class Device(Transport):
     def reboot(self):
         self.call_rpc("ietf-system:system-restart")
 
-    def call_action(self, xpath):
+    def call_action(self, xpath, input_data=None):
         path = xpath_to_uri(xpath)
         url = f"{self.restconf_url}/data{path}"
+
+        # RFC 8040 wraps action input as {"<action-module>:input": {...}}.
+        # The action's module is the prefix of the closest namespaced
+        # xpath segment; for "/a:foo/b:bar/baz" that's "b".
+        body = None
+        if input_data:
+            module = None
+            for seg in xpath.split("/"):
+                if ":" in seg:
+                    module = seg.split(":", 1)[0]
+            if not module:
+                raise ValueError(f"cannot determine action module from {xpath}")
+            body = {f"{module}:input": input_data}
+
         response = requests_workaround_post(
             url,
-            json=None,
+            json=body,
             headers=self.headers,
             auth=self.auth,
             verify=False

test/infamy/transport.py

@@ -47,7 +47,13 @@ class Transport(ABC):
         pass
 
     @abstractmethod
-    def call_action(self, xpath):
+    def call_action(self, xpath, input_data=None):
+        """Invoke a YANG action at `xpath`.
+
+        `input_data`, if supplied, is a dict of input leaves (e.g.
+        ``{"state": "on"}``).  Backends wrap it in the protocol-specific
+        envelope; actions that take no input may omit the argument.
+        """
         pass
 
     def __getitem__(self, key):