Fix tests

.github/workflows/tests.yml

@@ -16,6 +16,9 @@ jobs:
   test:
     name: ${{ matrix.platform }} py${{ matrix.python-version }}
     runs-on: ${{ matrix.platform }}
+    defaults:
+      run:
+        shell: bash -el {0}
     strategy:
       matrix:
         platform: [ubuntu-latest, windows-latest, macos-latest]
@@ -25,9 +28,10 @@ jobs:
       - uses: actions/checkout@v3
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: conda-incubator/setup-miniconda@v2
         with:
-          python-version: ${{ matrix.python-version }}
+          activate-environment: ""
+          auto-activate-base: true
 
       - name: Install dependencies constructor-manager-cli
         run: |

constructor-manager-cli/README.md

@@ -399,3 +399,10 @@ This command will install a specified on a fresh environment, deleting the old e
 ```bash
 constructor-manager restore "napari=0.4.16=*pyside*" -c conda-forge
 ```
+
+### Run tests
+
+```bash
+cd constructor-manager-cli/src
+pytest constructor_manager_cli --cov=constructor_manager_cli --cov-report term-missing
+```

constructor-manager-cli/src/constructor_manager_cli/actions.py

@@ -347,7 +347,16 @@ def lock_environment(
 ):
     """Lock the environment, using conda lock.
 
-    TODO
+    Parameters
+    ----------
+    package : str
+        Package specification.
+    channels : tuple of str, optional
+        Check for available versions on these channels.
+        Default is ``('conda-forge', )``.
+    platforms : tuple of str, optional
+        Platforms to lock for. If not provided conda is queried for the
+        current platform.
     """
     installer = CondaInstaller(channels=channels)
     yaml_spec = {"dependencies": [package]}
@@ -394,6 +403,3 @@ def lock_environment(
             if str(lockfile) != filepath and fh.read() == data:
                 lockfile.unlink()
                 break
-
-    # return installer._exit_codes[id]
-    return "1"

constructor-manager-cli/src/constructor_manager_cli/installer.py

@@ -46,7 +46,11 @@ def __init__(self) -> None:
 
     # -------------------------- Public API ------------------------------
     def install(
-        self, pkg_list: Sequence[str], *, prefix: Optional[str] = None
+        self,
+        pkg_list: Sequence[str],
+        *,
+        prefix: Optional[str] = None,
+        block: bool = False,
     ) -> job_id:
         """Install packages in `pkg_list` into `prefix`.
 
@@ -140,6 +144,13 @@ def _process_queue(self):
         self._processes[job_id] = popen
         if block:
             stdout, stderr = popen.communicate()
+
+            if isinstance(stdout, bytes):
+                stdout = stdout.decode()
+
+            if isinstance(stderr, bytes):
+                stderr = stderr.decode()
+
             # TODO: Write to file for logging and querying status
             return_code = popen.returncode
             self._on_process_finished(job_id, return_code, 0)
@@ -149,17 +160,27 @@ def _process_queue(self):
                 res = {"stdout": stdout}
             return res
         else:
-            for line in popen.stdout:
-                self._on_output(line)
-                # print(line, end='')
+            # for line in popen.stdout:
+            #     self._on_output(line)
+            #     # print(line, end='')
+
+            # for line in popen.stderr:
+            #     self._on_output(line)
+            #     # print(line, end='')
 
-            for line in popen.stderr:
-                self._on_output(line)
-                # print(line, end='')
+            # popen.stdout.close()
+            # popen.stderr.close()
+            # return_code = popen.wait()
+
+            while True:
+                output = popen.stdout.readline()
+                if isinstance(output, bytes):
+                    output = output.decode()
 
-            popen.stdout.close()
-            popen.stderr.close()
-            return_code = popen.wait()
+                if output == '' and popen.poll() is not None:
+                    break
+
+                return_code = popen.poll()
 
             self._on_process_finished(job_id, return_code, 0)
 
@@ -244,7 +265,10 @@ def _get_args(self, arg0, pkg_list: Sequence[str], prefix: Optional[str]):
         for channel in self._channels:
             cmd.extend(["-c", channel])
 
-        return tuple(cmd + list(pkg_list))
+        if pkg_list:
+            cmd.extend(pkg_list)
+
+        return tuple(cmd)
 
     # -------------------------- Public API ----------------------------------
     def info(self) -> Dict[Any, Any]:
@@ -253,120 +277,128 @@ def info(self) -> Dict[Any, Any]:
         res = cast(dict, self._queue_args(args, block=True))
         return res
 
-    def create(
-        self, pkg_list: Sequence[str], *, prefix: Optional[str] = None
-    ) -> job_id:
-        """Create a new conda environment with `pkg_list` in `prefix`.
+    def list(self, prefix: str, block=False) -> job_id:
+        """List packages for `prefix`.
 
         Parameters
         ----------
-        pkg_list : Sequence[str]
-            List of packages to install on new environment.
-        prefix : str, optional
-            Optional prefix for new environment.
+        prefix : str
+            Prefix from which to list packages.
 
         Returns
         -------
         job_id : int
             ID that can be used to cancel the process.
         """
-        return self._queue_args(self._get_create_args(pkg_list, prefix))
+        return self._queue_args(
+            ("list", "--prefix", str(prefix), "--json"), block=block
+        )
 
-    def remove(self, prefix) -> job_id:
-        """Remove a conda environment in `prefix`.
+    def lock(
+        self,
+        env_path: str,
+        platforms: Optional[Tuple[str, ...]] = None,
+        lockfile: Optional[str] = None,
+        block: bool = False,
+    ) -> job_id:
+        """List packages for `prefix`.
 
         Parameters
         ----------
-        prefix : str, optional
-            Optional prefix for new environment.
+        prefix : str
+            Prefix from which to list packages.
 
         Returns
         -------
         job_id : int
             ID that can be used to cancel the process.
         """
-        return self._queue_args(self._get_remove_args(prefix))
+        args = ["-f", env_path]
+        if platforms:
+            for platform in platforms:
+                args.extend(["-p", platform])
 
-    def install(
-        self, pkg_list: Sequence[str], *, prefix: Optional[str] = None
+        if lockfile:
+            args.extend(["--lockfile", lockfile])
+
+        return self._queue_args(args, bin="conda-lock", block=block)
+
+    def create(
+        self, prefix: str, *, pkg_list: Sequence[str] = (), block: bool = False
     ) -> job_id:
-        """Install packages in `pkg_list` into `prefix`.
+        """Create a new conda environment with `pkg_list` in `prefix`.
 
         Parameters
         ----------
         pkg_list : Sequence[str]
-            List of packages to install.
-        prefix : Optional[str], optional
-            Optional prefix to install packages into.
+            List of packages to install on new environment.
+        prefix : str, optional
+            Optional prefix for new environment.
 
         Returns
         -------
         job_id : int
             ID that can be used to cancel the process.
         """
-        return self._queue_args(self._get_install_args(pkg_list, prefix))
+        return self._queue_args(self._get_create_args(pkg_list, prefix), block=block)
 
-    def uninstall(
-        self, pkg_list: Sequence[str], *, prefix: Optional[str] = None
-    ) -> job_id:
-        """Uninstall packages in `pkg_list` from `prefix`.
+    def remove(self, prefix: str, block: bool = False) -> job_id:
+        """Remove a conda environment in `prefix`.
 
         Parameters
         ----------
-        pkg_list : Sequence[str]
-            List of packages to uninstall.
-        prefix : Optional[str], optional
-            Optional prefix from which to uninstall packages.
+        prefix : str, optional
+            Optional prefix for new environment.
 
         Returns
         -------
         job_id : int
             ID that can be used to cancel the process.
         """
-        return self._queue_args(self._get_uninstall_args(pkg_list, prefix))
+        return self._queue_args(self._get_remove_args(prefix), block=block)
 
-    def list(self, prefix: str, block=True) -> job_id:
-        """List packages for `prefix`.
+    def install(
+        self,
+        pkg_list: Sequence[str],
+        *,
+        prefix: Optional[str] = None,
+        block: bool = False,
+    ) -> job_id:
+        """Install packages in `pkg_list` into `prefix`.
 
         Parameters
         ----------
-        prefix : str
-            Prefix from which to list packages.
+        pkg_list : Sequence[str]
+            List of packages to install.
+        prefix : Optional[str], optional
+            Optional prefix to install packages into.
 
         Returns
         -------
         job_id : int
             ID that can be used to cancel the process.
         """
-        return self._queue_args(
-            ("list", "--prefix", str(prefix), "--json"), block=block
-        )
+        return self._queue_args(self._get_install_args(pkg_list, prefix), block=block)
 
-    def lock(
+    def uninstall(
         self,
-        env_path: str,
-        platforms: Optional[Tuple[str, ...]] = None,
-        lockfile: Optional[str] = None,
+        pkg_list: Sequence[str],
+        *,
+        prefix: Optional[str] = None,
         block: bool = False,
     ) -> job_id:
-        """List packages for `prefix`.
+        """Uninstall packages in `pkg_list` from `prefix`.
 
         Parameters
         ----------
-        prefix : str
-            Prefix from which to list packages.
+        pkg_list : Sequence[str]
+            List of packages to uninstall.
+        prefix : Optional[str], optional
+            Optional prefix from which to uninstall packages.
 
         Returns
         -------
         job_id : int
             ID that can be used to cancel the process.
         """
-        args = ["-f", env_path]
-        if platforms:
-            for platform in platforms:
-                args.extend(["-p", platform])
-
-        if lockfile:
-            args.extend(["--lockfile", lockfile])
-
-        return self._queue_args(args, bin="conda-lock", block=block)
+        return self._queue_args(self._get_uninstall_args(pkg_list, prefix), block=block)

constructor-manager-cli/src/constructor_manager_cli/main.py

@@ -231,8 +231,6 @@ def _handle_excecute(args, lock, lock_created=None):
     lock_created: bool, optional
         Whether the lock was created or not, by default ``None``.
     """
-    _execute(args, lock, lock_created)
-    return
     try:
         _execute(args, lock, lock_created)
     except Exception as e:

constructor-manager-cli/src/constructor_manager_cli/tests/test_installer.py

@@ -0,0 +1,63 @@
+import shutil
+import random
+
+import pytest
+
+from constructor_manager_cli.installer import CondaInstaller
+from constructor_manager_cli.utils.conda import get_base_prefix, get_prefix_by_name
+
+
+def test_conda_installer_info():
+    data = CondaInstaller().info()
+    assert data["conda_prefix"] == str(get_base_prefix())
+
+
+def test_conda_installer_list():
+    pkgs = CondaInstaller().list(get_base_prefix())
+    pkg_names = [pkg["name"] for pkg in pkgs]
+    assert pkgs
+    assert "conda" in pkg_names
+    assert "mamba" in pkg_names
+
+
+@pytest.mark.parametrize("use_mamba", [True, False])
+def test_conda_installer_create_remove(use_mamba):
+    prefix = get_prefix_by_name(
+        f"test-constructor-manager-{random.randint(1000, 9999)}"
+    )
+    if prefix.exists() and prefix.is_dir():
+        shutil.rmtree(prefix)
+    shutil.rmtree(prefix, ignore_errors=True)
+
+    installer = CondaInstaller(use_mamba=use_mamba)
+    # Create
+    _ = installer.create(prefix=prefix, block=True)
+    assert prefix.exists()
+    # assert installer._exit_codes[job_id] == 0
+
+    # Remove
+    _ = installer.remove(prefix=prefix, block=True)
+    assert not prefix.exists()
+    # assert installer._exit_codes[job_id] == 0
+
+
+@pytest.mark.parametrize("use_mamba", [True, False])
+def test_conda_installer_install_uninstall(use_mamba):
+    prefix = get_base_prefix()
+    pkg = "loghub"
+    installer = CondaInstaller(use_mamba=use_mamba)
+
+    # Install
+    _ = installer.install([pkg], prefix=prefix, block=True)
+
+    pkgs = installer.list(prefix=prefix)
+    print("HELLO", pkgs)
+    pkg_names = [pkg["name"] for pkg in pkgs]
+    assert pkg in pkg_names
+
+    # Uninstall
+    _ = installer.uninstall([pkg], prefix=prefix, block=True)
+    pkgs = installer.list(prefix=prefix)
+    print("HELLO 2", pkgs)
+    pkg_names = [pkg["name"] for pkg in pkgs]
+    assert pkg not in pkg_names