Add caching proxy

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

@@ -10,6 +10,8 @@
 
 from . import helpers
 
+__this_dir__ = pathlib.Path(__file__).parent.resolve()
+
 ####################
 #  Config options  #
 ####################
@@ -23,6 +25,12 @@ def pytest_addoption(parser):
         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",
@@ -44,21 +52,48 @@ def pytest_addoption(parser):
 
 
 ###########################
-#  Caching proxy fixture  #
+#  Test utility fixtures  #
 ###########################
 
 
 @pytest.fixture(scope="session")
-def mitmdump_exe(request) -> pathlib.Path:
+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"),
+    )
 
-    return pathlib.Path(shutil.which("mitmdump")).resolve()
+    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() -> None:
+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)
 
@@ -99,6 +134,29 @@ def tmp_environ(session_clean_env: None) -> Generator[Mapping[str, Any], None, N
     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

micromamba/tests/test_create.py

@@ -341,6 +341,7 @@ def test_multiple_spec_files(tmp_home, tmp_root_prefix, tmp_path, type):
 
 
 @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"]

micromamba/tests/test_proxy.py

@@ -11,6 +11,7 @@
 
 @pytest.mark.parametrize("auth", [None, "foo:bar", "user%40example.com:pass"])
 @pytest.mark.parametrize("ssl_verify", (True, False))
+@pytest.mark.parametrize("use_caching_proxy", [False], indirect=True)
 def test_proxy_install(
     mitmdump_exe, tmp_home, tmp_prefix, tmp_path, unused_tcp_port, auth, ssl_verify
 ):