Add caching proxy during tests

.github/workflows/unix.yml

@@ -221,7 +221,7 @@ jobs:
         run: |
           export TEST_MAMBA_EXE=$(pwd)/build/micromamba/micromamba
           unset CONDARC  # Interferes with tests
-          pytest -v --capture=tee-sys micromamba/tests/
+          pytest -v --capture=tee-sys micromamba/tests/ --mitmdump-exe="$(which mitmdump)"
       - name: micromamba local channel test
         shell: bash -l -eo pipefail {0}
         run: |

libmamba/src/api/configuration.cpp

@@ -1347,6 +1347,7 @@ namespace mamba
         insert(Configurable("proxy_servers", &m_context.remote_fetch_params.proxy_servers)
                    .group("Network")
                    .set_rc_configurable()
+                   .set_env_var_names()
                    .description("Use a proxy server for network connections")
                    .long_description(unindent(R"(
                         'proxy_servers' should be a dictionary where the key is either in the form of

micromamba/tests/caching_proxy.py

@@ -0,0 +1,46 @@
+import pathlib
+import pickle
+
+from mitmproxy import ctx, http
+from mitmproxy.tools.main import mitmdump
+
+
+class ResponseCapture:
+    def __init__(self):
+        self.cache_file = None
+        self.response_dict = {}
+
+    def load(self, loader):
+        loader.add_option(
+            name="cache_dir",
+            typespec=str,
+            default="",
+            help="Path to load/store request cache",
+        )
+
+    def configure(self, updates):
+        if "cache_dir" in updates:
+            self.cache_file = pathlib.Path(ctx.options.cache_dir) / "requests.pkl"
+            if self.cache_file.exists():
+                with open(self.cache_file, "rb") as f:
+                    self.response_dict = pickle.load(f)
+
+    def done(self):
+        if self.cache_file is not None:
+            self.cache_file.parent.mkdir(parents=True, exist_ok=True)
+            with open(self.cache_file, "wb") as f:
+                pickle.dump(self.response_dict, f)
+
+    def request(self, flow: http.HTTPFlow) -> None:
+        if (url := flow.request.url) in self.response_dict:
+            flow.response = self.response_dict[url]
+
+    def response(self, flow: http.HTTPFlow) -> None:
+        self.response_dict[flow.request.url] = flow.response
+
+
+addons = [ResponseCapture()]
+
+if __name__ == "__main__":
+    print("Starting mitmproxy...")
+    mitmdump(["-s", __file__])

micromamba/tests/conftest.py

@@ -2,19 +2,35 @@
 import os
 import pathlib
 import platform
+import shutil
+import subprocess
 from typing import Any, Generator, Mapping, Optional
 
 import pytest
 
 from . import helpers
 
+__this_dir__ = pathlib.Path(__file__).parent.resolve()
+
 ####################
 #  Config options  #
 ####################
 
 
 def pytest_addoption(parser):
     """Add command line argument to pytest."""
+    parser.addoption(
+        "--mitmdump-exe",
+        action="store",
+        default=None,
+        help="Path to mitmdump proxy executable",
+    )
+    parser.addoption(
+        "--caching-proxy-dir",
+        action="store",
+        default=None,
+        help="Path to a directory to store requests between calls",
+    )
     parser.addoption(
         "--mamba-pkgs-dir",
         action="store",
@@ -35,13 +51,79 @@ def pytest_addoption(parser):
     )
 
 
-##################
-#  Test fixture  #
-##################
+###########################
+#  Test utility fixtures  #
+###########################
+
+
+@pytest.fixture(scope="session")
+def mitmdump_exe(request) -> Optional[pathlib.Path]:
+    """Get the path to the ``mitmdump`` executable."""
+    if (p := request.config.getoption("--mitmdump-exe")) is not None:
+        return pathlib.Path(p).resolve()
+    elif (p := shutil.which("mitmdump")) is not None:
+        return pathlib.Path(p).resolve()
+    return None
+
+
+@pytest.fixture(scope="session")
+def session_cache_proxy(
+    request, mitmdump_exe, unused_tcp_port_factory, tmp_path_factory
+) -> Generator[Optional[helpers.MitmProxy], None, None]:
+    """Launch and a caching proxy to speed up tests, not used automatically."""
+    assert mitmdump_exe is not None
+
+    proxy = helpers.MitmProxy(
+        exe=mitmdump_exe,
+        scripts=str(__this_dir__ / "caching_proxy.py"),
+        confdir=tmp_path_factory.mktemp("mitmproxy"),
+    )
+
+    options = []
+    if (p := request.config.getoption("--caching-proxy-dir")) is not None:
+        options += ["--set", f"cache_dir={p}"]
+
+    proxy.start_proxy(unused_tcp_port_factory(), options)
+
+    yield proxy
+
+    proxy.stop_proxy()
+
+
+@pytest.fixture(scope="session", autouse=True)
+def session_clean_env(
+    mitmdump_exe,  # To help resolve the executable before we clean the env
+) -> None:
+    """Remove all Conda/Mamba activation artifacts from environment."""
+    old_environ = copy.deepcopy(os.environ)
+
+    for k, v in os.environ.items():
+        if k.startswith(("CONDA", "_CONDA", "MAMBA", "_MAMBA", "XDG_")):
+            del os.environ[k]
+
+    def keep_in_path(
+        p: str, prefix: Optional[str] = old_environ.get("CONDA_PREFIX")
+    ) -> bool:
+        if "condabin" in p:
+            return False
+        # On windows, PATH is also used for dyanamic libraries.
+        if (prefix is not None) and (platform.system() != "Windows"):
+            p = str(pathlib.Path(p).expanduser().resolve())
+            prefix = str(pathlib.Path(prefix).expanduser().resolve())
+            return not p.startswith(prefix)
+        return True
+
+    path_list = os.environ["PATH"].split(os.pathsep)
+    path_list = [p for p in path_list if keep_in_path(p)]
+    os.environ["PATH"] = os.pathsep.join(path_list)
+
+    yield
+
+    os.environ.update(old_environ)
 
 
 @pytest.fixture(autouse=True)
-def tmp_environ() -> Generator[Mapping[str, Any], None, None]:
+def tmp_environ(session_clean_env: None) -> Generator[Mapping[str, Any], None, None]:
     """Saves and restore environment variables.
 
     This is used for test that need to modify ``os.environ``
@@ -52,6 +134,29 @@ def tmp_environ() -> Generator[Mapping[str, Any], None, None]:
     os.environ.update(old_environ)
 
 
+@pytest.fixture(params=[True])
+def use_caching_proxy(request) -> bool:
+    """A dummy fixture to control the use of the caching proxy."""
+    return request.param
+
+
+@pytest.fixture(autouse=True)
+def tmp_use_proxy(
+    use_caching_proxy, tmp_environ, session_clean_env: None, session_cache_proxy
+):
+    if use_caching_proxy:
+        port = session_cache_proxy.port
+        assert port is not None
+        os.environ["MAMBA_PROXY_SERVERS"] = (
+            "{"
+            + f"http: http://localhost:{port}, https: https://localhost:{port}"
+            + "}"
+        )
+        os.environ["MAMBA_CACERT_PATH"] = str(
+            session_cache_proxy.confdir / "mitmproxy-ca-cert.pem"
+        )
+
+
 @pytest.fixture
 def tmp_home(
     request, tmp_environ, tmp_path_factory: pytest.TempPathFactory
@@ -83,31 +188,6 @@ def tmp_home(
             pass
 
 
-@pytest.fixture
-def tmp_clean_env(tmp_environ: None) -> None:
-    """Remove all Conda/Mamba activation artifacts from environment."""
-    for k, v in os.environ.items():
-        if k.startswith(("CONDA", "_CONDA", "MAMBA", "_MAMBA", "XDG_")):
-            del os.environ[k]
-
-    def keep_in_path(
-        p: str, prefix: Optional[str] = tmp_environ.get("CONDA_PREFIX")
-    ) -> bool:
-        if "condabin" in p:
-            return False
-        # On windows, PATH is also used for dyanamic libraries.
-        if (prefix is not None) and (platform.system() != "Windows"):
-            p = str(pathlib.Path(p).expanduser().resolve())
-            prefix = str(pathlib.Path(prefix).expanduser().resolve())
-            return not p.startswith(prefix)
-        return True
-
-    path_list = os.environ["PATH"].split(os.pathsep)
-    path_list = [p for p in path_list if keep_in_path(p)]
-    os.environ["PATH"] = os.pathsep.join(path_list)
-    # os.environ restored by tmp_clean_env and tmp_environ
-
-
 @pytest.fixture(scope="session")
 def tmp_pkgs_dirs(tmp_path_factory: pytest.TempPathFactory, request) -> pathlib.Path:
     """A common package cache for mamba downloads.
@@ -132,7 +212,7 @@ def shared_pkgs_dirs(request) -> bool:
 def tmp_root_prefix(
     request,
     tmp_path_factory: pytest.TempPathFactory,
-    tmp_clean_env: None,
+    tmp_environ: None,
     tmp_pkgs_dirs: pathlib.Path,
     shared_pkgs_dirs: bool,
 ) -> Generator[pathlib.Path, None, None]:
@@ -151,7 +231,7 @@ def tmp_root_prefix(
     if not request.config.getoption("--no-eager-clean"):
         if new_root_prefix.exists():
             helpers.rmtree(new_root_prefix)
-    # os.environ restored by tmp_clean_env and tmp_environ
+    # os.environ restored by tmp_environ
 
 
 @pytest.fixture(params=[helpers.random_string])

micromamba/tests/helpers.py

@@ -6,14 +6,57 @@
 import shutil
 import string
 import subprocess
-import sys
+import time
 from enum import Enum
 from pathlib import Path
+from typing import Optional
 
 import pytest
 import yaml
 
 
+class MitmProxy:
+    def __init__(
+        self,
+        exe: Path,
+        scripts: Optional[Path] = None,
+        confdir: Optional[Path] = None,
+        outfile: Optional[Path] = None,
+    ):
+        self.exe = Path(exe).resolve()
+        self.scripts = Path(scripts).resolve() if scripts is not None else None
+        self.confdir = Path(confdir).resolve() if confdir is not None else None
+        self.outfile = Path(outfile).resolve() if outfile is not None else None
+        self.process = None
+        self.port = None
+
+    def start_proxy(self, port, options=[]):
+        assert self.process is None
+        args = [self.exe, "--listen-port", str(port)]
+        if self.scripts is not None:
+            args += ["--scripts", str(self.scripts)]
+        if self.confdir is not None:
+            args += ["--set", f"confdir={self.confdir}"]
+        if self.outfile is not None:
+            args += ["--set", f"outfile={self.outfile}"]
+        args += options
+        self.process = subprocess.Popen(args)
+
+        # Wait until mitmproxy has generated its certificate or some tests might fail
+        while not (Path(self.confdir) / "mitmproxy-ca-cert.pem").exists():
+            time.sleep(1)
+        self.port = port
+
+    def stop_proxy(self):
+        self.process.terminate()
+        try:
+            self.process.wait(3)
+        except subprocess.TimeoutExpired:
+            self.process.kill()
+        self.process = None
+        self.port = None
+
+
 def subprocess_run(*args: str, **kwargs) -> str:
     """Execute a command in a subprocess while properly capturing stderr in exceptions."""
     try:

micromamba/tests/test_activation.py

@@ -635,7 +635,6 @@ def test_env_activation(tmp_home, winreg_value, tmp_root_prefix, tmp_path, inter
 @pytest.mark.parametrize("interpreter", get_interpreters())
 def test_activation_envvars(
     tmp_home,
-    tmp_clean_env,
     winreg_value,
     tmp_root_prefix,
     tmp_path,

micromamba/tests/test_create.py

@@ -340,14 +340,9 @@ def test_multiple_spec_files(tmp_home, tmp_root_prefix, tmp_path, type):
             assert res["specs"] == [explicit_specs[0]]
 
 
-def test_multiprocessing():
-    if platform.system() == "Windows":
-        return
-
-    root_prefix = Path(os.environ["MAMBA_ROOT_PREFIX"])
-    if os.path.exists(root_prefix / "pkgs"):
-        shutil.rmtree(root_prefix / "pkgs")
-
+@pytest.mark.skipif(platform.system() == "Windows", reason="Windows")
+@pytest.mark.parametrize("use_caching_proxy", [False], indirect=True)
+def test_multiprocessing(tmp_home, tmp_root_prefix):
     cmd = [helpers.get_umamba()]
     cmd += ["create", "-n", "env1", "-y"]
     cmd += ["airflow"]
@@ -641,7 +636,8 @@ def test_spec_with_channel(tmp_home, tmp_root_prefix, tmp_path):
             assert link["version"].startswith("0.22.")
 
 
-def test_spec_with_channel_and_subdir():
+@pytest.mark.parametrize("shared_pkgs_dirs", [True], indirect=True)
+def test_spec_with_channel_and_subdir(tmp_home, tmp_root_prefix):
     env_name = "myenv"
     try:
         res = helpers.create("-n", env_name, "conda-forge/noarch::xtensor", "--dry-run")