Merge pull request #346 from bundlewrap/action-pipe

Actions: Pipes

bundlewrap/items/actions.py

@@ -5,6 +5,7 @@
 
 from bundlewrap.exceptions import ActionFailure, BundleError
 from bundlewrap.items import format_comment, Item
+from bundlewrap.utils import Fault
 from bundlewrap.utils.ui import io
 from bundlewrap.utils.text import mark_for_translation as _
 from bundlewrap.utils.text import blue, bold, wrap_question
@@ -17,6 +18,7 @@ class Action(Item):
     BUNDLE_ATTRIBUTE_NAME = 'actions'
     ITEM_ATTRIBUTES = {
         'command': None,
+        'data_stdin': None,
         'expected_stderr': None,
         'expected_stdout': None,
         'expected_return_code': 0,
@@ -68,7 +70,10 @@ def _get_result(
                 ))
                 return (self.STATUS_SKIPPED, ["unless"])
 
-        question_body = self.attributes['command']
+        question_body = ""
+        if self.attributes['data_stdin'] is not None:
+            question_body += "<" + _("data") + "> | "
+        question_body += self.attributes['command']
         if self.comment:
             question_body += format_comment(self.comment)
 
@@ -128,13 +133,25 @@ def get_result(self, *args, **kwargs):
         return status_code
 
     def run(self):
+        if self.attributes['data_stdin'] is not None:
+            data_stdin = self.attributes['data_stdin']
+            # Allow users to use either a string/unicode object or raw
+            # bytes -- or Faults.
+            if isinstance(data_stdin, Fault):
+                data_stdin = data_stdin.value
+            if type(data_stdin) is not bytes:
+                data_stdin = data_stdin.encode('UTF-8')
+        else:
+            data_stdin = None
+
         with io.job(_("  {node}  {bundle}  {item}  running...").format(
             bundle=self.bundle.name,
             item=self.id,
             node=self.node.name,
         )):
             result = self.bundle.node.run(
                 self.attributes['command'],
+                data_stdin=data_stdin,
                 may_fail=True,
             )
 

bundlewrap/node.py

@@ -665,7 +665,7 @@ def partial_metadata(self):
         """
         return self.repo._metadata_for_node(self.name, partial=True)
 
-    def run(self, command, may_fail=False, log_output=False):
+    def run(self, command, data_stdin=None, may_fail=False, log_output=False):
         if log_output:
             def log_function(msg):
                 io.stdout("{x} {node}  {msg}".format(
@@ -701,6 +701,7 @@ def log_function(msg):
             self.hostname,
             command,
             add_host_keys=add_host_keys,
+            data_stdin=data_stdin,
             ignore_failure=may_fail,
             log_function=log_function,
             wrapper_inner=self.cmd_wrapper_inner,

bundlewrap/operations.py

@@ -90,6 +90,7 @@ def run(
     hostname,
     command,
     add_host_keys=False,
+    data_stdin=None,
     ignore_failure=False,
     log_function=None,
     wrapper_inner="{}",
@@ -138,6 +139,9 @@ def run(
     )
     io._ssh_pids.append(ssh_process.pid)
 
+    if data_stdin is not None:
+        ssh_process.stdin.write(data_stdin)
+
     quit_event = Event()
     stdout_thread = Thread(
         args=(stdout_lb, stdout_fd_r, quit_event, True),

docs/content/items/action.md

@@ -24,6 +24,12 @@ The only required attribute. This is the command that will be run on the node wi
 
 <br>
 
+### data_stdin
+
+You can pipe data directly to the command running on the node. To do so, use this attribute. If it's a string or unicode object, it will always be encoded as UTF-8. Alternatively, you can use raw bytes.
+
+<br>
+
 ### expected_return_code
 
 Defaults to `0`. If the return code of your command is anything else, the action is considered failed. You can also set this to `None` and any return code will be accepted.

tests/integration/bw_apply_actions.py

@@ -46,3 +46,51 @@ def test_action_fail(tmpdir):
         },
     )
     run("bw apply localhost", path=str(tmpdir))
+
+
+def test_action_pipe_binary(tmpdir):
+    make_repo(
+        tmpdir,
+        bundles={
+            "test": {
+                'actions': {
+                    "pipe": {
+                        'command': "cat",
+                        'data_stdin': b"hello\000world",
+                        'expected_stdout': b"hello\000world",
+                    },
+                },
+            },
+        },
+        nodes={
+            "localhost": {
+                'bundles': ["test"],
+                'os': host_os(),
+            },
+        },
+    )
+    run("bw apply localhost", path=str(tmpdir))
+
+
+def test_action_pipe_utf8(tmpdir):
+    make_repo(
+        tmpdir,
+        bundles={
+            "test": {
+                'actions': {
+                    "pipe": {
+                        'command': "cat",
+                        'data_stdin': "hello 🐧\n",
+                        'expected_stdout': "hello 🐧\n",
+                    },
+                },
+            },
+        },
+        nodes={
+            "localhost": {
+                'bundles': ["test"],
+                'os': host_os(),
+            },
+        },
+    )
+    run("bw apply localhost", path=str(tmpdir))