Method test_stream_helper_decoding

tests/unit/api_test.py:328–362

(self)

            self.client.create_host_config(security_opt='wrong')

    def test_stream_helper_decoding(self):
        status_code, content = fake_api.fake_responses[f"{url_prefix}events"]()
        content_str = json.dumps(content)
        content_str = content_str.encode('utf-8')
        body = io.BytesIO(content_str)

        # mock a stream interface
        raw_resp = urllib3.HTTPResponse(body=body)
        raw_resp._fp.chunked = True
        raw_resp._fp.chunk_left = len(body.getvalue()) - 1

        # pass `decode=False` to the helper
        raw_resp._fp.seek(0)
        resp = response(status_code=status_code, content=content, raw=raw_resp)
        result = next(self.client._stream_helper(resp))
        assert result == content_str

        # pass `decode=True` to the helper
        raw_resp._fp.seek(0)
        resp = response(status_code=status_code, content=content, raw=raw_resp)
        result = next(self.client._stream_helper(resp, decode=True))
        assert result == content

        # non-chunked response, pass `decode=False` to the helper
        raw_resp._fp.chunked = False
        raw_resp._fp.seek(0)
        resp = response(status_code=status_code, content=content, raw=raw_resp)
        result = next(self.client._stream_helper(resp))
        assert result == content_str.decode('utf-8')

        # non-chunked response, pass `decode=True` to the helper
        raw_resp._fp.seek(0)
        resp = response(status_code=status_code, content=content, raw=raw_resp)
        result = next(self.client._stream_helper(resp, decode=True))
        assert result == content


class UnixSocketStreamTest(unittest.TestCase):