Repository: canonical/operator
Pull request #667 (2021-11-08 16:34:00)

Memory-efficient pull via pebble
To allow for pulling larger files from pebble in a memory-efficient way, I wrote a custom parser which stores incoming file bodies into temporary files.  Other bits are still handled in memory as before.

Headers are still parsed using the same standard library logic.  However, handling of parts has been customized for our purposes.

This may not be the best solution, but maybe it can start discussion on how we want to handle this.

jnsgruk: Despite failing tests, I think this is ready for broader review while the minor test issues are worked through.
petevg: The tests on this one are failing because the regex pattern for our boundary is ```boundary_rgx_pattern = r"--{}(--)?\r\n``` (windows newlines), and the boundary strings we are submitting in the tests have unix newlines. I'm not sure which is correct. (Possibly, we should be checking for both?)
jameinel: I wonder if we are reading a file in text mode vs binary mode. IIRC HTTP
transfer protocols actually specify that it *must* be exactly `\r\n`
(carriage return and line feed). Not just one or the other especially for
headers and boundaries. That sounds like something is genuinely wrong on
Windows.


On Fri, Dec 17, 2021 at 7:31 PM Pen Vander Giessen ***@***.***>
wrote:

> The tests on this one are failing because the regex pattern for our
> boundary is boundary_rgx_pattern = r"--{}(--)?\r\n (windows newlines),
> and the boundary strings we are submitting in the tests have unix newlines.
> I'm not sure which is correct. (Possibly, we should be checking for both?)
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/canonical/operator/pull/667#issuecomment-997106657>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AABRQ7I6FJTNO7W4PHA26GTURPI4PANCNFSM5HTDK7OA>
> .
> Triage notifications on the go with GitHub Mobile for iOS
> <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675>
> or Android
> <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>.
>
> You are receiving this because your review was requested.Message ID:
> ***@***.***>
>

Vultaire: I haven't been able to take a look at this but just noticed the comments.  \r\n is correct at the protocol level.  It sounds like an issue with reading a file in text mode on Windows.  Probably the test in question needs to test using binary data to avoid the conversion of \r\n to \n on Windows.
Vultaire: It looks like one of my latter changesets (changeset 6dc39a82c34e06a4f91d15bd097ebc0859032dbb, "Rework: leave tempfiles open until completion of writing data") broke the "real" pebble tests used for validating this.  I'll try to find time to repair this.
Vultaire: I just did a correction to one of my patches to fix some of the test errors, and am looking at the remaining issues which appear to be coming from test.test_pebble.

Specifically, I'm seeing failures in tests from test.test_pebble.TestClient.  After reviewing the tests, my interpretation is that I'm surprised the tests worked to begin with.  These are tests which leverage the MockClient, which inherits from the ops.pebble.Client modified by this MR.  A number of these tests now fail because of the new code.

Upon looking at the tests, I see multiple issues:
1. We have multi-line bytes literals representing response bodies.  These response bodies do not include carriage returns when checked out with UNIX-style line endings.  This appears to be something that the previous parser was forgiving about, but which is technically incorrect per RFC 2046, which expects CRLF line endings on the boundaries rather than merely LF.
2. The response bodies also include a leading LF.  This does not match the behavior of the requests library; it parses the request headers and strips the particular CRLF, starting the request body on the boundary line.

My next patches will try to correct these tests in a way which should also work if the sources are checked out with Windows-style line endings.
Vultaire: Note: PEP 278 seems to accomodate for Windows vs. UNIX line endings in multiline constants, treating both as LF.
pengale: > I just did a correction to one of my patches to fix some of the test errors, and am looking at the remaining issues which appear to be coming from test.test_pebble.

Thank you for continuing to look into this! I know that you're busy, and I really appreciate not having to catch up on some of the gotchas here right now.

Let me know if you run out of time, and need to hand things off, though ...
Vultaire: I think I've fixed the unit test breakage overall, although I'm not certain about Windows at this point.

There is more rework I wish to do re: Gustavo's points, but for the time being I'm going to have to step away again.  Others are welcome to pick up the torch.  I made some comments in my reply to Gustavo; I also see some bits which seem less readable than I previously thought when I went back to review the code over these last few days, so perhaps some cleanup and refactoring is also in order.
jameinel: b is a byte string, not a raw string.
r'' is a raw string, b is just saying it shouldn't be treated as unicode
chars.


On Wed, Jan 5, 2022 at 4:09 PM Ben Hoyt ***@***.***> wrote:

> ***@***.**** commented on this pull request.
> ------------------------------
>
> In test/test_pebble.py
> <https://github.com/canonical/operator/pull/667#discussion_r779138203>:
>
> > @@ -1481,21 +1481,21 @@ def test_get_services_names(self):
>      def test_pull_text(self):
>          self.client.responses.append((
>              {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
> -            b"""
> ---01234567890123456789012345678901
> -Content-Disposition: form-data; name="files"; filename="/etc/hosts"
> -
> -127.0.0.1 localhost  # """ + b'\xf0\x9f\x98\x80\nfoo\r\nbar' + b"""
> ---01234567890123456789012345678901
> -Content-Disposition: form-data; name="response"
> -
> +            b"""\
> +--01234567890123456789012345678901\r
> +Content-Disposition: form-data; name="files"; filename="/etc/hosts"\r
> +\r
>
> I don't understand how these \r and suchlike are working. Isn't b""" a
> raw string, meaning that there are no escape sequences, and \r will end
> up as a literal backslash followed by a literal r?
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/canonical/operator/pull/667#pullrequestreview-845066103>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AABRQ7MMW2D63ZIV2ZUDPBDUUSXQ3ANCNFSM5HTDK7OA>
> .
> Triage notifications on the go with GitHub Mobile for iOS
> <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675>
> or Android
> <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>.
>
> You are receiving this because your review was requested.Message ID:
> ***@***.***>
>

Vultaire: Summarizing today's changes:
* Cleaned up a couple of tests so they don't leave dangling file handles.
* Primed buffer with CRLF for the special initial case re: boundary detection, letting us include the initial CRLF in the boundary pattern.
* Replaced boundary regex with 2 distinct byte strings, to avoid unexpected issues with escapes and regex.
* Extracted some code into helper functions for reuse or clarity.
* Ultimately moved away from the FSM model in favor of a more straightforward handler function.

Changes were performed against existing unit tests (including the "real pebble" tests); no new tests were added.
benhoyt: Left a bunch of review comments. I think there are a couple of minor things that need fixing, but overall I think this is significantly clearer and simpler than the FSM approach -- nice work!
rwcarlsen: `tempfile.NamedTemporaryFile` says that re-opening a file after closing only works on unix-based systems.  The code as currently written relies on this - do we care about non-unix systems?

Regarding tempfile cleanup - the temp file object supports automatic deletion (when the file is gc'd or closed) - but that doesn't allow the file we return to the caller of `pull` to have different encoding/mode/permissions than those used by the parser when writing the file.

Would it be worth modifying `pull` to optionally take a writeable stream/buffer arg that the parser dumps the file contents into - e.g. 

```python
def pull(self, path: str, dst : io.RawIOBase = None, ...)
    ...

# using pull:
with open('myfile', 'wb') as f:
    container.pull(path, dst=f) # maybe pull does seek(0) on f before returning
    data = f.read()
    ...

# or
buf = container.pull(path, dst=io.StringIO())
data = buf.read()
buf.close()
...
```

seems a little unpythonic (more Go-ey) :-/  Or maybe we just mv the temp file to a non-temp location after parsing is done...
rwcarlsen: All have been addressed except this one:

How should we handle temp file cleanup on windows?  Our options are:

* re-add a `__del__`/gc tempfile unlink/removal - the problem here is that __del__ was actually on the parser object - but we want the file to stick around until the file-object is gc'd which could be long after the parser object is gc'd - so maybe this isn't an option.
* wrap the file in e.g. a `tempfile._TemporaryFileWrapper` that will auto-delete on close or gc before returning it from `pull`
* do nothing - just let the temp file live on in windows environments/containers.

It's still not clear to me under what circumstances charm code would be running on Windows.
jnsgruk: Nice work, Robert. I think option 2 sounds like the most appropriate here.
rwcarlsen: @niemeyer - this has been sitting for a couple weeks.  Are you still planning on taking another look or are we clear to move forward?
rwcarlsen: This should be ready.
rwcarlsen: @niemeyer's comments helped me see that the parser as coded was a bit tricky/unclear to follow.  I reworked it a bit - separating the protocol/pebble message bits away from the mime multipart bits - and I think things are much more clear.  I also added several table-driven tests for the multipart parsing giving me better confidence that things are working like we want.
niemeyer: @jnsgruk @rwcarlsen Five stars for caring about the layout of the code and the design of the parser, but this seems to have completely redesigned the logic from the ground up once more, which means it now needs a proper review once more too. Can you please ask for someone else (Ben? Harry?) to review this logic first so you're not holding on for me right now, as I'm traveling and will be on another sprint in the next few days?
jnsgruk: @niemeyer this is ready for another pass now I believe :)

Patch:
diff --git a/ops/pebble.py b/ops/pebble.py
index 24437d9..b578332 100644
--- a/ops/pebble.py
+++ b/ops/pebble.py
@@ -34,6 +34,7 @@ import shutil
 import signal
 import socket
 import sys
+import tempfile
 import threading
 import time
 import types
@@ -1636,45 +1637,40 @@ class Client:
         if not boundary:
             raise ProtocolError('invalid boundary {!r}'.format(boundary))
 
-        # We have to manually write the Content-Type with boundary, because
-        # email.parser expects the entire multipart message with headers.
-        parser = email.parser.BytesFeedParser()
-        parser.feed(b'Content-Type: multipart/form-data; boundary='
-                    + boundary.encode('utf-8') + b'\r\n\r\n')
+        parser = _FilesParser(boundary)
 
-        # Then read the rest of the response and feed it to the parser.
         while True:
             chunk = response.read(self._chunk_size)
             if not chunk:
                 break
             parser.feed(chunk)
-        message = parser.close()
-
-        # Walk over the multipart parts and read content and metadata.
-        resp = None
-        content = None
-        for part in message.walk():
-            name = part.get_param('name', header='Content-Disposition')
-            if name == 'response':
-                resp = _json_loads(part.get_payload())
-            elif name == 'files':
-                filename = part.get_filename()
-                if filename != path:
-                    raise ProtocolError('path not expected: {}'.format(filename))
-                # decode=True, ironically, avoids decoding bytes to str
-                content = part.get_payload(decode=True)
 
+        resp = parser.get_response()
         if resp is None:
             raise ProtocolError('no "response" field in multipart body')
         self._raise_on_path_error(resp, path)
 
-        if content is None:
+        filenames = parser.filenames()
+        if not filenames:
             raise ProtocolError('no file content in multipart response')
-        if encoding is not None:
-            reader = io.StringIO(content.decode(encoding))
-        else:
-            reader = io.BytesIO(content)
-        return reader
+        elif len(filenames) > 1:
+            raise ProtocolError('single file request resulted in a multi-file response')
+
+        filename = filenames[0]
+        if filename != path:
+            raise ProtocolError('path not expected: {!r}'.format(filename))
+
+        f = parser.get_file(path, encoding)
+
+        # The opened file remains usable after removing/unlinking on posix
+        # platforms until it is closed.  This prevents callers from
+        # interacting with it on disk.  But e.g. Windows doesn't allow
+        # removing opened files, and so we use the tempfile lib's
+        # helper class to auto-delete on close/gc for us.
+        if os.name != 'posix' or sys.platform == 'cygwin':
+            return tempfile._TemporaryFileWrapper(f, f.name, delete=True)
+        parser.remove_files()
+        return f
 
     @staticmethod
     def _raise_on_path_error(resp, path):
@@ -2164,3 +2160,238 @@ class Client:
             query['names'] = names
         resp = self._request('GET', '/v1/checks', query)
         return [CheckInfo.from_dict(info) for info in resp['result']]
+
+
+class _FilesParser:
+    """A limited purpose multi-part parser backed by files for memory efficiency."""
+
+    def __init__(self, boundary: typing.Union[bytes, str]):
+        self._response = None
+        self._files = {}
+
+        # Prepare the MIME multipart boundary line patterns.
+        if isinstance(boundary, str):
+            boundary = boundary.encode()
+
+        # State vars, as we may enter the feed() function multiple times.
+        self._headers = None
+        self._part_type = None
+        self._response_data = bytearray()
+
+        self._max_lookahead = 8 * 1024 * 1024
+
+        self._parser = _MultipartParser(
+            boundary,
+            self._process_header,
+            self._process_body,
+            max_lookahead=self._max_lookahead)
+
+        # RFC 2046 says that the boundary string needs to be preceded by a CRLF.
+        # Unfortunately, the request library's header parsing logic strips off one of
+        # these, so we'll prime the parser buffer with that missing sequence.
+        self._parser.feed(b'\r\n')
+
+    def _process_header(self, data: bytes):
+        parser = email.parser.BytesFeedParser()
+        parser.feed(data)
+        self._headers = parser.close()
+
+        content_disposition = self._headers.get_content_disposition()
+        if content_disposition != 'form-data':
+            raise ProtocolError(
+                'unexpected content disposition: {!r}'.format(content_disposition))
+
+        name = self._headers.get_param('name', header='content-disposition')
+        if name == 'files':
+            filename = self._headers.get_filename()
+            if filename is None:
+                raise ProtocolError('multipart "files" part missing filename')
+            self._prepare_tempfile(filename)
+        elif name != 'response':
+            raise ProtocolError(
+                'unexpected name in content-disposition header: {!r}'.format(name))
+
+        self._part_type = name
+
+    def _process_body(self, data: bytes, done=False):
+        if self._part_type == 'response':
+            self._response_data.extend(data)
+            if done:
+                if len(self._response_data) > self._max_lookahead:
+                    raise ProtocolError('response end marker not found')
+                self._response = json.loads(self._response_data.decode())
+                self._response_data = bytearray()
+        elif self._part_type == 'files':
+            if done:
+                # This is the final write.
+                outfile = self._get_open_tempfile()
+                outfile.write(data)
+                outfile.close()
+                self._headers = None
+            else:
+                # Not the end of file data yet. Don't open/close file for intermediate writes
+                outfile = self._get_open_tempfile()
+                outfile.write(data)
+
+    def remove_files(self):
+        """Remove all temporary files on disk."""
+        for file in self._files.values():
+            os.unlink(file.name)
+        self._files.clear()
+
+    def feed(self, data: bytes):
+        """Provide more data to the running parser."""
+        self._parser.feed(data)
+
+    def _prepare_tempfile(self, filename):
+        tf = tempfile.NamedTemporaryFile(delete=False)
+        self._files[filename] = tf
+        self.current_filename = filename
+
+    def _get_open_tempfile(self):
+        return self._files[self.current_filename]
+
+    def get_response(self):
+        """Return the deserialized JSON object from the multipart "response" field."""
+        return self._response
+
+    def filenames(self):
+        """Return a list of filenames from the "files" parts of the response."""
+        return list(self._files.keys())
+
+    def get_file(self, path, encoding):
+        """Return an open file object containing the data."""
+        mode = 'r' if encoding else 'rb'
+        # We're using text-based file I/O purely for file encoding purposes, not for
+        # newline normalization.  newline='' serves the line endings as-is.
+        newline = '' if encoding else None
+        return open(self._files[path].name, mode, encoding=encoding, newline=newline)
+
+
+class _MultipartParser:
+    def __init__(
+            self,
+            marker: bytes,
+            handle_header,
+            handle_body,
+            max_lookahead: int = 0,
+            max_boundary_length: int = 0):
+        r"""Configures a parser for mime multipart messages.
+
+        Args:
+            marker: the multipart boundary marker (i.e. in "\r\n--<marker>--\r\n")
+
+            handle_header(data): called once with the entire contents of a part
+            header as encountered in data fed to the parser
+
+            handle_body(data, done=False): called incrementally as part body
+            data is fed into the parser - its "done" parameter is set to true when
+            the body is complete.
+
+            max_lookahead: maximum amount of bytes to buffer when searching for a complete header.
+
+            max_boundary_length: maximum number of bytes that can make up a part
+            boundary (e.g. \r\n--<marker>--\r\n")
+        """
+        self._marker = marker
+        self._handle_header = handle_header
+        self._handle_body = handle_body
+        self._max_lookahead = max_lookahead
+        self._max_boundary_length = max_boundary_length
+
+        self._buf = bytearray()
+        self._pos = 0  # current position in buf
+        self._done = False  # whether we have found the terminal boundary and are done parsing
+        self._header_terminator = b'\r\n\r\n'
+
+        # RFC 2046 notes optional "linear whitespace" (e.g. [ \t]+) after the boundary pattern
+        # and the optional "--" suffix.  The boundaries strings can be constructed as follows:
+        #
+        #     boundary = \r\n--<marker>[ \t]+\r\n
+        #     terminal_boundary = \r\n--<marker>--[ \t]+\r\n
+        #
+        # 99 is arbitrarily chosen to represent a max number of linear
+        # whitespace characters to help avoid wrongly writing boundary
+        # characters into a (temporary) file.
+        if not max_boundary_length:
+            self._max_boundary_length = len(b'\r\n--' + marker + b'--\r\n') + 99
+
+    def feed(self, data: bytes):
+        """Feeds data incrementally into the parser."""
+        if self._done:
+            return
+        self._buf.extend(data)
+
+        while True:
+            # seek to a boundary if we aren't already on one
+            i, n, self._done = _next_part_boundary(self._buf, self._marker)
+            if i == -1 or self._done:
+                return  # waiting for more data or terminal boundary reached
+
+            if self._pos == 0:
+                # parse the part header
+                if self._max_lookahead and len(self._buf) - self._pos > self._max_lookahead:
+                    raise ProtocolError('header terminator not found')
+                term_index = self._buf.find(self._header_terminator)
+                if term_index == -1:
+                    return  # waiting for more data
+
+                start = i + n
+                # data includes the double CRLF at the end of the header.
+                end = term_index + len(self._header_terminator)
+
+                self._handle_header(self._buf[start:end])
+                self._pos = end
+            else:
+                # parse the part body
+                ii, nn, self._done = _next_part_boundary(self._buf, self._marker, start=self._pos)
+                safe_bound = max(0, len(self._buf) - self._max_boundary_length)
+                if ii != -1:
+                    # part body is finished
+                    self._handle_body(self._buf[self._pos:ii], done=True)
+                    self._buf = self._buf[ii:]
+                    self._pos = 0
+                    if self._done:
+                        return  # terminal boundary reached
+                elif safe_bound > self._pos:
+                    # write partial body data
+                    data = self._buf[self._pos:safe_bound]
+                    self._pos = safe_bound
+                    self._handle_body(data)
+                    return  # waiting for more data
+                else:
+                    return  # waiting for more data
+
+
+def _next_part_boundary(buf, marker, start=0):
+    """Returns the index of the next boundary marker in buf beginning at start.
+
+    Returns:
+        (index, length, is_terminal) or (-1, -1, False) if no boundary is found.
+    """
+    prefix = b'\r\n--' + marker
+    suffix = b'\r\n'
+    terminal_midfix = b'--'
+
+    i = buf.find(prefix, start)
+    if i == -1:
+        return -1, -1, False
+
+    pos = i + len(prefix)
+    is_terminal = False
+    if buf[pos:].startswith(terminal_midfix):
+        is_terminal = True
+        pos += len(terminal_midfix)
+
+    # Note: RFC 2046 notes optional "linear whitespace" (e.g. [ \t]) after the boundary pattern
+    # and the optional "--" suffix.
+    tail = buf[pos:]
+    for c in tail:
+        if c not in b' \t':
+            break
+        pos += 1
+
+    if buf[pos:].startswith(suffix):
+        pos += len(suffix)
+        return i, pos - i, is_terminal
+    return -1, -1, False
diff --git a/tox.ini b/tox.ini
index 4c193dd..f9c1cb8 100644
--- a/tox.ini
+++ b/tox.ini
@@ -83,4 +83,4 @@ deps =
     logassert
     -r{toxinidir}/requirements.txt
 commands =
-    bash -c "(pebble run --create-dirs &>/dev/null & ) ; sleep 1; pytest -v --tb native -k RealPebble {posargs} ; killall -y 3m pebble"
+    bash -c "umask 0; (pebble run --http=':4000' --create-dirs &>/dev/null & ) ; sleep 1; pytest -v --tb native -k RealPebble {posargs} ; killall -y 3m pebble"


Test patch:
diff --git a/test/test_pebble.py b/test/test_pebble.py
index 9f7365a..8c57a00 100644
--- a/test/test_pebble.py
+++ b/test/test_pebble.py
@@ -30,6 +30,7 @@ import unittest.mock
 import unittest.util
 import urllib.error
 import urllib.request
+import uuid
 
 import pytest
 
@@ -1094,6 +1095,158 @@ def build_mock_change_dict(change_id='70'):
     }
 
 
+class TestMultipartParser(unittest.TestCase):
+    class _Case:
+        def __init__(
+                self,
+                name,
+                data,
+                want_headers,
+                want_bodies,
+                want_bodies_done,
+                max_boundary=14,
+                max_lookahead=8 * 1024,
+                error=''):
+            self.name = name
+            self.data = data
+            self.want_headers = want_headers
+            self.want_bodies = want_bodies
+            self.want_bodies_done = want_bodies_done
+            self.max_boundary = max_boundary
+            self.max_lookahead = max_lookahead
+            self.error = error
+
+    def test_multipart_parser(self):
+        tests = [
+            TestMultipartParser._Case(
+                'baseline',
+                b'\r\n--qwerty\r\nheader foo\r\n\r\nfoo bar\nfoo bar\r\n--qwerty--\r\n',
+                [b'header foo\r\n\r\n'],
+                [b'foo bar\nfoo bar'],
+                want_bodies_done=[True],
+            ),
+            TestMultipartParser._Case(
+                'incomplete header',
+                b'\r\n--qwerty\r\nheader foo\r\n',
+                [],
+                [],
+                want_bodies_done=[],
+            ),
+            TestMultipartParser._Case(
+                'missing header',
+                b'\r\n--qwerty\r\nheader foo\r\n' + 40 * b' ',
+                [],
+                [],
+                want_bodies_done=[],
+                max_lookahead=40,
+                error='header terminator not found',
+            ),
+            TestMultipartParser._Case(
+                'incomplete body terminator',
+                b'\r\n--qwerty\r\nheader foo\r\n\r\nfoo bar\r\n--qwerty\rhello my name is joe and I work in a button factory',  # noqa
+                [b'header foo\r\n\r\n'],
+                [b'foo bar\r\n--qwerty\rhello my name is joe and I work in a '],
+                want_bodies_done=[False],
+            ),
+            TestMultipartParser._Case(
+                'empty body',
+                b'\r\n--qwerty\r\nheader foo\r\n\r\n\r\n--qwerty\r\n',
+                [b'header foo\r\n\r\n'],
+                [b''],
+                want_bodies_done=[True],
+            ),
+            TestMultipartParser._Case(
+                'ignore leading garbage',
+                b'hello my name is joe\r\n\n\n\n\r\n--qwerty\r\nheader foo\r\n\r\nfoo bar\r\n--qwerty\r\n',  # noqa
+                [b'header foo\r\n\r\n'],
+                [b'foo bar'],
+                want_bodies_done=[True],
+            ),
+            TestMultipartParser._Case(
+                'ignore trailing garbage',
+                b'\r\n--qwerty\r\nheader foo\r\n\r\nfoo bar\r\n--qwerty\r\nhello my name is joe',
+                [b'header foo\r\n\r\n'],
+                [b'foo bar'],
+                want_bodies_done=[True],
+            ),
+            TestMultipartParser._Case(
+                'boundary allow linear whitespace',
+                b'\r\n--qwerty \t \r\nheader foo\r\n\r\nfoo bar\r\n--qwerty\r\n',
+                [b'header foo\r\n\r\n'],
+                [b'foo bar'],
+                want_bodies_done=[True],
+                max_boundary=20,
+            ),
+            TestMultipartParser._Case(
+                'terminal boundary allow linear whitespace',
+                b'\r\n--qwerty\r\nheader foo\r\n\r\nfoo bar\r\n--qwerty-- \t \r\n',
+                [b'header foo\r\n\r\n'],
+                [b'foo bar'],
+                want_bodies_done=[True],
+                max_boundary=20,
+            ),
+            TestMultipartParser._Case(
+                'multiple parts',
+                b'\r\n--qwerty \t \r\nheader foo\r\n\r\nfoo bar\r\n--qwerty\r\nheader bar\r\n\r\nfoo baz\r\n--qwerty--\r\n',  # noqa
+                [b'header foo\r\n\r\n', b'header bar\r\n\r\n'],
+                [b'foo bar', b'foo baz'],
+                want_bodies_done=[True, True],
+            ),
+            TestMultipartParser._Case(
+                'ignore after terminal boundary',
+                b'\r\n--qwerty \t \r\nheader foo\r\n\r\nfoo bar\r\n--qwerty--\r\nheader bar\r\n\r\nfoo baz\r\n--qwerty--\r\n',  # noqa
+                [b'header foo\r\n\r\n'],
+                [b'foo bar'],
+                want_bodies_done=[True],
+            ),
+        ]
+
+        chunk_sizes = [1, 2, 3, 4, 5, 7, 13, 17, 19, 23, 29, 31, 37, 42, 50, 100, 1000]
+        marker = b'qwerty'
+        for i, test in enumerate(tests):
+            for chunk_size in chunk_sizes:
+                headers = []
+                bodies = []
+                bodies_done = []
+
+                def handle_header(data):
+                    headers.append(bytes(data))
+                    bodies.append(bytes())
+                    bodies_done.append(False)
+
+                def handle_body(data, done=False):
+                    bodies[-1] += data
+                    bodies_done[-1] = done
+
+                parser = pebble._MultipartParser(
+                    marker,
+                    handle_header,
+                    handle_body,
+                    max_boundary_length=test.max_boundary,
+                    max_lookahead=test.max_lookahead)
+                src = io.BytesIO(test.data)
+
+                try:
+                    while True:
+                        data = src.read(chunk_size)
+                        if not data:
+                            break
+                        parser.feed(data)
+                except Exception as err:
+                    if not test.error:
+                        self.fail('unexpected error:', err)
+                        break
+                    self.assertEqual(test.error, err.message())
+                else:
+                    if test.error:
+                        self.fail('missing expected error: {!r}'.format(test.error))
+
+                    msg = 'test case {} ({}), chunk size {}'.format(i + 1, test.name, chunk_size)
+                    self.assertEqual(test.want_headers, headers, msg)
+                    self.assertEqual(test.want_bodies, bodies, msg)
+                    self.assertEqual(test.want_bodies_done, bodies_done, msg)
+
+
 class TestClient(unittest.TestCase):
     maxDiff = None
 
@@ -1679,28 +1832,60 @@ services:
             ('GET', '/v1/services', {'names': 'svc2'}, None),
         ])
 
-    def test_pull_text(self):
+    def test_pull_boundary_spanning_chunk(self):
         self.client.responses.append((
             {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
-            b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="files"; filename="/etc/hosts"
+            b"""\
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="files"; filename="/etc/hosts"\r
+\r
+127.0.0.1 localhost  # \xf0\x9f\x98\x80\nfoo\r\nbar\r
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="response"\r
+\r
+{
+    "result": [{"path": "/etc/hosts"}],
+    "status": "OK",
+    "status-code": 200,
+    "type": "sync"
+}\r
+--01234567890123456789012345678901--\r
+""",
+        ))
 
-127.0.0.1 localhost  # """ + b'\xf0\x9f\x98\x80\nfoo\r\nbar' + b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="response"
+        self.client._chunk_size = 13
+        with self.client.pull('/etc/hosts') as infile:
+            content = infile.read()
+        self.assertEqual(content, '127.0.0.1 localhost  # 😀\nfoo\r\nbar')
 
+        self.assertEqual(self.client.requests, [
+            ('GET', '/v1/files', {'action': 'read', 'path': '/etc/hosts'},
+                {'Accept': 'multipart/form-data'}, None),
+        ])
+
+    def test_pull_text(self):
+        self.client.responses.append((
+            {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
+            b"""\
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="files"; filename="/etc/hosts"\r
+\r
+127.0.0.1 localhost  # \xf0\x9f\x98\x80\nfoo\r\nbar\r
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="response"\r
+\r
 {
     "result": [{"path": "/etc/hosts"}],
     "status": "OK",
     "status-code": 200,
     "type": "sync"
-}
---01234567890123456789012345678901--
+}\r
+--01234567890123456789012345678901--\r
 """,
         ))
 
-        content = self.client.pull('/etc/hosts').read()
+        with self.client.pull('/etc/hosts') as infile:
+            content = infile.read()
         self.assertEqual(content, '127.0.0.1 localhost  # 😀\nfoo\r\nbar')
 
         self.assertEqual(self.client.requests, [
@@ -1711,25 +1896,26 @@ Content-Disposition: form-data; name="response"
     def test_pull_binary(self):
         self.client.responses.append((
             {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
-            b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="files"; filename="/etc/hosts"
-
-127.0.0.1 localhost  # """ + b'\xf0\x9f\x98\x80\nfoo\r\nbar' + b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="response"
-
+            b"""\
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="files"; filename="/etc/hosts"\r
+\r
+127.0.0.1 localhost  # \xf0\x9f\x98\x80\nfoo\r\nbar\r
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="response"\r
+\r
 {
     "result": [{"path": "/etc/hosts"}],
     "status": "OK",
     "status-code": 200,
     "type": "sync"
-}
---01234567890123456789012345678901--
+}\r
+--01234567890123456789012345678901--\r
 """,
         ))
 
-        content = self.client.pull('/etc/hosts', encoding=None).read()
+        with self.client.pull('/etc/hosts', encoding=None) as infile:
+            content = infile.read()
         self.assertEqual(content, b'127.0.0.1 localhost  # \xf0\x9f\x98\x80\nfoo\r\nbar')
 
         self.assertEqual(self.client.requests, [
@@ -1740,10 +1926,10 @@ Content-Disposition: form-data; name="response"
     def test_pull_path_error(self):
         self.client.responses.append((
             {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
-            b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="response"
-
+            b"""\
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="response"\r
+\r
 {
     "result": [
         {"path": "/etc/hosts", "error": {"kind": "not-found", "message": "not found"}}
@@ -1751,8 +1937,8 @@ Content-Disposition: form-data; name="response"
     "status": "OK",
     "status-code": 200,
     "type": "sync"
-}
---01234567890123456789012345678901--
+}\r
+--01234567890123456789012345678901--\r
 """,
         ))
 
@@ -1782,26 +1968,35 @@ Content-Disposition: form-data; name="response"
 
         self.client.responses.append((
             {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
-            b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="files"; filename="/bad"
-
-bad path
---01234567890123456789012345678901--
+            b"""\
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="files"; filename="/bad"\r
+\r
+bad path\r
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="response"\r
+\r
+{
+    "result": [{"path": "/etc/hosts"}],
+    "status": "OK",
+    "status-code": 200,
+    "type": "sync"
+}\r
+--01234567890123456789012345678901--\r
 """,
         ))
         with self.assertRaises(pebble.ProtocolError) as cm:
             self.client.pull('/etc/hosts')
-        self.assertEqual(str(cm.exception), "path not expected: /bad")
+        self.assertEqual(str(cm.exception), "path not expected: '/bad'")
 
         self.client.responses.append((
             {'Content-Type': 'multipart/form-data; boundary=01234567890123456789012345678901'},
-            b"""
---01234567890123456789012345678901
-Content-Disposition: form-data; name="files"; filename="/etc/hosts"
-
-bad path
---01234567890123456789012345678901--
+            b"""\
+--01234567890123456789012345678901\r
+Content-Disposition: form-data; name="files"; filename="/etc/hosts"\r
+\r
+bad path\r
+--01234567890123456789012345678901--\r
 """,
         ))
         with self.assertRaises(pebble.ProtocolError) as cm:
@@ -3129,6 +3324,15 @@ class TestRealPebble(unittest.TestCase):
         self.assertEqual(out, b'FOO\nBAR\n')
         self.assertEqual(err, b'')
 
+    def test_push_pull(self):
+        fname = os.path.join(tempfile.gettempdir(), 'pebbletest-{}'.format(uuid.uuid4()))
+        content = 'foo\nbar\nbaz-42'
+        self.client.push(fname, content)
+        with self.client.pull(fname) as f:
+            data = f.read()
+            self.assertEqual(data, content)
+        os.remove(fname)
+
     def test_exec_timeout(self):
         process = self.client.exec(['sleep', '0.2'], timeout=0.1)
         with self.assertRaises(pebble.ChangeError) as cm: