MAINT - Mock out call to env solve in test_generate_constructor_installer

conda-store-server/conda_store_server/_internal/build.py

@@ -136,7 +136,7 @@ def build_cleanup(
     build_active_tasks = collections.defaultdict(list)
     for worker_name, tasks in active_tasks.items():
         for task in tasks:
-            match = re.fullmatch("build-(\d+)-(.*)", str(task["id"]))
+            match = re.fullmatch(r"build-(\d+)-(.*)", str(task["id"]))
             if match:
                 build_id, name = match.groups()
                 build_active_tasks[build_id].append(task["name"])

conda-store-server/conda_store_server/_internal/orm.py

@@ -806,7 +806,9 @@ class KeyValueStore(Base):
     value = Column(JSON)
 
 
-def new_session_factory(url="sqlite:///:memory:", reset=False, **kwargs):
+def new_session_factory(
+    url="sqlite:///:memory:", reset=False, **kwargs
+) -> sessionmaker:
     engine = create_engine(
         url,
         # See the comment on the CustomJSONEncoder class on why this is needed

conda-store-server/conda_store_server/app.py

@@ -8,7 +8,7 @@
 import pydantic
 
 from celery import Celery, group
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
 from sqlalchemy.pool import QueuePool
 from traitlets import (
     Bool,
@@ -376,7 +376,7 @@ def _docker_base_image(build: orm.Build):
     )
 
     @property
-    def session_factory(self):
+    def session_factory(self) -> sessionmaker:
         if hasattr(self, "_session_factory"):
             return self._session_factory
 

conda-store-server/tests/test_actions.py

@@ -2,21 +2,26 @@
 import datetime
 import pathlib
 import re
-import subprocess
 import sys
+import tempfile
 
 from unittest import mock
 
 import pytest
 import yaml
 import yarl
 
+from conda.base.context import context as conda_base_context
+from constructor import construct
 from fastapi.responses import RedirectResponse
 from traitlets import TraitError
 
 from conda_store_server import BuildKey, api
 from conda_store_server._internal import action, conda_utils, orm, schema, server, utils
-from conda_store_server._internal.action import generate_lockfile
+from conda_store_server._internal.action import (
+    generate_constructor_installer,
+    generate_lockfile,
+)
 from conda_store_server.server.auth import DummyAuthentication
 
 
@@ -170,50 +175,70 @@ def test_solve_lockfile_multiple_platforms(conda_store, specification, request):
 def test_generate_constructor_installer(
     conda_store, specification_name, request, tmp_path
 ):
+    """Test that generate_construction_installer correctly produces the files needed by `constructor`."""
     specification = request.getfixturevalue(specification_name)
     installer_dir = tmp_path / "installer_dir"
     is_lockfile = specification_name in [
         "simple_lockfile_specification",
         "simple_lockfile_specification_with_pip",
     ]
 
-    # Creates the installer
-    context = action.action_generate_constructor_installer(
-        conda_command=conda_store.conda_command,
-        specification=specification,
-        installer_dir=installer_dir,
-        version="1",
-        is_lockfile=is_lockfile,
-    )
+    # action_generate_constructor_installer uses a temporary directory context manager
+    # to create and store the installer, but it usually gets deleted when the function
+    # exits. Here, we manually create that temporary directory, run the action,
+    # persisting the directory (so that we can verify the contents). Only then do we
+    # manually clean up afterward.
+    class PersistentTemporaryDirectory(tempfile.TemporaryDirectory):
+        def __exit__(self, exc, value, tb):
+            pass
+
+    temp_directory = None
+
+    def tmp_dir_side_effect(*args, **kwargs):
+        nonlocal temp_directory
+        temp_directory = PersistentTemporaryDirectory(*args, **kwargs)
+        return temp_directory
+
+    with mock.patch.object(
+        generate_constructor_installer, "tempfile", wraps=tempfile
+    ) as mock_tempfile:
+        mock_tempfile.TemporaryDirectory.side_effect = tmp_dir_side_effect
+
+        # Create the installer, but don't actually run `constructor` - it uses conda to solve the
+        # environment, which we don't need to do for the purposes of this test.
+        with mock.patch(
+            "conda_store_server._internal.action.generate_constructor_installer.logged_command"
+        ) as mock_command:
+            generate_constructor_installer.action_generate_constructor_installer(
+                conda_command=conda_store.conda_command,
+                specification=specification,
+                installer_dir=installer_dir,
+                version="1",
+                is_lockfile=is_lockfile,
+            )
 
-    # Checks that the installer was created
-    installer = context.result
-    assert installer.exists()
+    mock_command.assert_called()
 
-    tmp_dir = tmp_path / "tmp"
+    # First call to `constructor` is used to check that it is installed
+    mock_command.call_args_list[0].args[1] == ["constructor", "--help"]
 
-    # Runs the installer
-    out_dir = pathlib.Path(tmp_dir) / "out"
-    if sys.platform == "win32":
-        subprocess.check_output([installer, "/S", f"/D={out_dir}"])
-    else:
-        subprocess.check_output([installer, "-b", "-p", str(out_dir)])
+    # Second call is used to build the installer
+    call_args = mock_command.call_args_list[1].args[1]
+    cache_dir = pathlib.Path(call_args[3])
+    platform = call_args[5]
+    tmp_dir = pathlib.Path(call_args[6])
+    assert call_args[0:3] == ["constructor", "-v", "--cache-dir"]
+    assert str(cache_dir).endswith("pkgs")
+    assert call_args[4:6] == ["--platform", conda_base_context.subdir]
+    assert str(tmp_dir).endswith("build")
 
-    # Checks the output directory
-    assert out_dir.exists()
-    lib_dir = out_dir / "lib"
-    if specification_name in ["simple_specification", "simple_lockfile_specification"]:
-        if sys.platform == "win32":
-            assert any(str(x).endswith("zlib.dll") for x in out_dir.iterdir())
-        elif sys.platform == "darwin":
-            assert any(str(x).endswith("libz.dylib") for x in lib_dir.iterdir())
-        else:
-            assert any(str(x).endswith("libz.so") for x in lib_dir.iterdir())
-    else:
-        # Uses rglob to not depend on the version of the python
-        # directory, which is where site-packages is located
-        flask = pathlib.Path("site-packages") / "flask"
-        assert any(str(x).endswith(str(flask)) for x in out_dir.rglob("*"))
+    # Use some of the constructor internals to verify the action's artifacts are valid
+    # constructor input
+    info = construct.parse(str(tmp_dir / "construct.yaml"), platform)
+    construct.verify(info)
+
+    assert temp_directory is not None
+    temp_directory.cleanup()
 
 
 def test_fetch_and_extract_conda_packages(tmp_path, simple_conda_lock):