Merge pull request #7374 from chrahunt/maint/use-server-fixture-not-internet

Use server fixture for network test

Author: chrahunt

tests/conftest.py

@@ -6,18 +6,27 @@
 import shutil
 import subprocess
 import sys
+from contextlib import contextmanager
 
 import pytest
 import six
+from pip._vendor.contextlib2 import ExitStack
 from setuptools.wheel import Wheel
 
 from pip._internal.main import main as pip_entry_point
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 from tests.lib import DATA_DIR, SRC_DIR, TestData
 from tests.lib.certs import make_tls_cert, serialize_cert, serialize_key
 from tests.lib.path import Path
 from tests.lib.scripttest import PipTestEnvironment
+from tests.lib.server import make_mock_server, server_running
 from tests.lib.venv import VirtualEnvironment
 
+if MYPY_CHECK_RUNNING:
+    from typing import Dict, Iterable
+
+    from tests.lib.server import MockServer as _MockServer, Responder
+
 
 def pytest_addoption(parser):
     parser.addoption(
@@ -404,3 +413,60 @@ def factory():
         return str(output_path)
 
     return factory
+
+
+class MockServer(object):
+    def __init__(self, server):
+        # type: (_MockServer) -> None
+        self._server = server
+        self._running = False
+        self.context = ExitStack()
+
+    @property
+    def port(self):
+        return self._server.port
+
+    @property
+    def host(self):
+        return self._server.host
+
+    def set_responses(self, responses):
+        # type: (Iterable[Responder]) -> None
+        assert not self._running, "responses cannot be set on running server"
+        self._server.mock.side_effect = responses
+
+    def start(self):
+        # type: () -> None
+        assert not self._running, "running server cannot be started"
+        self.context.enter_context(server_running(self._server))
+        self.context.enter_context(self._set_running())
+
+    @contextmanager
+    def _set_running(self):
+        self._running = True
+        try:
+            yield
+        finally:
+            self._running = False
+
+    def stop(self):
+        # type: () -> None
+        assert self._running, "idle server cannot be stopped"
+        self.context.close()
+
+    def get_requests(self):
+        # type: () -> Dict[str, str]
+        """Get environ for each received request.
+        """
+        assert not self._running, "cannot get mock from running server"
+        return [
+            call.args[0] for call in self._server.mock.call_args_list
+        ]
+
+
+@pytest.fixture
+def mock_server():
+    server = make_mock_server()
+    test_server = MockServer(server)
+    with test_server.context:
+        yield test_server

tests/functional/test_install_config.py

@@ -4,6 +4,8 @@
 
 import pytest
 
+from tests.lib.server import file_response, package_page
+
 
 def test_options_from_env_vars(script):
     """
@@ -44,26 +46,15 @@ def test_command_line_options_override_env_vars(script, virtualenv):
 def test_env_vars_override_config_file(script, virtualenv):
     """
     Test that environmental variables override settings in config files.
-
     """
-    fd, config_file = tempfile.mkstemp('-pip.cfg', 'test-')
-    try:
-        _test_env_vars_override_config_file(script, virtualenv, config_file)
-    finally:
-        # `os.close` is a workaround for a bug in subprocess
-        # https://bugs.python.org/issue3210
-        os.close(fd)
-        os.remove(config_file)
-
-
-def _test_env_vars_override_config_file(script, virtualenv, config_file):
+    config_file = script.scratch_path / "test-pip.cfg"
     # set this to make pip load it
-    script.environ['PIP_CONFIG_FILE'] = config_file
+    script.environ['PIP_CONFIG_FILE'] = str(config_file)
     # It's important that we test this particular config value ('no-index')
     # because there is/was a bug which only shows up in cases in which
     # 'config-item' and 'config_item' hash to the same value modulo the size
     # of the config dictionary.
-    (script.scratch_path / config_file).write_text(textwrap.dedent("""\
+    config_file.write_text(textwrap.dedent("""\
         [global]
         no-index = 1
         """))
@@ -133,56 +124,55 @@ def test_command_line_appends_correctly(script, data):
     ), 'stdout: {}'.format(result.stdout)
 
 
-@pytest.mark.network
-def test_config_file_override_stack(script, virtualenv):
+def test_config_file_override_stack(
+    script, virtualenv, mock_server, shared_data
+):
     """
     Test config files (global, overriding a global config with a
     local, overriding all with a command line flag).
-
     """
-    fd, config_file = tempfile.mkstemp('-pip.cfg', 'test-')
-    try:
-        _test_config_file_override_stack(script, virtualenv, config_file)
-    finally:
-        # `os.close` is a workaround for a bug in subprocess
-        # https://bugs.python.org/issue3210
-        os.close(fd)
-        os.remove(config_file)
+    mock_server.set_responses([
+        package_page({}),
+        package_page({}),
+        package_page({"INITools-0.2.tar.gz": "/files/INITools-0.2.tar.gz"}),
+        file_response(shared_data.packages.joinpath("INITools-0.2.tar.gz")),
+    ])
+    mock_server.start()
+    base_address = "http://{}:{}".format(mock_server.host, mock_server.port)
 
+    config_file = script.scratch_path / "test-pip.cfg"
 
-def _test_config_file_override_stack(script, virtualenv, config_file):
     # set this to make pip load it
-    script.environ['PIP_CONFIG_FILE'] = config_file
-    (script.scratch_path / config_file).write_text(textwrap.dedent("""\
+    script.environ['PIP_CONFIG_FILE'] = str(config_file)
+
+    config_file.write_text(textwrap.dedent("""\
         [global]
-        index-url = https://download.zope.org/ppix
-        """))
-    result = script.pip('install', '-vvv', 'INITools', expect_error=True)
-    assert (
-        "Getting page https://download.zope.org/ppix/initools" in result.stdout
-    )
+        index-url = {}/simple1
+        """.format(base_address)))
+    script.pip('install', '-vvv', 'INITools', expect_error=True)
     virtualenv.clear()
-    (script.scratch_path / config_file).write_text(textwrap.dedent("""\
+
+    config_file.write_text(textwrap.dedent("""\
         [global]
-        index-url = https://download.zope.org/ppix
+        index-url = {address}/simple1
         [install]
-        index-url = https://pypi.gocept.com/
-        """))
-    result = script.pip('install', '-vvv', 'INITools', expect_error=True)
-    assert "Getting page https://pypi.gocept.com/initools" in result.stdout
-    result = script.pip(
-        'install', '-vvv', '--index-url', 'https://pypi.org/simple/',
-        'INITools',
-    )
-    assert (
-        "Getting page http://download.zope.org/ppix/INITools"
-        not in result.stdout
+        index-url = {address}/simple2
+        """.format(address=base_address))
     )
-    assert "Getting page https://pypi.gocept.com/INITools" not in result.stdout
-    assert (
-        "Getting page https://pypi.org/simple/initools" in result.stdout
+    script.pip('install', '-vvv', 'INITools', expect_error=True)
+    script.pip(
+        'install', '-vvv', '--index-url', "{}/simple3".format(base_address),
+        'INITools',
     )
 
+    mock_server.stop()
+    requests = mock_server.get_requests()
+    assert len(requests) == 4
+    assert requests[0]["PATH_INFO"] == "/simple1/initools/"
+    assert requests[1]["PATH_INFO"] == "/simple2/initools/"
+    assert requests[2]["PATH_INFO"] == "/simple3/initools/"
+    assert requests[3]["PATH_INFO"] == "/files/INITools-0.2.tar.gz"
+
 
 def test_options_from_venv_config(script, virtualenv):
     """

tests/lib/server.py

@@ -51,9 +51,9 @@ def blocked_signals():
             signal.pthread_sigmask(signal.SIG_SETMASK, old_mask)
 
 
-class RequestHandler(WSGIRequestHandler):
+class _RequestHandler(WSGIRequestHandler):
     def make_environ(self):
-        environ = super(RequestHandler, self).make_environ()
+        environ = super(_RequestHandler, self).make_environ()
 
         # From pallets/werkzeug#1469, will probably be in release after
         # 0.16.0.
@@ -76,7 +76,7 @@ def make_environ(self):
         return environ
 
 
-def mock_wsgi_adapter(mock):
+def _mock_wsgi_adapter(mock):
     # type: (Callable[[Environ, StartResponse], Responder]) -> Responder
     """Uses a mock to record function arguments and provide
     the actual function that should respond.
@@ -91,10 +91,39 @@ def adapter(environ, start_response):
 
 def make_mock_server(**kwargs):
     # type: (Any) -> MockServer
-    kwargs.setdefault("request_handler", RequestHandler)
+    """Creates a mock HTTP(S) server listening on a random port on localhost.
+
+    The `mock` property of the returned server provides and records all WSGI
+    interactions, so one approach to testing could be
+
+        server = make_mock_server()
+        server.mock.side_effects = [
+            page1,
+            page2,
+        ]
+
+        with server_running(server):
+            # ... use server...
+            ...
+
+        assert server.mock.call_count > 0
+        call_args_list = server.mock.call_args_list
+
+        # `environ` is a dictionary defined as per PEP 3333 with the associated
+        # contents. Additional properties may be added by werkzeug.
+        environ, _ = call_args_list[0].args
+        assert environ["PATH_INFO"].startswith("/hello/simple")
+
+    Note that the server interactions take place in a different thread, so you
+    do not want to touch the server.mock within the `server_running` block.
+
+    Note also for pip interactions that "localhost" is a "secure origin", so
+    be careful using this for failure tests of `--trusted-host`.
+    """
+    kwargs.setdefault("request_handler", _RequestHandler)
 
     mock = Mock()
-    app = mock_wsgi_adapter(mock)
+    app = _mock_wsgi_adapter(mock)
     server = _make_server("localhost", 0, app=app, **kwargs)
     server.mock = mock
     return server
@@ -103,6 +132,8 @@ def make_mock_server(**kwargs):
 @contextmanager
 def server_running(server):
     # type: (BaseWSGIServer) -> None
+    """Context manager for running the provided server in a separate thread.
+    """
     thread = threading.Thread(target=server.serve_forever)
     thread.daemon = True
     with blocked_signals():