Merge lp:~zyga/checkbox/fix-1401996 into lp:checkbox

[Zygmunt Krynicki]

75c3fff plainbox:vendor:extcmd: respect the on_line() protocol

[Daniel Manrique (roadmr)]

Command to test:
plainbox -C -T plainbox.ctrl run -i 2013.com.canonical.certification::cdimage

Before the patch:
INFO plainbox.ctrl: Storing resource record '2013.com.canonical.certification::cdimage': Resource({'official': 'Alpha\narchitecture: amd64\ncodename: Trusty Tahr\nrelease: 14.04 LTS\ndate: 20140304\ndistributor: Ubuntu'})

After:
INFO plainbox.ctrl: Storing resource record '2013.com.canonical.certification::cdimage': Resource({'date': '20140304', 'codename': 'Trusty Tahr', 'architecture': 'amd64', 'distributor': 'Ubuntu', 'official': 'Alpha', 'release': '14.04 LTS'})

so +1 on functionality, and the code is quite clear and easy to follow (yay for a nice, loopy algorithm). My only request would be some sort of automated testing to ensure this doesn't bite us again, but I leave that up to you for reasons discussed on IRC (because... reasons).

review: Approve

[Zygmunt Krynicki (zyga)]

There are no unit tests for extcmd that would cover this functionality (ideas on how to unit-test this are welcome though, I have none). We do have some tests that caught this, namely, the regression test provider.

Without this patch it breaks horribly, with this patch it works as expected.

=== modified file 'plainbox/plainbox/vendor/extcmd/glibc.py'
--- plainbox/plainbox/vendor/extcmd/glibc.py	2014-12-12 14:49:06 +0000
+++ plainbox/plainbox/vendor/extcmd/glibc.py	2014-12-12 18:52:47 +0000
@@ -215,17 +215,51 @@
         _logger.debug("Sending SIGQUIT to process %d", pid)
         os.kill(pid, signal.SIGQUIT)
 
-    def _read_pipe(self, fd, name):
+    def _read_pipe(self, fd, name, buffer_map):
         assert name in ('stdout', 'stderr')
         data = os.read(fd, PIPE_BUF)
+        done_reading = len(data) == 0
         _logger.debug("Read %d bytes of data from %s", len(data), name)
-        if len(data) == 0:
-            return True
-        self._delegate.on_line(name, data)
+        buf = buffer_map[name]
+        if buf is not None:
+            # Combine this buffer with the previous one before attempting
+            # to detect newlines. This way, on the outside, we always
+            # return complete lines and no something partial or in the
+            # middle.
+            buf.extend(data)
+        else:
+            buf = bytearray(data)
+        assert isinstance(buf, bytearray)
+        # Split the buffer into lines, the last line (aka line_list[-1])
+        # may be partial and we have to keep it around for the next time
+        # we're called.
+        line_buffer_list = buf.splitlines(True)
+        for line_buffer in line_buffer_list[0:-1]:
+            self._delegate.on_line(name, bytes(line_buffer))
+        if len(line_buffer_list) > 0:
+            last_line_buffer = line_buffer_list[-1]
+            if last_line_buffer.endswith(b'\n') or done_reading:
+                # If the last line is complete (ends with the newline byte)
+                # or this is the last chunk (we're done reading) then send it
+                # out immedaitely.
+                self._delegate.on_line(name, bytes(last_line_buffer))
+                buffer_map[name] = None
+            else:
+                # Otherwise keep the last line around for the next time we're
+                # called
+                buffer_map[name] = last_line_buffer
+        else:
+            buffer_map[name] = None
+        return done_reading
 
     def _loop(self, selector, pid):
         waiting_for = set(['stdout', 'stderr', 'proc'])
         return_code = None
+        # This buffer holds partial data that was not yet published
+        buffer_map = {
+            'stdout': None,
+            'stderr': None,
+        }
         while waiting_for:
             event_list = selector.select()
             for key, events in event_list:
@@ -253,7 +287,7 @@
                             "Unexpected event mask for signalfd: %d", events)
                 else:
                     if events & EVENT_READ:
-                        if self._read_pipe(key.fd, key.data):
+                        if self._read_pipe(key.fd, key.data, buffer_map):
                             _logger.debug(
                                 "pipe %s depleted, unregistering and closing",
                                 key.data)