Add tests

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

@@ -3,6 +3,6 @@
 # license that can be found in the LICENSE file.
 
 from conda_store_server._internal.plugins.lock.conda_lock import conda_lock
-from conda_store_server._internal.plugins.lock.other_lock import other_lock
 
-BUILTIN_PLUGINS = [conda_lock, other_lock]
+
+BUILTIN_PLUGINS = [conda_lock]

conda-store-server/conda_store_server/plugins/plugin_context.py

@@ -17,7 +17,11 @@ class PluginContext:
     """
 
     def __init__(
-        self, conda_store=None, stdout=None, stderr=None, log_level=logging.INFO
+        self,
+        conda_store=None,
+        stdout=None,
+        stderr=None,
+        log_level=logging.INFO,
     ):
         if stdout is not None and stderr is None:
             stderr = stdout
@@ -33,7 +37,12 @@ def __init__(
         self.log.setLevel(log_level)
         self.conda_store = conda_store
 
-    def run_command(self, command, redirect_stderr=True, **kwargs):
+    def run_command(
+            self,
+            command: str,
+            redirect_stderr: bool = True,
+            **kwargs
+        ):
         """Runs command and immediately writes to logs"""
         self.log.info("Running command: %s", command)
 
@@ -49,12 +58,13 @@ def run_command(self, command, redirect_stderr=True, **kwargs):
             bufsize=1,
             universal_newlines=True,
             **kwargs,
-        ) as p:
-            for line in p.stdout:
+        ) as proc:
+            for line in proc.stdout:
                 self.stdout.write(line)
             if not redirect_stderr:
-                for line in p.stderr:
+                import pdb; pdb.set_trace()
+                for line in proc.stderr:
                     self.stderr.write(line)
 
-        if p.returncode != 0:
-            raise subprocess.CalledProcessError(p.returncode, p.args)
+        if proc.returncode != 0:
+            raise subprocess.CalledProcessError(proc.returncode, proc.args)

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

@@ -0,0 +1,61 @@
+# 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 logging
+import subprocess
+import pytest
+import os
+
+from conda_store_server.plugins.plugin_context import PluginContext
+
+class TestOutPutCapture:
+    def __init__(self):
+        self.output = ""
+
+    def write(self, line:str):
+        self.output += line
+
+
+def test_run_command_no_logs():
+    out = TestOutPutCapture()
+    err = TestOutPutCapture()
+    pr = PluginContext(stdout=out, stderr=err, log_level=logging.ERROR)
+
+    pr.run_command(["echo", "testing"])
+    assert err.output == ''
+    assert out.output == 'testing\n'
+
+
+def test_run_command_log_info():
+    out = TestOutPutCapture()
+    err = TestOutPutCapture()
+    pr = PluginContext(stdout=out, stderr=err, log_level=logging.INFO)
+
+    pr.run_command(["echo", "testing"])
+    assert err.output == ''
+    assert out.output == """Running command: ['echo', 'testing']
+testing
+"""
+
+
+def test_run_command_errors():
+    out = TestOutPutCapture()
+    err = TestOutPutCapture()
+    pr = PluginContext(stdout=out, stderr=err, log_level=logging.ERROR)
+
+    with pytest.raises(subprocess.CalledProcessError):
+        pr.run_command(["conda-store-server", "-thiswillreturnanonzeroexitcode"])
+
+
+def test_run_command_kwargs():
+    """Ensure that kwargs get passed to subprocess"""
+    out = TestOutPutCapture()
+    err = TestOutPutCapture()
+    pr = PluginContext(stdout=out, stderr=err, log_level=logging.ERROR)
+
+    # set the cwd to this directory and check that this file exists
+    dir_path = os.path.dirname(os.path.realpath(__file__))
+    pr.run_command(["stat", "test_plugin_context.py"], check=True, cwd=dir_path)
+    assert err.output == ''
+    assert "File: test_plugin_context.py" in out.output

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

@@ -2,10 +2,39 @@
 # Use of this source code is governed by a BSD-style
 # license that can be found in the LICENSE file.
 
-from conda_store_server.plugins import BUILTIN_PLUGINS
+import pytest
+
+from conda_store_server.plugins import BUILTIN_PLUGINS, types
+from conda_store_server._internal.plugins.lock.conda_lock import conda_lock
+from conda_store_server.exception import CondaStorePluginNotFoundError
 
 
 def test_collect_plugins(plugin_manager):
     plugin_manager.collect_plugins()
     for plugin in BUILTIN_PLUGINS:
         assert plugin_manager.is_registered(plugin)
+
+
+def test_get_lock_plugins(plugin_manager):
+    plugin_manager.collect_plugins()
+    lp = plugin_manager.get_lock_plugins()
+
+    # Ensure built in lock plugins are accounted for
+    assert "lock-conda_lock" in lp
+
+    # Ensure all plugins are lock plugins
+    for plugin in lp.values():
+        assert isinstance(plugin, types.TypeLockPlugin) 
+
+
+def get_lock_plugin(plugin_manager):
+    plugin_manager.collect_plugins()
+    lp = plugin_manager.lock_plugin("lock-conda_lock")
+    assert lp.name == "lock-conda_lock"
+    assert lp.backend == conda_lock.CondaLock
+
+
+def get_lock_plugin_does_not_exist(plugin_manager):
+    plugin_manager.collect_plugins()
+    with pytest.raises(CondaStorePluginNotFoundError):
+        plugin_manager.lock_plugin("i really don't exist")

conda-store-server/tests/test_app.py

@@ -9,6 +9,8 @@
 
 from conda_store_server import api
 from conda_store_server._internal import action, conda_utils, schema
+from conda_store_server._internal.plugins.lock.conda_lock import conda_lock
+from conda_store_server.exception import CondaStorePluginNotFoundError
 
 
 @pytest.mark.long_running_test
@@ -181,4 +183,15 @@ def test_conda_store_register_environment_duplicate_force_true(db, conda_store):
 
 
 def test_conda_store_get_lock_plugin(conda_store):
-    assert conda_store.lock_plugin() is not None
+    lock_plugin_setting = "lock-conda_lock"
+    conda_store.lock_plugin_name = lock_plugin_setting
+    name, plugin = conda_store.lock_plugin()
+    assert name == lock_plugin_setting
+    assert isinstance(plugin, conda_lock.CondaLock)
+
+
+def test_conda_store_get_lock_plugin_does_not_exist(conda_store):
+    lock_plugin_setting = "idontexist"
+    conda_store.lock_plugin_name = lock_plugin_setting
+    with pytest.raises(CondaStorePluginNotFoundError):
+        conda_store.lock_plugin()