#!/usr/bin/env python3 """One interpreter start for the whole of 99-python3-urllib3. Emits `key=value' lines, one per line, in file order; the suite reads them back with grep. A python3 start on this arm is tens of seconds and every measurement below is a string parse, a dict operation or a few hundred bytes through zlib, so the start is the entire cost of the suite and taking it once is the difference between a half-minute run and a ten-minute one. NOTHING HERE OPENS A SOCKET. urllib3 is an HTTP client and the obvious way to exercise it is to make a request, which is exactly what this suite must not do: the on-target verify has no network and a suite that waited for one would hang rather than fail. Every measurement is therefore taken on the half of urllib3 that runs before a connection exists -- URL parsing, retry and timeout arithmetic, header containers, multipart encoding -- or on `HTTPResponse' driven from a `BytesIO' body, which is the seam upstream's own test suite uses to test decoding without a server. TWO RULES THE VALUES OBEY. The runner parses its input line by line, so no value may contain a newline. `one()' collapses any value that might, and the multipart body -- the one measurement that is inherently multi-line -- is never emitted whole: what is asserted about it is a shape (a line count, a substring, a field count), never the text. And no value may depend on the encoding stdout happens to be in. Every value is ASCII. The one non-ASCII fixture is a Cyrillic hostname written as a `\\u' escape, and what is asserted about it is its PUNYCODE, which is ASCII by construction. """ import io import sys import zlib emitted = [] def emit(key, value): emitted.append(key) print('%s=%s' % (key, value)) def one(v): """`v' as a single line, so a stray newline cannot desynchronise the runner.""" return str(v).replace('\r', '\\r').replace('\n', '\\n') def attempt(fn, *a, **kw): """`fn(*a)' as a one-line string, or `!' on a refusal. The exception NAME rather than its message: several of the messages below carry upstream's wording verbatim, and a release that reworded one would read as a behavioural change when nothing about the refusal had moved. The cases that DO care about a message ask for it separately.""" try: return one(fn(*a, **kw)) except Exception as exc: return '!%s' % type(exc).__name__ def msg(fn, *a, **kw): """The exception name and its message, for the cases that are about the message.""" try: fn(*a, **kw) except Exception as exc: return '%s: %s' % (type(exc).__name__, one(exc)) return 'no-exception' try: import urllib3 except Exception as exc: # pragma: no cover print('import_error=%s: %s' % (type(exc).__name__, exc)) sys.exit(1) from urllib3 import exceptions as u3exc from urllib3 import fields as u3fields from urllib3 import filepost as u3filepost from urllib3 import response as u3response from urllib3.util import retry as u3retry from urllib3.util import ssl_ as u3ssl from urllib3.util import timeout as u3timeout from urllib3.util import url as u3url # =========================================================== identity ======== emit('module_file', getattr(urllib3, '__file__', '')) emit('is_package', 'yes' if hasattr(urllib3, '__path__') else 'no') emit('version', urllib3.__version__) emit('version_module', u3url.__name__.split('.')[0]) # `__all__' IS DECLARED, unlike most of this Epic's ports, and it is the # package's own statement of its public surface. Emitted as a sorted # comma-joined list so one added or removed name turns exactly one case red. emit('all_count', len(urllib3.__all__)) emit('all_sorted', ','.join(sorted(urllib3.__all__))) # ==================================================== the image's shape ====== # THE TWO OPTIONAL DECODERS ARE ABSENT ON THIS IMAGE, and that is a property of # the image rather than of urllib3. `response.py' imports brotli and zstandard # inside try/except and degrades silently; neither package is in IMAGE_INSTALL, # so both degradations are live here and both are asserted rather than left to be # discovered by a caller wondering why a `Content-Encoding: br' body came back # undecoded. emit('brotli_absent', 'yes' if u3response.brotli is None else 'no') emit('has_zstd', 'yes' if u3response.HAS_ZSTD else 'no') emit('content_decoders', ','.join(u3response.HTTPResponse.CONTENT_DECODERS)) emit('decoder_count', len(u3response.HTTPResponse.CONTENT_DECODERS)) emit('redirect_statuses', ','.join(str(s) for s in sorted(u3response.HTTPResponse.REDIRECT_STATUSES))) # WHAT THAT MEANS ON THE WIRE: `make_headers(accept_encoding=True)' builds its # value from the decoder list, so this image ADVERTISES a narrower Accept-Encoding # than a desktop urllib3 with brotli installed. This is the one measurement here # that a remote server would see. emit('accept_encoding_default', urllib3.util.make_headers(accept_encoding=True)['accept-encoding']) emit('accept_encoding_explicit', urllib3.util.make_headers(accept_encoding='gzip')['accept-encoding']) # ============================================== what does NOT import ========= # THREE SUBMODULES OF THIS PACKAGE CANNOT BE IMPORTED ON THIS IMAGE, each because # a third-party dependency of it is absent. They ship as .py files regardless -- # the package is installed whole -- so the files being present and the imports # failing are two separate facts and both are asserted. # # `contrib.pyopenssl' is the one that decides this port's RDEPENDS: it is the # ONLY importer of cryptography and pyopenssl in the package, and it is # deprecated upstream. See the bbappend. def _import(name): try: __import__(name) return 'ok' except Exception as exc: return '!%s' % type(exc).__name__ emit('import_contrib', _import('urllib3.contrib')) emit('import_pyopenssl', _import('urllib3.contrib.pyopenssl')) emit('import_socks', _import('urllib3.contrib.socks')) emit('import_http2', _import('urllib3.http2')) emit('import_emscripten', _import('urllib3.contrib.emscripten')) emit('import_util_url', _import('urllib3.util.url')) emit('import_ssltransport', _import('urllib3.util.ssltransport')) emit('import_connection', _import('urllib3.connection')) emit('import_poolmanager', _import('urllib3.poolmanager')) emit('import_request_methods', _import('urllib3._request_methods')) # ================================================= public API presence ======= for _n in ('PoolManager', 'ProxyManager', 'HTTPConnectionPool', 'HTTPSConnectionPool', 'HTTPResponse', 'BaseHTTPResponse', 'HTTPHeaderDict', 'Retry', 'Timeout', 'Rreq'): emit('has_' + _n, 'yes' if hasattr(urllib3, _n) else 'no') for _n in ('request', 'connection_from_url', 'proxy_from_url', 'encode_multipart_formdata', 'make_headers', 'disable_warnings', 'add_stderr_logger'): emit('callable_' + _n, 'yes' if callable(getattr(urllib3, _n, None)) else 'no') # ================================================ exception hierarchy ======== # The hierarchy is what a caller writes `except' against, so each edge is stated # rather than the tree being asserted as a whole: a release that re-parented one # exception should turn one case red and name it. def base(cls): return cls.__mro__[1].__name__ emit('exc_HTTPError_base', base(u3exc.HTTPError)) emit('exc_PoolError_base', base(u3exc.PoolError)) emit('exc_RequestError_base', base(u3exc.RequestError)) emit('exc_MaxRetryError_base', base(u3exc.MaxRetryError)) emit('exc_TimeoutError_base', base(u3exc.TimeoutError)) emit('exc_ConnectTimeoutError_bases', ','.join(c.__name__ for c in u3exc.ConnectTimeoutError.__mro__[1:3])) emit('exc_ReadTimeoutError_is_httperror', 'yes' if issubclass(u3exc.ReadTimeoutError, u3exc.HTTPError) else 'no') emit('exc_LocationParseError_base', base(u3exc.LocationParseError)) emit('exc_NewConnectionError_bases', ','.join(c.__name__ for c in u3exc.NewConnectionError.__mro__[1:3])) emit('exc_ProtocolError_base', base(u3exc.ProtocolError)) emit('exc_ProtocolError_is_ConnectionError', 'yes' if u3exc.ProtocolError is u3exc.ConnectionError else 'no') emit('exc_DecodeError_base', base(u3exc.DecodeError)) emit('exc_SSLError_base', base(u3exc.SSLError)) emit('exc_ResponseError_base', base(u3exc.ResponseError)) emit('exc_HTTPWarning_base', base(u3exc.HTTPWarning)) emit('exc_InsecureRequestWarning_base', base(u3exc.InsecureRequestWarning)) emit('exc_IncompleteRead_bases', ','.join(c.__name__ for c in u3exc.IncompleteRead.__mro__[1:3])) emit('exc_InvalidChunkLength_bases', ','.join(c.__name__ for c in u3exc.InvalidChunkLength.__mro__[1:3])) # `HTTPError' is NOT an OSError, which is the assumption a caller migrating from # the stdlib makes most often. emit('exc_HTTPError_is_oserror', 'yes' if issubclass(u3exc.HTTPError, OSError) else 'no') # ==================================================== URL parsing ============ # The largest cheap surface in the package and the one most callers touch # first. `parse_url' is a pure function over a string; every case below is one # call and the whole block costs microseconds. def purl(s, attr): try: return one(getattr(u3url.parse_url(s), attr)) except Exception as exc: return '!%s' % type(exc).__name__ emit('url_scheme', purl('http://a.com/x', 'scheme')) emit('url_host', purl('http://a.com/x', 'host')) emit('url_path', purl('http://a.com/x', 'path')) emit('url_port_none', purl('http://a.com/x', 'port')) emit('url_port', purl('http://a.com:8080/x', 'port')) emit('url_port_type', attempt(lambda: type(u3url.parse_url('http://a.com:8080/').port).__name__)) emit('url_query', purl('http://a.com/x?y=1', 'query')) emit('url_fragment', purl('http://a.com/x#f', 'fragment')) emit('url_auth', purl('http://u:p@a.com/', 'auth')) emit('url_netloc', purl('http://u:p@a.com:81/x', 'netloc')) emit('url_request_uri', purl('http://a.com/x?y=1#f', 'request_uri')) emit('url_url_roundtrip', purl('http://a.com/x?y=1#f', 'url')) emit('url_str', attempt(lambda: str(u3url.parse_url('http://a.com/x?y=1')))) # SCHEME AND HOST ARE LOWERCASED, the path is not. This is the pair that bites # a caller comparing two parsed URLs for equality. emit('url_scheme_lowered', purl('HTTP://A.COM/X', 'scheme')) emit('url_host_lowered', purl('HTTP://A.COM/X', 'host')) emit('url_path_case_kept', purl('HTTP://A.COM/X', 'path')) # THE EMPTY PATH IS `None', NOT `"/"'. urllib3 does not normalise it, and # `request_uri' is what supplies the "/" a request line needs. emit('url_empty_path', purl('http://a.com', 'path')) emit('url_empty_path_request_uri', purl('http://a.com', 'request_uri')) emit('url_root_path', purl('http://a.com/', 'path')) # A BARE HOST WITH NO SCHEME PARSES, and what it does with a colon is the # surprise: `a.com:80' is host+port, not scheme+path. emit('url_bare_host', purl('a.com', 'host')) emit('url_bare_scheme', purl('a.com', 'scheme')) emit('url_bare_hostport_host', purl('a.com:80', 'host')) emit('url_bare_hostport_port', purl('a.com:80', 'port')) emit('url_relative_path', purl('/x/y', 'path')) emit('url_relative_host', purl('/x/y', 'host')) # IPv6 LITERALS keep their brackets in `host'. emit('url_ipv6_host', purl('http://[::1]/x', 'host')) emit('url_ipv6_port', purl('http://[::1]:8080/x', 'port')) emit('url_ipv6_netloc', purl('http://[::1]:8080/x', 'netloc')) emit('url_ipv4_host', purl('http://127.0.0.1/x', 'host')) # PERCENT-ESCAPES ARE UPPERCASED AND NOTHING IS DECODED. Measured, not assumed: # `%2f' comes back `%2F' and `%7E' comes back `%7E' rather than as the `~' it # stands for, so two URLs that differ only in whether a byte was escaped stay # different after parsing. Unescaped bytes that need escaping are escaped. emit('url_pct_upper', purl('http://a.com/%2f', 'path')) emit('url_pct_unreserved', purl('http://a.com/%7Euser', 'path')) emit('url_space_encoded', purl('http://a.com/a b', 'path')) emit('url_query_encoded', purl('http://a.com/?a=b c', 'query')) # REFUSALS. Each of these is a `LocationParseError' and the suite states which # input produces one, because the set is not obvious: an empty port is fine, a # non-numeric one is not. emit('url_empty_port', purl('http://a.com:/x', 'port')) emit('url_bad_port', purl('http://a.com:x/', 'host')) emit('url_port_out_of_range', purl('http://a.com:70000/', 'host')) emit('url_bad_ipv6', purl('http://[::1/', 'host')) emit('url_null_byte', purl('http://a.com/\x00', 'host')) emit('url_bad_port_message', msg(u3url.parse_url, 'http://a.com:x/')) # THE IDNA PATH IS THE ONE LAZY IMPORT IN THE PACKAGE and the reason # `python3-idna' stays in RDEPENDS. `_idna_encode' is called only for a # non-ASCII host label; without the module it raises rather than resolving. The # fixture is a Cyrillic hostname and the assertion is its ASCII punycode. emit('url_idna_host', purl('http://\u043f\u0440\u0438\u043c\u0435\u0440.\u0440\u0444/', 'host')) emit('url_idna_scheme', purl('http://\u043f\u0440\u0438\u043c\u0435\u0440.\u0440\u0444/', 'scheme')) emit('url_idna_mixed', purl('http://www.\u043f\u0440\u0438\u043c\u0435\u0440.\u0440\u0444/x', 'host')) emit('url_idna_module_importable', _import('idna')) emit('url_ascii_host_no_idna', purl('http://xn--e1afmkfd.xn--p1ai/', 'host')) # `Url' IS A NAMEDTUPLE, so it compares by value and unpacks positionally. A # caller memoising parsed URLs depends on both. emit('url_is_tuple', 'yes' if isinstance(u3url.parse_url('http://a.com/'), tuple) else 'no') emit('url_fields', ','.join(u3url.Url._fields)) emit('url_eq', 'yes' if u3url.parse_url('http://a.com/') == u3url.parse_url('http://a.com/') else 'no') emit('url_hostname_alias', 'yes' if u3url.parse_url('http://a.com/').hostname == 'a.com' else 'no') # ==================================================== Retry ================== # Retry arithmetic is pure and is the part of urllib3 a caller is most likely to # get wrong, so it is measured rather than described. R = u3retry.Retry emit('retry_default_total', one(R.DEFAULT.total)) emit('retry_default_connect', one(R.DEFAULT.connect)) emit('retry_backoff_max', one(R.DEFAULT_BACKOFF_MAX)) emit('retry_allowed_methods', ','.join(sorted(R.DEFAULT_ALLOWED_METHODS))) emit('retry_status_forcelist', ','.join(str(s) for s in sorted(R.RETRY_AFTER_STATUS_CODES))) emit('retry_default_redirect', one(R.DEFAULT.redirect)) emit('retry_raise_on_redirect', one(R.DEFAULT.raise_on_redirect)) emit('retry_raise_on_status', one(R.DEFAULT.raise_on_status)) emit('retry_respect_retry_after', one(R.DEFAULT.respect_retry_after_header)) # BACKOFF IS ZERO UNTIL THE SECOND RETRY. `get_backoff_time' returns 0 while # fewer than two entries are in the history, which is the off-by-one every # caller reading "exponential backoff" gets wrong. def backoff(n, factor=1.0): # `total=None' is unlimited, so twenty increments do not raise. The backoff # is a function of the HISTORY length alone, which is what is being measured. r = R(total=None, backoff_factor=factor) for _ in range(n): r = r.increment(method='GET', url='/') return r.get_backoff_time() emit('retry_backoff_0', one(backoff(0))) emit('retry_backoff_1', one(backoff(1))) emit('retry_backoff_2', one(backoff(2))) emit('retry_backoff_3', one(backoff(3))) emit('retry_backoff_4', one(backoff(4))) emit('retry_backoff_capped', one(backoff(20))) emit('retry_backoff_factor_half_2', one(backoff(2, 0.5))) emit('retry_is_exhausted_fresh', one(R(total=1).is_exhausted())) emit('retry_total_after_increment', one(R(total=2).increment(method='GET', url='/').total)) emit('retry_exhausted_raises', msg(lambda: R(total=0).increment(method='GET', url='/')).split(':')[0]) emit('retry_history_len', one(len(R(total=3).increment(method='GET', url='/') .increment(method='GET', url='/').history))) emit('retry_history_type', one(type(R(total=3).increment(method='GET', url='/').history[0]).__name__)) emit('retry_from_int_total', one(R.from_int(3).total)) emit('retry_from_int_none', one(R.from_int(None, default=R(total=7)).total)) emit('retry_from_int_bool_true', one(type(R.from_int(True)).__name__)) emit('retry_repr_prefix', one(repr(R(total=3)).split('(')[0])) # `is_retry' IS METHOD-SENSITIVE: POST is not in DEFAULT_ALLOWED_METHODS, so an # identical 503 retries for GET and does not for POST. emit('retry_is_retry_get_503', one(R(total=3, status_forcelist=[503]).is_retry('GET', 503))) emit('retry_is_retry_post_503', one(R(total=3, status_forcelist=[503]).is_retry('POST', 503))) emit('retry_is_retry_get_200', one(R(total=3, status_forcelist=[503]).is_retry('GET', 200))) emit('retry_parse_retry_after_seconds', one(R().parse_retry_after('120'))) emit('retry_parse_retry_after_bad', msg(R().parse_retry_after, 'not-a-date').split(':')[0]) # ==================================================== Timeout ================ T = u3timeout.Timeout emit('timeout_default_sentinel', one(T.DEFAULT_TIMEOUT)) emit('timeout_from_float_connect', one(T.from_float(3.0).connect_timeout)) emit('timeout_from_float_read', one(T.from_float(3.0).read_timeout)) emit('timeout_total_none', one(T(connect=1, read=2).total)) emit('timeout_clone_type', one(type(T(connect=1, read=2).clone()).__name__)) emit('timeout_negative_refused', msg(T, connect=-1).split(':')[0]) emit('timeout_string_refused', msg(T, connect='x').split(':')[0]) emit('timeout_bool_refused', msg(T, connect=True).split(':')[0]) emit('timeout_repr_prefix', one(repr(T(connect=1, read=2)).split('(')[0])) # `read_timeout' WITH A TOTAL BUDGET AND NO `start_connect' DOES NOT RAISE and # does not consult the total -- it hands back the bare read timeout. So a # Timeout(total=5, read=2) reused across requests without being re-started # enforces 2s per read and no overall budget at all, silently. The state machine # only objects to being STARTED twice, which is the other case here. emit('timeout_read_without_start', one(T(connect=1, read=2, total=5).read_timeout)) emit('timeout_read_after_start', one((lambda t: (t.start_connect(), t.read_timeout)[1])(T(read=2, total=5)))) emit('timeout_start_twice', msg(lambda: (lambda t: (t.start_connect(), t.start_connect()))(T(total=5))).split(':')[0]) # ============================================= HTTPHeaderDict ================ # The multi-valued, case-insensitive mapping every response carries. Its # divergences from `dict' are the point: `add' accumulates where `[]=' replaces, # and reading back a multi-valued header with `[]' joins with ", ". H = urllib3.HTTPHeaderDict _h = H() _h['Content-Type'] = 'text/plain' emit('hdr_case_insensitive_get', one(_h['content-type'])) emit('hdr_case_preserved', ','.join(_h.keys())) emit('hdr_in_lower', one('content-type' in _h)) _h.add('Set-Cookie', 'a=1') _h.add('set-cookie', 'b=2') emit('hdr_add_joins', one(_h['Set-Cookie'])) emit('hdr_getlist', ','.join(_h.getlist('set-cookie'))) emit('hdr_getlist_absent', one(_h.getlist('nope'))) emit('hdr_len', one(len(_h))) _h['Set-Cookie'] = 'c=3' emit('hdr_setitem_replaces', one(_h['Set-Cookie'])) emit('hdr_eq_case', one(H({'A': '1'}) == H({'a': '1'}))) emit('hdr_eq_plain_dict', one(H({'A': '1'}) == {'a': '1'})) emit('hdr_copy_type', one(type(H({'A': '1'}).copy()).__name__)) emit('hdr_discard_missing', attempt(H({'A': '1'}).discard, 'B')) emit('hdr_from_pairs', one(H([('a', '1'), ('a', '2')])['a'])) emit('hdr_repr_prefix', one(repr(H({'A': '1'})).split('(')[0])) # ============================================= fields / multipart ============ # `encode_multipart_formdata' is pure and is what every file upload goes # through. The body is multi-line so only its SHAPE is emitted. _body, _ctype = u3filepost.encode_multipart_formdata( [('f', 'v'), ('file', ('n.txt', b'hello', 'text/plain'))], boundary='BOUND') _text = _body.decode('utf-8') emit('mp_content_type', _ctype) emit('mp_body_is_bytes', 'yes' if isinstance(_body, bytes) else 'no') emit('mp_line_count', one(len(_text.split('\r\n')))) emit('mp_crlf_only', 'yes' if '\n' not in _text.replace('\r\n', '') else 'no') emit('mp_boundary_count', one(_text.count('--BOUND'))) emit('mp_has_terminator', 'yes' if _text.endswith('--BOUND--\r\n') else 'no') emit('mp_has_filename', 'yes' if 'filename="n.txt"' in _text else 'no') emit('mp_has_name', 'yes' if 'name="f"' in _text else 'no') emit('mp_declared_ctype', 'yes' if 'Content-Type: text/plain' in _text else 'no') emit('mp_payload_present', 'yes' if 'hello' in _text else 'no') emit('mp_random_boundary_len', one(len(u3filepost.encode_multipart_formdata([('a', 'b')])[1].split('=')[1]))) # `render_headers' READS `self.headers', WHICH IS EMPTY UNTIL `make_multipart' # HAS RUN. A field built and rendered without that call emits no headers at all # and no error -- both halves are asserted, because the empty render is what a # caller assembling a body by hand gets. _rf_bare = u3fields.RequestField('name', b'data', filename='f.txt') emit('rf_headers_before_multipart', one(len([l for l in _rf_bare.render_headers().split('\r\n') if l]))) _rf = u3fields.RequestField('name', b'data', filename='f.txt') _rf.make_multipart(content_type='text/plain') emit('rf_render_headers_lines', one(len([l for l in _rf.render_headers().split('\r\n') if l]))) emit('rf_disposition', one([l for l in _rf.render_headers().split('\r\n') if l.startswith('Content-Disposition')][0])) emit('rf_from_tuples_type', one(type(u3fields.RequestField.from_tuples('k', 'v')).__name__)) emit('rf_guess_content_type_txt', one(u3fields.guess_content_type('a.txt'))) emit('rf_guess_content_type_unknown', one(u3fields.guess_content_type('a.nosuchext'))) emit('rf_guess_content_type_none', one(u3fields.guess_content_type(None))) # RFC 2231 vs RFC 7578: urllib3 2.x defaults to the HTML5 form of a header # parameter, which percent-encodes rather than emitting a `name*=' parameter. emit('rf_param_ascii', one(u3fields.format_multipart_header_param('name', 'v'))) emit('rf_param_quote', one(u3fields.format_multipart_header_param('name', 'a"b'))) emit('rf_param_newline', one(u3fields.format_multipart_header_param('name', 'a\nb'))) emit('rf_param_rfc2231_ascii', one(u3fields.format_header_param_rfc2231('name', 'v'))) emit('rf_param_rfc2231_utf8', one(u3fields.format_header_param_rfc2231('name', '\u00e9'))) # ============================================= response decoding ============= # `HTTPResponse' driven from a `BytesIO' is the seam that exercises decoding with # no server. This is where the missing brotli/zstd decoders become visible as # behaviour rather than as a flag. def mkresp(body, headers=None, **kw): return u3response.HTTPResponse(body=io.BytesIO(body), headers=headers or {}, preload_content=False, **kw) _plain = mkresp(b'hello') emit('resp_read_all', one(_plain.read().decode())) emit('resp_status_default', one(mkresp(b'').status)) emit('resp_closed_after_read', one(_plain.closed)) _gzobj = zlib.compressobj(9, zlib.DEFLATED, 16 + zlib.MAX_WBITS) _gzbytes = _gzobj.compress(b'hello gzip') + _gzobj.flush() _dfbytes = zlib.compress(b'hello deflate') emit('resp_gzip', one(mkresp(_gzbytes, {'content-encoding': 'gzip'}).read().decode())) emit('resp_gzip_xgzip', one(mkresp(_gzbytes, {'content-encoding': 'x-gzip'}).read().decode())) emit('resp_deflate', one(mkresp(_dfbytes, {'content-encoding': 'deflate'}).read().decode())) # THE DEFLATE DECODER ACCEPTS BOTH SHAPES. `Content-Encoding: deflate' is # specified as a zlib stream, but servers ship raw DEFLATE often enough that # urllib3 retries headerless on the first failure -- so the wrong shape decodes # too, and a caller cannot use urllib3 to tell which one it was sent. _rawobj = zlib.compressobj(9, zlib.DEFLATED, -zlib.MAX_WBITS) _rawbytes = _rawobj.compress(b'raw deflate') + _rawobj.flush() emit('resp_deflate_raw', one(mkresp(_rawbytes, {'content-encoding': 'deflate'}).read().decode())) emit('resp_identity', one(mkresp(b'raw', {'content-encoding': 'identity'}).read().decode())) # AN UNKNOWN CONTENT-ENCODING IS PASSED THROUGH UNDECODED rather than raising -- # including `br' and `zstd' on this image, which is exactly the degradation the # absent packages produce. A caller that asked for them gets bytes it cannot use # and no exception to tell it so. # `.hex()' rather than `repr()': the bytes are non-ASCII by construction (they # are the magic numbers of the two formats) and hex is the one rendering that is # both ASCII and free of the quotes and backslashes a shell-quoted expected value # would have to carry. The assertion is that the bytes come back UNCHANGED. emit('resp_br_passthrough', mkresp(b'\x0b\x02\x80abc\x03', {'content-encoding': 'br'}).read().hex()) emit('resp_zstd_passthrough', mkresp(b'\x28\xb5\x2f\xfd', {'content-encoding': 'zstd'}).read().hex()) emit('resp_decode_content_false', one(len(mkresp(_gzbytes, {'content-encoding': 'gzip'}).read(decode_content=False)))) emit('resp_corrupt_gzip', msg(lambda: mkresp(b'not-gzip', {'content-encoding': 'gzip'}).read()).split(':')[0]) _r = mkresp(b'0123456789') emit('resp_read_amt', one(_r.read(4).decode())) emit('resp_read_rest', one(_r.read().decode())) emit('resp_tell', one(_r.tell())) emit('resp_headers_type', one(type(mkresp(b'', {'a': '1'}).headers).__name__)) emit('resp_getheader', one(mkresp(b'', {'A': '1'}).headers.get('a'))) emit('resp_stream_chunks', ','.join(c.decode() for c in mkresp(b'abcdef').stream(2, decode_content=False))) emit('resp_readinto', attempt(lambda: mkresp(b'abc').readinto(bytearray(3)))) emit('resp_readable', one(mkresp(b'').readable())) emit('resp_isclosed_fresh', one(mkresp(b'').isclosed())) emit('resp_data_property', one(mkresp(b'xyz').data.decode())) emit('resp_json', attempt(lambda: mkresp(b'{"k": 1}', {'content-type': 'application/json'}).json())) emit('resp_json_bad', msg(lambda: mkresp(b'nope').json()).split(':')[0]) emit('resp_url_none', one(mkresp(b'').url)) emit('resp_retries_none', one(mkresp(b'').retries)) emit('resp_version_default', one(mkresp(b'').version)) emit('resp_reason_default', one(mkresp(b'').reason)) emit('resp_get_redirect_location_none', one(mkresp(b'').get_redirect_location())) emit('resp_get_redirect_location_302', one(mkresp(b'', {'location': '/x'}, status=302).get_redirect_location())) emit('resp_get_redirect_location_200_with_header', one(mkresp(b'', {'location': '/x'}, status=200).get_redirect_location())) # ============================================= ssl_ helpers ================== # Pure functions only. Anything that builds a real SSLContext depends on the # OpenSSL this image was linked against and would be asserting the platform # rather than the package. # `import urllib3' REACHES `ssl' unconditionally (util/ssltransport.py and # util/ssl_.py both import it at module level), so a python3 whose _ssl extension # did not load would fail the bare import above rather than reaching this block. # 99-python3 already asserts OpenSSL 3.x on this image; these cases assert that # urllib3's own view of it agrees. emit('ssl_module_importable', _import('ssl')) emit('ssl_is_pyopenssl', one(u3ssl.IS_PYOPENSSL)) emit('ssl_has_sni', one(u3ssl.HAS_NEVER_CHECK_COMMON_NAME)) _ctx = None try: _ctx = u3ssl.create_urllib3_context() except Exception as _e: emit('ssl_context', '!%s' % type(_e).__name__) else: emit('ssl_context', 'ok') emit('ssl_context_min_version', one(_ctx.minimum_version.name)) emit('ssl_context_check_hostname', one(_ctx.check_hostname)) emit('ssl_context_verify_mode', one(_ctx.verify_mode.name)) emit('ssl_context_no_compression', one(bool(_ctx.options & __import__('ssl').OP_NO_COMPRESSION))) emit('ssl_context_cert_none_no_hostname', one(u3ssl.create_urllib3_context(cert_reqs=0).check_hostname)) emit('ssl_resolve_cert_reqs_none', one(u3ssl.resolve_cert_reqs('CERT_NONE').name)) emit('ssl_resolve_cert_reqs_required', one(u3ssl.resolve_cert_reqs('CERT_REQUIRED').name)) emit('ssl_resolve_cert_reqs_default', one(u3ssl.resolve_cert_reqs(None).name)) emit('ssl_resolve_cert_reqs_bare', one(u3ssl.resolve_cert_reqs('REQUIRED').name)) # The digests below are md5/sha1/sha256 of the three bytes `abc', which is the # most-tabulated test vector there is: 900150983cd24fb0d6963f7d28e17f72, # a9993e364706816aba3e25717850c26c9cd0d89d and ba7816bf...15ad. `assert_fingerprint' # picks the hash by HEX LENGTH, so the three also assert that dispatch. emit('ssl_assert_fingerprint_md5', attempt(u3ssl.assert_fingerprint, b'abc', '90:01:50:98:3c:d2:4f:b0:d6:96:3f:7d:28:e1:7f:72')) emit('ssl_assert_fingerprint_sha1', attempt(u3ssl.assert_fingerprint, b'abc', 'a9993e364706816aba3e25717850c26c9cd0d89d')) emit('ssl_assert_fingerprint_bad', msg(u3ssl.assert_fingerprint, b'abc', '00:11').split(':')[0]) emit('ssl_assert_fingerprint_sha256', attempt(u3ssl.assert_fingerprint, b'abc', 'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad')) # ============================================= poolmanager (no I/O) ========== # Constructing a pool does NOT connect: a `PoolManager' is a dict of pools keyed # by a normalised context, and every assertion here is about that key. _pm = urllib3.PoolManager() emit('pm_pool_type', one(type(_pm.connection_from_url('http://a.com/')).__name__)) emit('pm_pool_type_https', one(type(_pm.connection_from_url('https://a.com/')).__name__)) emit('pm_pool_count_after_two', one((lambda p: (p.connection_from_url('http://a.com/'), p.connection_from_url('http://a.com/x'), len(p.pools)))(urllib3.PoolManager())[2])) emit('pm_pool_count_distinct_hosts', one((lambda p: (p.connection_from_url('http://a.com/'), p.connection_from_url('http://b.com/'), len(p.pools)))(urllib3.PoolManager())[2])) emit('pm_pool_host', one(_pm.connection_from_url('http://a.com:81/').host)) emit('pm_pool_port', one(_pm.connection_from_url('http://a.com:81/').port)) emit('pm_default_port_http', one(_pm.connection_from_url('http://a.com/').port)) emit('pm_default_port_https', one(_pm.connection_from_url('https://a.com/').port)) emit('pm_scheme', one(_pm.connection_from_url('http://a.com/').scheme)) emit('pm_bad_scheme', msg(_pm.connection_from_url, 'ftp://a.com/').split(':')[0]) emit('pm_proxy_from_url_type', one(type(urllib3.proxy_from_url('http://p:3128/')).__name__)) emit('pm_proxy_host', one(urllib3.proxy_from_url('http://p:3128/').proxy.host)) emit('pm_key_fields_count', one(len(urllib3.poolmanager.PoolKey._fields))) emit('pm_default_pools_size', one(urllib3.PoolManager().pools._maxsize)) emit('pool_default_maxsize', one(urllib3.HTTPConnectionPool('a.com').pool.maxsize)) # NOT `repr()': `HTTPConnectionPool' has no `__repr__' of its own, so the default # one carries a memory address and could never be asserted. emit('pool_class_name', one(type(urllib3.HTTPConnectionPool('a.com')).__name__)) emit('pool_str', one(str(urllib3.HTTPConnectionPool('a.com', 81)))) emit('pool_default_headers_empty', one(len(urllib3.HTTPConnectionPool('a.com').headers))) emit('pool_block_default', one(urllib3.HTTPConnectionPool('a.com').block)) # ============================================= module-level request ========== # `urllib3.request' is backed by a lazily-created module-level PoolManager; the # assertion is that reaching it costs no connection, not that a request works. emit('top_level_poolmanager_type', one(type(urllib3._DEFAULT_POOL).__name__ if hasattr(urllib3, '_DEFAULT_POOL') else 'absent')) emit('disable_warnings_returns_none', one(urllib3.disable_warnings())) # ============================================================== the end ====== emit('emitted_count', len(emitted)) emit('probe_done', 'yes')