Httpx intercept

.github/workflows/tests.yaml

@@ -11,8 +11,6 @@ jobs:
                 include:
                 - python: 3.x
                   toxenv: pep8
-                - python: 3.8
-                  toxenv: py38
                 - python: 3.9
                   toxenv: py39
                 - python: "3.10"
@@ -23,10 +21,8 @@ jobs:
                   toxenv: py312
                 - python: "3.13"
                   toxenv: py313
-                - python: pypy-3.8
+                - python: pypy-3.10
                   toxenv: pypy3
-                - python: 3.8
-                  toxenv: py38-pytest
                 - python: 3.9
                   toxenv: py39-pytest
                 - python: "3.10"
@@ -37,12 +33,12 @@ jobs:
                   toxenv: py312-pytest
                 - python: "3.13"
                   toxenv: py313-pytest
-                - python: 3.9
-                  toxenv: py39-failskip
-                - python: 3.9
-                  toxenv: py39-limit
-                - python: 3.9
-                  toxenv: py39-prefix
+                - python: "3.13"
+                  toxenv: py313-failskip
+                - python: "3.13"
+                  toxenv: py313-limit
+                - python: "3.13"
+                  toxenv: py313-prefix
         name: ${{ matrix.toxenv }} on Python ${{ matrix.python }}
         steps:
         - uses: actions/checkout@v2

README.rst

@@ -20,7 +20,7 @@ looks like this::
 See the docs_ for more details on the many features and formats for
 setting request headers and bodies and evaluating responses.
 
-Gabbi is tested with Python 3.8, 3.9, 3.10, 3.11, 3.12, 3.13 and pypy3.
+Gabbi is tested with Python 3.9, 3.10, 3.11, 3.12, 3.13 and pypy3.10.
 
 Tests can be run using `unittest`_ style test runners, `pytest`_
 or from the command line with a `gabbi-run`_ script.

docs/source/host.rst

@@ -6,9 +6,8 @@ Gabbi intends to preserve the flow and semantics of HTTP interactions
 as much as possible, and every HTTP request needs to be directed at a host
 of some form. Gabbi provides three ways to control this:
 
-* Using `wsgi-intercept`_ to provide a fake socket and ``WSGI``
-  environment on an arbitrary host and port attached to a ``WSGI``
-  application (see `intercept examples`_).
+* Using `WSGITransport` of httpx to provide a ``WSGI`` environment on
+  directly attached to a ``WSGI`` application (see `intercept examples`_).
 * Using fully qualified ``url`` values in the YAML defined tests (see
   `full examples`_).
 * Using a host and (optionally) port defined at test build time (see
@@ -18,15 +17,15 @@ The intercept and live methods are mutually exclusive per test builder,
 but either kind of test can freely intermix fully qualified URLs into the
 sequence of tests in a YAML file.
 
-For test driven development and local tests the intercept style of
-testing lowers test requirements (no web server required) and is fast.
-Interception is performed as part of :doc:`fixtures` processing as the most
-deeply nested fixture. This allows any configuration or database
-setup to be performed prior to the WSGI application being created.
+For Python-based test driven development and local tests the intercept
+style of testing lowers test requirements (no web server required) and
+is fast. Interception is performed as part of the per-test-case http
+client. Configuration or database setup may be performed using
+:doc:`fixtures`.
 
 For the implementation of the above see :meth:`~gabbi.driver.build_tests`.
 
-.. _wsgi-intercept: https://pypi.python.org/pypi/wsgi_intercept
+.. _WSGITransport: https://www.python-httpx.org/advanced/transports/#wsgi-transport
 .. _intercept examples: https://github.com/cdent/gabbi/blob/main/gabbi/tests/test_intercept.py
 .. _full examples: https://github.com/cdent/gabbi/blob/main/gabbi/tests/gabbits_live/google.yaml
 .. _live examples: https://github.com/cdent/gabbi/blob/main/gabbi/tests/test_live.py

gabbi/case.py

@@ -14,7 +14,7 @@
 
 The test case encapsulates the request headers and body and expected
 response headers and body. When the test is run an HTTP request is
-made using urllib3. Assertions are made against the response.
+made using httpx. Assertions are made against the response.
 """
 
 from collections import OrderedDict
@@ -29,8 +29,6 @@
 from unittest import result as unitresult
 import urllib.parse as urlparse
 
-import wsgi_intercept
-
 from gabbi import __version__
 from gabbi import exception
 from gabbi.handlers import base
@@ -492,29 +490,19 @@ def _run_request(
         redirect=False,
         timeout=30,
     ):
-        """Run the http request and decode output.
-
-        The call to make the request will catch a WSGIAppError from
-        wsgi_intercept so that the real traceback from a catastrophic
-        error in the intercepted app can be examined.
-        """
+        """Run the http request and decode output."""
 
         if 'user-agent' not in (key.lower() for key in headers):
-            headers['user-agent'] = "gabbi/%s (Python urllib3)" % __version__
-
-        try:
-            response, content = self.http.request(
-                url,
-                method=method,
-                headers=headers,
-                body=body,
-                redirect=redirect,
-                timeout=timeout,
-            )
-        except wsgi_intercept.WSGIAppError as exc:
-            # Extract and re-raise the wrapped exception.
-            raise exc.exception_type(exc.exception_value).with_traceback(
-                   exc.traceback) from None
+            headers['user-agent'] = "gabbi/%s (Python httpx)" % __version__
+
+        response, content = self.http.request(
+            url,
+            method=method,
+            headers=headers,
+            body=body,
+            redirect=redirect,
+            timeout=timeout,
+        )
 
         # Set headers and location attributes for follow on requests
         self.response = response

gabbi/driver.py

@@ -52,7 +52,7 @@ def build_tests(path, loader, host=None, port=8001, intercept=None,
     :param loader: The TestLoader.
     :param host: The host to test against. Do not use with ``intercept``.
     :param port: The port to test against. Used with ``host``.
-    :param intercept: WSGI app factory for wsgi-intercept.
+    :param intercept: WSGI app factory for httpclient WSGITransport.
     :param test_loader_name: Base name for test classes. Use this to align the
                              naming of the tests with other tests in a system.
     :param fixture_module: Python module containing fixture classes.
@@ -158,7 +158,7 @@ def build_tests(path, loader, host=None, port=8001, intercept=None,
 
 
 def py_test_generator(test_dir, host=None, port=8001, intercept=None,
-                      prefix=None, test_loader_name=None,
+                      prefix='', test_loader_name=None,
                       fixture_module=None, response_handlers=None,
                       content_handlers=None, require_ssl=False, url=None,
                       metafunc=None, use_prior_test=True,

gabbi/httpclient.py

@@ -11,53 +11,60 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import logging
 import os
 import sys
 
-import certifi
-import urllib3
+import httpx
 
 from gabbi.handlers import jsonhandler
 from gabbi import utils
 
-# Disable SSL warnings otherwise tests which process stderr will get
-# extra information.
-urllib3.disable_warnings()
+logging.getLogger('httpx').setLevel(logging.WARNING)
 
 
-class Http(urllib3.PoolManager):
-    """A subclass of the ``urllib3.PoolManager`` to munge the data.
+class Http:
+    """A class to munge the HTTP response.
 
     This transforms the response to look more like what httplib2
     provided when it was used as the HTTP client.
     """
 
+    def __init__(self, **kwargs):
+        self.extensions = {}
+        if 'server_hostname' in kwargs:
+            self.extensions['sni_hostname'] = kwargs['server_hostname']
+        transport = kwargs.get('intercept')
+        if transport:
+            transport = httpx.WSGITransport(
+                app=transport(), script_name=kwargs.get("prefix", "")
+            )
+        self.client = httpx.Client(
+            transport=transport, verify=kwargs.get("cert_validate", True)
+        )
+
     def request(self, absolute_uri, method, body, headers, redirect, timeout):
-        if redirect:
-            retry = urllib3.util.Retry(raise_on_redirect=False, redirect=5)
-        else:
-            retry = urllib3.util.Retry(total=False, redirect=False)
-        response = super(Http, self).request(
-            method,
-            absolute_uri,
-            body=body,
+        response = self.client.request(
+            method=method,
+            url=absolute_uri,
             headers=headers,
-            retries=retry,
+            content=body,
             timeout=timeout,
+            follow_redirects=redirect,
+            extensions=self.extensions,
         )
 
         # Transform response into something akin to httplib2
         # response object.
-        content = response.data
-        status = response.status
-        reason = response.reason
+        content = response.content
+        status = response.status_code
+        reason = response.reason_phrase
+        http_version = response.http_version
         headers = response.headers
         headers['status'] = str(status)
-        headers['reason'] = reason
+        headers['reason'] = str(reason)
+        headers['http_protocol_version'] = str(http_version)
 
-        # Shut down open PoolManagers whose connections have completed to
-        # save on socket file descriptors.
-        self.clear()
         return headers, content
 
 
@@ -87,6 +94,7 @@ class VerboseHttp(Http):
     HEADER_BLACKLIST = [
         'status',
         'reason',
+        'http_protocol_version',
     ]
 
     REQUEST_PREFIX = '>'
@@ -106,27 +114,26 @@ def __init__(self, **kwargs):
         self._stream = kwargs.pop('stream')
         if self._use_color:
             self.colorize = utils.get_colorizer(self._stream)
-        super(VerboseHttp, self).__init__(**kwargs)
+        super().__init__(**kwargs)
 
     def request(self, absolute_uri, method, body, headers, redirect, timeout):
         """Display request parameters before requesting."""
 
-        self._verbose_output('#### %s ####' % self.caption,
+        self._verbose_output(f'#### {self.caption} ####',
                              color=self.COLORMAP['caption'])
-        self._verbose_output('%s %s' % (method, absolute_uri),
+        self._verbose_output(f'{method} {absolute_uri}',
                              prefix=self.REQUEST_PREFIX,
                              color=self.COLORMAP['request'])
 
         self._print_headers(headers, prefix=self.REQUEST_PREFIX)
         self._print_body(headers, body)
 
-        response, content = super(VerboseHttp, self).request(
+        response, content = super().request(
             absolute_uri, method, body, headers, redirect, timeout)
 
         # Blank line for division
         self._verbose_output('')
-        self._verbose_output('%s %s' % (response['status'],
-                                        response['reason']),
+        self._verbose_output(f'{response["status"]} {response["reason"]}',
                              prefix=self.RESPONSE_PREFIX,
                              color=self.COLORMAP['status'])
         self._print_headers(response, prefix=self.RESPONSE_PREFIX)
@@ -172,8 +179,8 @@ def _print_body(self, headers, content):
 
     def _print_header(self, name, value, prefix='', stream=None):
         """Output one single header."""
-        header = self.colorize(self.COLORMAP['header'], "%s:" % name)
-        self._verbose_output("%s %s" % (header, value), prefix=prefix,
+        header = self.colorize(self.COLORMAP['header'], f'{name}:')
+        self._verbose_output(f'{header} {value}', prefix=prefix,
                              stream=stream)
 
     def _verbose_output(self, message, prefix='', color=None, stream=None):
@@ -191,32 +198,32 @@ def get_http(
     caption='',
     cert_validate=True,
     hostname=None,
+    intercept=None,
+    prefix='',
     timeout=30,
 ):
     """Return an ``Http`` class for making requests."""
-    cert_validation = {'cert_reqs': 'CERT_NONE'} if not cert_validate else {}
-
     if not verbose:
         return Http(
-            strict=True,
-            ca_certs=certifi.where(),
             server_hostname=hostname,
             timeout=timeout,
-            **cert_validation
+            cert_validate=cert_validate,
+            intercept=intercept,
+            prefix=prefix,
         )
 
-    headers = False if verbose == 'body' else True
-    body = False if verbose == 'headers' else True
+    headers = verbose != 'body'
+    body = verbose != 'headers'
 
     return VerboseHttp(
         headers=headers,
         body=body,
         stream=sys.stdout,
         caption=caption,
         colorize=True,
-        strict=True,
-        ca_certs=certifi.where(),
         server_hostname=hostname,
         timeout=timeout,
-        **cert_validation
+        cert_validate=cert_validate,
+        intercept=intercept,
+        prefix=prefix,
     )

gabbi/suite.py

@@ -19,8 +19,6 @@
 import sys
 import unittest
 
-from wsgi_intercept import interceptor
-
 from gabbi import fixture
 
 
@@ -46,16 +44,11 @@ def run(self, result, debug=False):
         are no fixtures.
         """
 
-        fixtures, intercept, host, port, prefix = self._get_intercept()
+        fixtures, host, port = self._get_fixtures()
 
         try:
             with fixture.nest([fix() for fix in fixtures]):
-                if intercept:
-                    with interceptor.Urllib3Interceptor(
-                            intercept, host, port, prefix):
-                        result = super(GabbiSuite, self).run(result, debug)
-                else:
-                    result = super(GabbiSuite, self).run(result, debug)
+                result = super(GabbiSuite, self).run(result, debug)
         except unittest.SkipTest as exc:
             for test in self._tests:
                 result.addSkip(test, str(exc))
@@ -85,7 +78,7 @@ def run(self, result, debug=False):
     def start(self, result, tests=None):
         """Start fixtures when using pytest."""
         tests = tests or []
-        fixtures, intercept, host, port, prefix = self._get_intercept()
+        fixtures, host, port = self._get_fixtures()
 
         self.used_fixtures = []
         try:
@@ -100,38 +93,25 @@ def start(self, result, tests=None):
                 test.run = noop
                 test.add_marker('skip')
             result.addSkip(self, str(exc))
-        if intercept:
-            intercept_fixture = interceptor.Urllib3Interceptor(
-                intercept, host, port, prefix)
-            intercept_fixture.__enter__()
-            self.used_fixtures.append(intercept_fixture)
 
     def stop(self):
         """Stop fixtures when using pytest."""
         for fix in reversed(self.used_fixtures):
             fix.__exit__(None, None, None)
 
-    def _get_intercept(self):
+    def _get_fixtures(self):
         fixtures = [fixture.GabbiFixture]
-        intercept = host = port = prefix = None
+        host = port = None
         try:
             first_test = self._find_first_full_test()
             fixtures = first_test.fixtures
             host = first_test.host
             port = first_test.port
-            prefix = first_test.prefix
-            intercept = first_test.intercept
-
-            # Unbind a passed in WSGI application. During the
-            # metaclass building process intercept becomes bound.
-            try:
-                intercept = intercept.__func__
-            except AttributeError:
-                pass
+
         except AttributeError:
             pass
 
-        return fixtures, intercept, host, port, prefix
+        return fixtures, host, port
 
     def _find_first_full_test(self):
         """Traverse a sparse test suite to find the first HTTPTestCase.

gabbi/suitemaker.py

@@ -90,6 +90,8 @@ def make_one_test(self, test_dict, prior_test):
                                          caption=test['name'],
                                          cert_validate=test['cert_validate'],
                                          hostname=hostname,
+                                         intercept=self.intercept,
+                                         prefix=self.prefix,
                                          timeout=int(test["timeout"]))
         if prior_test:
             history = prior_test.history

gabbi/tests/external_server.py

@@ -0,0 +1,32 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import sys
+from wsgiref import simple_server
+
+from gabbi.tests import simple_wsgi
+
+
+def run(host, port):
+    server = simple_server.make_server(
+        host,
+        int(port),
+        simple_wsgi.SimpleWsgi(),
+    )
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    host = sys.argv[1]
+    port = sys.argv[2]
+    run(host, port)

gabbi/tests/gabbits_intercept/host-header.yaml

@@ -1,4 +1,4 @@
-# Test, against wsgi-intercept, that SNI and host header handling behaves.
+# Test, against intercepted WSGI app, that SNI and host header handling behaves.
 
 tests:
 

gabbi/tests/simple_wsgi.py

@@ -30,17 +30,22 @@ def __call__(self, environ, start_response):
         global METHODS
         global CURRENT_POLL
 
+        script_name = environ.get('SCRIPT_NAME', '')
+        path_info = environ.get('PATH_INFO', '').removeprefix(script_name)
         request_method = environ['REQUEST_METHOD'].upper()
-        query_data = urlparse.parse_qs(environ.get('QUERY_STRING', ''))
-        request_url = environ.get('REQUEST_URI',
-                                  environ.get('RAW_URI', 'unknown'))
-        path_info = environ.get('PATH_INFO', '')
+        query_string = environ.get('QUERY_STRING', '')
+        query_data = urlparse.parse_qs(query_string)
+        request_url = script_name + path_info
         accept_header = environ.get('HTTP_ACCEPT')
         content_type_header = environ.get('CONTENT_TYPE', '')
 
-        full_request_url = self._fully_qualify(environ, request_url)
+        full_request_url = self._fully_qualify(
+            environ,
+            request_url,
+            query_string,
+        )
 
-        if accept_header:
+        if accept_header and accept_header != '*/*':
             response_content_type = accept_header
         else:
             # JSON doesn't need a charset but we throw one in here
@@ -64,7 +69,8 @@ def __call__(self, environ, start_response):
             return []
 
         if request_method.startswith('P'):
-            body = environ['wsgi.input'].read()
+            length = int(environ.get('CONTENT_LENGTH', '0'))
+            body = environ['wsgi.input'].read(length)
             if body:
                 if not content_type_header:
                     start_response('400 Bad request', headers)
@@ -147,7 +153,7 @@ def __call__(self, environ, start_response):
         return [query_output.encode('utf-8')]
 
     @staticmethod
-    def _fully_qualify(environ, url):
+    def _fully_qualify(environ, url, query_data):
         """Turn a URL path into a fully qualified URL."""
         split_url = urlparse.urlsplit(url)
         server_name = environ.get('SERVER_NAME')
@@ -159,4 +165,4 @@ def _fully_qualify(environ, url):
             netloc = server_name
 
         return urlparse.urlunsplit((server_scheme, netloc, split_url.path,
-                                    split_url.query, split_url.fragment))
+                                    query_data, split_url.fragment))

gabbi/tests/test_runner.py

@@ -14,31 +14,61 @@
 """
 
 from io import StringIO
+import os
+import socket
+import subprocess
 import sys
+import time
 import unittest
-from uuid import uuid4
-
-from wsgi_intercept.interceptor import Urllib3Interceptor
 
 from gabbi import exception
 from gabbi.handlers import base
 from gabbi.handlers.jsonhandler import JSONHandler
 from gabbi import runner
-from gabbi.tests.simple_wsgi import SimpleWsgi
+
+
+def get_free_port():
+    sock = socket.socket()
+    sock.bind(('', 0))
+    return sock.getsockname()[1]
+
+
+class ForkedWSGIServer:
+
+    def __init__(self, host, port):
+        self.host = host
+        self.port = port
+
+    def start(self):
+        self.process = subprocess.Popen(
+            [
+                "python",
+                "gabbi/tests/external_server.py",
+                self.host,
+                str(self.port)
+            ],
+            env=os.environ.update({"PYTHONPATH": "."}),
+            close_fds=True,
+        )
+        # We need to sleep a bit to let the wsgi server start.
+        # TODO(cdent): This is regrettable.
+        time.sleep(.4)
+
+    def stop(self):
+        self.process.terminate()
 
 
 class RunnerTest(unittest.TestCase):
 
+    port = get_free_port()
+
     def setUp(self):
         super(RunnerTest, self).setUp()
 
-        # NB: random host ensures that we're not accidentally connecting to an
-        #     actual server
-        host, port = (str(uuid4()), 8000)
-        self.host = host
-        self.port = port
-        self.server = lambda: Urllib3Interceptor(
-            SimpleWsgi, host=host, port=port)
+        self.host = "0.0.0.0"
+        self.resolved_host = socket.getfqdn()
+        self.server = ForkedWSGIServer(self.host, self.port)
+        self.server.start()
 
         self._stdin = sys.stdin
 
@@ -49,41 +79,39 @@ def setUp(self):
         sys.stderr = StringIO()  # swallow output to avoid confusion
 
         self._argv = sys.argv
-        sys.argv = ['gabbi-run', '%s:%s' % (host, port)]
+        sys.argv = ['gabbi-run', '%s:%s' % (self.host, self.port)]
 
     def tearDown(self):
         sys.stdin = self._stdin
         sys.stdout = self._stdout
         sys.stderr = self._stderr
         sys.argv = self._argv
+        self.server.stop()
 
     def test_input_files(self):
         sys.argv = ['gabbi-run', 'http://%s:%s/foo' % (self.host, self.port)]
 
         sys.argv.append('--')
         sys.argv.append('gabbi/tests/gabbits_runner/success.yaml')
 
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
         sys.argv.append('gabbi/tests/gabbits_runner/failure.yaml')
 
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertFailure(err)
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertFailure(err)
 
         sys.argv.append('gabbi/tests/gabbits_runner/success_alt.yaml')
 
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertFailure(err)
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertFailure(err)
 
     def test_unsafe_yaml(self):
         sys.argv = ['gabbi-run', 'http://%s:%s/nan' % (self.host, self.port)]
@@ -92,11 +120,10 @@ def test_unsafe_yaml(self):
         sys.argv.append('--')
         sys.argv.append('gabbi/tests/gabbits_runner/nan.yaml')
 
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
     def test_target_url_parsing(self):
         sys.argv = ['gabbi-run', 'http://%s:%s/foo' % (self.host, self.port)]
@@ -108,36 +135,11 @@ def test_target_url_parsing(self):
           status: 200
           response_headers:
             x-gabbi-url: http://%s:%s/foo/baz
-        """ % (self.host, self.port))
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
-
-    def test_target_url_parsing_standard_port(self):
-        # NOTE(cdent): For reasons unclear this regularly fails in
-        # py.test and sometimes fails with testr. So there is
-        # some state that is not being properly cleard somewhere.
-        # Within SimpleWsgi, the environ thinks url_scheme is
-        # 'https'.
-        self.server = lambda: Urllib3Interceptor(
-            SimpleWsgi, host=self.host, port=80)
-        sys.argv = ['gabbi-run', 'http://%s/foo' % self.host]
-
-        sys.stdin = StringIO("""
-        tests:
-        - name: expected success
-          GET: /baz
-          status: 200
-          response_headers:
-            x-gabbi-url: http://%s/foo/baz
-        """ % self.host)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+        """ % (self.resolved_host, self.port))
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
     def test_custom_response_handler(self):
         sys.stdin = StringIO("""
@@ -160,11 +162,11 @@ def test_custom_response_handler(self):
               h1: Hello World
               p: lorem ipsum dolor sit amet
         """)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
         sys.stdin = StringIO("""
         tests:
@@ -173,11 +175,11 @@ def test_custom_response_handler(self):
           response_html:
               h1: lipsum
         """)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertFailure(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertFailure(err)
 
         sys.argv.insert(3, "-r")
         sys.argv.insert(4, "gabbi.tests.test_intercept:StubResponseHandler")
@@ -191,11 +193,11 @@ def test_custom_response_handler(self):
           response_test:
           - COWAnother line
         """)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
         sys.argv.insert(5, "-r")
         sys.argv.insert(6, "gabbi.tests.custom_response_handler")
@@ -208,11 +210,11 @@ def test_custom_response_handler(self):
           - Hello World
           - lorem ipsum dolor sit amet
         """)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
     def test_exit_code(self):
         sys.stdin = StringIO()
@@ -225,6 +227,7 @@ def test_exit_code(self):
           GET: /
           status: 666
         """)
+
         try:
             runner.run()
         except SystemExit as err:
@@ -236,23 +239,23 @@ def test_exit_code(self):
           GET: /
           status: 200
         """)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
     def test_verbose_output_formatting(self):
         """Confirm that a verbose test handles output properly."""
         sys.argv = ['gabbi-run', 'http://%s:%s/foo' % (self.host, self.port)]
 
         sys.argv.append('--')
         sys.argv.append('gabbi/tests/gabbits_runner/test_verbose.yaml')
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
         sys.stdout.seek(0)
         output = sys.stdout.read()
@@ -270,11 +273,10 @@ def test_data_dir_good(self):
         sys.argv.append('--')
         sys.argv.append('gabbi/tests/gabbits_runner/test_data.yaml')
 
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
         # Compare the verbose output of tests with pretty printed
         # data.
@@ -298,21 +300,20 @@ def test_stdin_data_dir(self):
           response_json_paths:
             $.items.house: blue
         """)
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
     def _run_verbosity_arg(self):
         sys.argv.append('--')
         sys.argv.append('gabbi/tests/gabbits_runner/verbosity.yaml')
 
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
 
         sys.stdout.seek(0)
         output = sys.stdout.read()
@@ -364,12 +365,13 @@ def test_quiet_is_quiet(self):
           status: 200
           response_headers:
             x-gabbi-url: http://%s:%s/foo/baz
-        """ % (self.host, self.port))
-        with self.server():
-            try:
-                runner.run()
-            except SystemExit as err:
-                self.assertSuccess(err)
+        """ % (self.resolved_host, self.port))
+
+        try:
+            runner.run()
+        except SystemExit as err:
+            self.assertSuccess(err)
+
         sys.stdout.seek(0)
         sys.stderr.seek(0)
         stdoutput = sys.stdout.read()

requirements.txt

@@ -1,8 +1,7 @@
 pbr
 pytest
 PyYAML
-urllib3>=1.26.9,<2.0.0
+httpx
 certifi
 jsonpath-rw-ext>=1.0.0
-wsgi-intercept>=1.13.0
 colorama

setup.cfg

@@ -15,7 +15,6 @@ classifier =
     Operating System :: POSIX
     Programming Language :: Python
     Programming Language :: Python :: 3
-    Programming Language :: Python :: 3.8
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
     Programming Language :: Python :: 3.11

tox.ini

@@ -1,7 +1,7 @@
 [tox]
 minversion = 3.1.1
 skipsdist = True
-envlist = pep8,py38,py39,py310,py311,py312,py313,pypy3,pep8,limit,failskip,docs,py39-prefix,py39-limit,py39-verbosity,py39-failskip,py36-pytest,py38-pytest,py39-pytest,py310-pytest,py311-pytest,py312-pytest,py313-pytest
+envlist = pep8,py39,py310,py311,py312,py313,pypy3,pep8,limit,failskip,docs,py313-prefix,py313-limit,py313-verbosity,py313-failskip,py36-pytest,py39-pytest,py310-pytest,py311-pytest,py312-pytest,py313-pytest
 
 [testenv]
 deps = -r{toxinidir}/requirements.txt
@@ -22,9 +22,6 @@ commands = {posargs}
 [testenv:py36-pytest]
 commands = py.test gabbi
 
-[testenv:py38-pytest]
-commands = py.test gabbi
-
 [testenv:py39-pytest]
 commands = py.test gabbi
 
@@ -38,9 +35,9 @@ commands = py.test gabbi
 commands = py.test gabbi
 
 [testenv:py313-pytest]
-commands = py.test gabbi
+commands = py.test {posargs} gabbi
 
-[testenv:py39-prefix]
+[testenv:py313-prefix]
 setenv = GABBI_PREFIX=/snoopy
 
 [testenv:pep8]
@@ -49,15 +46,15 @@ deps = hacking
 commands =
     flake8
 
-[testenv:py39-limit]
+[testenv:py313-limit]
 allowlist_externals = {toxinidir}/test-limit.sh
 commands = {toxinidir}/test-limit.sh
 
-[testenv:py39-verbosity]
+[testenv:py313-verbosity]
 allowlist_externals = {toxinidir}/test-verbosity.sh
 commands = {toxinidir}/test-verbosity.sh
 
-[testenv:py39-failskip]
+[testenv:py313-failskip]
 allowlist_externals = {toxinidir}/test-failskip.sh
 commands = {toxinidir}/test-failskip.sh