Add tests for lock plugin

conda-store-server/tests/_internal/action/test_actions.py

@@ -11,7 +11,6 @@
 from unittest import mock
 
 import pytest
-import yaml
 import yarl
 from celery.result import AsyncResult
 from conda.base.context import context as conda_base_context
@@ -23,7 +22,6 @@
 from conda_store_server._internal import action, conda_utils, orm, schema, server, utils
 from conda_store_server._internal.action import (
     generate_constructor_installer,
-    generate_lockfile,
 )
 from conda_store_server.server.auth import DummyAuthentication
 
@@ -72,108 +70,6 @@ def test_function(context):
     assert not context.result.exists()
 
 
-@pytest.mark.parametrize(
-    "specification",
-    [
-        "simple_specification",
-        "simple_specification_with_pip",
-    ],
-)
-@mock.patch.object(generate_lockfile, "yaml", wraps=yaml)
-@mock.patch("conda_store_server._internal.action.generate_lockfile.logged_command")
-@mock.patch("conda_store_server._internal.action.generate_lockfile.run_lock")
-@pytest.mark.long_running_test
-def test_solve_lockfile(
-    mock_run_lock,
-    mock_logged_command,
-    mock_yaml,
-    conda_store,
-    specification,
-    request,
-):
-    """Test that the call to conda_lock.run_lock is formed correctly.
-
-    Mock out logged_command, which is used to call `conda info`; this is an
-    extremely slow call, and we aren't testing any aspect of it here.
-
-    Mock out the yaml package, it is used to load the lockfile normally
-    generated by conda-lock.
-    """
-
-    # Dump dummy data to the expected lockfile output location
-    def run_lock_side_effect(lockfile_path, **kwargs):
-        with open(lockfile_path, "w") as f:
-            yaml.dump({"foo": "bar"}, f)
-
-    mock_run_lock.side_effect = run_lock_side_effect
-
-    platforms = [conda_utils.conda_platform()]
-    specification = request.getfixturevalue(specification)
-    if specification.variables is None:
-        cuda_version = None
-    else:
-        cuda_version = specification.variables.get("CONDA_OVERRIDE_CUDA")
-
-    context = generate_lockfile.action_solve_lockfile(
-        conda_command=conda_store.conda_command,
-        specification=specification,
-        platforms=platforms,
-    )
-
-    # Check that the call to `conda_lock` is correctly formed
-    mock_run_lock.assert_called_once()
-    call_args = mock_run_lock.call_args_list[0][1]
-    assert str(call_args["environment_files"][0]).endswith("environment.yaml")
-    assert call_args["platforms"] == platforms
-    assert str(call_args["lockfile_path"]).endswith("conda-lock.yaml")
-    assert call_args["conda_exe"] == conda_store.conda_command
-    assert call_args["with_cuda"] == cuda_version
-
-    assert context.result["foo"] == "bar"
-
-
-def test_solve_lockfile_valid_conda_flags(conda_store, simple_specification):
-    context = action.action_solve_lockfile(
-        conda_command=conda_store.conda_command,
-        specification=simple_specification,
-        platforms=[conda_utils.conda_platform()],
-        conda_flags="--strict-channel-priority",
-    )
-    assert len(context.result["package"]) != 0
-
-
-# Checks that conda_flags is used by conda-lock
-@pytest.mark.long_running_test
-def test_solve_lockfile_invalid_conda_flags(conda_store, simple_specification):
-    with pytest.raises(
-        Exception, match=(r"Command.*--this-is-invalid.*returned non-zero exit status")
-    ):
-        action.action_solve_lockfile(
-            conda_command=conda_store.conda_command,
-            specification=simple_specification,
-            platforms=[conda_utils.conda_platform()],
-            conda_flags="--this-is-invalid",
-        )
-
-
-@pytest.mark.parametrize(
-    "specification",
-    [
-        "simple_specification",
-        "simple_specification_with_pip",
-    ],
-)
-@pytest.mark.long_running_test
-def test_solve_lockfile_multiple_platforms(conda_store, specification, request):
-    specification = request.getfixturevalue(specification)
-    context = action.action_solve_lockfile(
-        conda_command=conda_store.conda_command,
-        specification=specification,
-        platforms=["osx-64", "linux-64", "win-64", "osx-arm64"],
-    )
-    assert len(context.result["package"]) != 0
-
-
 @pytest.mark.parametrize(
     "specification_name",
     [

conda-store-server/tests/plugins/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (c) conda-store development team. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file.

conda-store-server/tests/plugins/lock/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (c) conda-store development team. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file.

conda-store-server/tests/plugins/lock/test_conda_lock.py

@@ -0,0 +1,123 @@
+# Copyright (c) conda-store development team. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file.
+
+import pytest
+import yaml
+import os
+
+from unittest import mock
+
+from conda_store_server._internal import conda_utils
+from conda_store_server.plugins import plugin_context
+from conda_store_server.plugins.lock import conda_lock
+
+
+@pytest.mark.parametrize(
+    "specification",
+    [
+        "simple_specification",
+        "simple_specification_with_pip",
+    ],
+)
+@mock.patch("conda_store_server.plugins.lock.conda_lock.run_lock")
+@pytest.mark.long_running_test
+def test_solve_lockfile(
+    mock_run_lock,
+    specification,
+    tmp_path,
+    request,
+):
+    """Test that the call to conda_lock.run_lock is formed correctly.
+    """
+    tmp_path.mkdir(exist_ok=True)
+    os.chdir(tmp_path)
+
+    # Dump dummy data to the expected lockfile output location
+    def run_lock_side_effect(lockfile_path, **kwargs):
+        with open(lockfile_path, "w") as f:
+            yaml.dump({"foo": "bar"}, f)
+
+    mock_run_lock.side_effect = run_lock_side_effect
+
+    platforms = [conda_utils.conda_platform()]
+    specification = request.getfixturevalue(specification)
+    if specification.variables is None:
+        cuda_version = None
+    else:
+        cuda_version = specification.variables.get("CONDA_OVERRIDE_CUDA")
+
+    locker = conda_lock.CondaLock(conda_command="mamba")
+    lock_result = locker.lock_environment(
+        context=plugin_context.PluginContext(),
+        spec=specification,
+        platforms=platforms,
+    )
+
+    # Check that the call to `conda_lock` is correctly formed
+    mock_run_lock.assert_called_once()
+    call_args = mock_run_lock.call_args_list[0][1]
+    assert str(call_args["environment_files"][0]).endswith("environment.yaml")
+    assert call_args["platforms"] == platforms
+    assert str(call_args["lockfile_path"]).endswith("conda-lock.yaml")
+    assert call_args["conda_exe"] == "mamba"
+    assert call_args["with_cuda"] == cuda_version
+
+    assert lock_result["foo"] == "bar"
+
+
+def test_solve_lockfile_simple(tmp_path, simple_specification):
+    tmp_path.mkdir(exist_ok=True)
+    os.chdir(tmp_path)
+
+    locker = conda_lock.CondaLock(
+        conda_command="mamba", conda_flags="--strict-channel-priority"
+    )
+    lock_result = locker.lock_environment(
+        context=plugin_context.PluginContext(),
+        spec=simple_specification,
+        platforms=[conda_utils.conda_platform()],
+    )
+    assert len(lock_result["package"]) != 0
+
+
+@pytest.mark.parametrize(
+    "specification",
+    [
+        "simple_specification",
+        "simple_specification_with_pip",
+    ],
+)
+@pytest.mark.long_running_test
+def test_solve_lockfile_multiple_platforms(tmp_path, specification, request):
+    tmp_path.mkdir(exist_ok=True)
+    os.chdir(tmp_path)
+
+    specification = request.getfixturevalue(specification)
+    locker = conda_lock.CondaLock(
+        conda_command="mamba", conda_flags="--strict-channel-priority"
+    )
+    lock_result = locker.lock_environment(
+        context=plugin_context.PluginContext(),
+        spec=specification,
+        platforms=["osx-64", "linux-64", "win-64", "osx-arm64"],
+    )
+    assert len(lock_result["package"]) != 0
+
+# Checks that conda_flags is used by conda-lock
+def test_solve_lockfile_invalid_conda_flags(tmp_path, simple_specification):
+    tmp_path.mkdir(exist_ok=True)
+    os.chdir(tmp_path)
+
+    locker = conda_lock.CondaLock(
+        conda_command="mamba", conda_flags="--this-is-invalid"
+    )
+
+    with pytest.raises(
+        Exception, match=(r"Command.*--this-is-invalid.*returned non-zero exit status")
+    ):
+         locker.lock_environment(
+            context=plugin_context.PluginContext(),
+            spec=simple_specification,
+            platforms=[conda_utils.conda_platform()],
+        )