Add verify command (#3)

Author: RobertRosca

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pyvarium"
-version = "0.1.1"
+version = "0.2.0"
 description = "Tool for managing mixed Spack and pip packages"
 readme = "README.md"
 authors = ["Robert Rosca <[email protected]>"]

src/pyvarium/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.1"
+__version__ = "0.2.0"

src/pyvarium/cli/__init__.py

@@ -11,7 +11,7 @@
 from rich.markdown import Markdown
 from rich.prompt import Confirm
 
-from . import add, config, install, modulegen, new, sync
+from . import add, config, install, modulegen, new, sync, verify
 
 app = typer.Typer()
 
@@ -89,6 +89,7 @@ def main(
 app.add_typer(modulegen.app, name="modulegen")
 app.add_typer(new.app, name="new")
 app.add_typer(sync.app, name="sync")
+app.add_typer(verify.app, name="verify")
 
 
 if __name__ == "__main__":

src/pyvarium/cli/new.py

@@ -23,7 +23,7 @@ def main(path: Path = typer.Argument(..., file_okay=False)):
     with Status("Spack setup") as status:
         se = spack.SpackEnvironment(path, status=status)
         se.new()
-        se.add("python", "py-pip")
+        se.add("python", "py-pip", "py-setuptools")
         se.concretize()
         se.install()
 

src/pyvarium/cli/verify.py

@@ -0,0 +1,52 @@
+from pathlib import Path
+
+import typer
+from rich.status import Status
+from loguru import logger
+
+from pyvarium.installers import spack
+
+app = typer.Typer(
+    help="Check that python packages in view are still provided by spack."
+)
+
+
+@app.callback(invoke_without_command=True)
+def main(path: Path = typer.Option(".", file_okay=False), fix: bool = False):
+    path = path.resolve()
+
+    with Status("Checking status of Spack packages in view") as status:
+        se = spack.SpackEnvironment(path, status=status)
+        warnings = se.verify()
+
+        if all(len(w) == 0 for w in warnings.values()):
+            logger.info("All packages in view are correctly symlinked to spack")
+            raise typer.Exit(0)
+
+        for path, warning in warnings.items():
+            package_name = path.name
+            if len(warning) > 0:
+                logger.warning(
+                    f"[bold]{package_name}[/bold] has "
+                    f"[bold red]{len(warning)}[/bold red] files which are not linked correctly"
+                )
+                if fix:
+                    logger.info(f"Fixing {package_name}")
+                    for file, target in warning:
+                        if file.exists():
+                            file.unlink()
+                        file.symlink_to(target)
+
+        if fix:
+            logger.info("Re-checking status of Spack packages in view")
+            warnings = se.verify()
+            if all(len(w) == 0 for w in warnings.values()):
+                logger.info("All packages in view are correctly symlinked to spack")
+                raise typer.Exit(0)
+            else:
+                broken_packages = [p.name for p in warnings if len(warnings[p]) > 0]
+                logger.error(
+                    f"Some packages are still not linked correctly: {broken_packages}"
+                )
+
+        raise typer.Exit(1)

src/pyvarium/config.py

@@ -1,7 +1,7 @@
 import shutil
 from enum import Enum
 from pathlib import Path
-from typing import Dict, Optional, Tuple
+from typing import Dict, Optional
 
 import rtoml
 from dynaconf import Dynaconf  # type: ignore
@@ -16,7 +16,7 @@ class Scope(str, Enum):
     local = "local"
 
 
-def to_snake_case(d: dict) -> dict:
+def to_snake_case(d: Dict) -> Dict:
     """Recursively converts the keys of a dictionary to snake_case style."""
     res = {}
     for k, v in d.items():
@@ -28,7 +28,7 @@ def to_snake_case(d: dict) -> dict:
     return res
 
 
-def to_str(d: dict) -> dict:
+def to_str(d: Dict) -> Dict:
     """Recursively converts the values of a dictionary to str."""
     res = {}
     for k, v in d.items():
@@ -78,7 +78,7 @@ def load_dynaconf(cls) -> "Settings":
         )
 
         # TODO: there is a planned feature for dynaconf to allow defining schemas with
-        # pydantic directly, which would avoid this weird dynaconf -> dict -> dict with
+        # pydantic directly, which would avoid this weird dynaconf -> Dict -> Dict with
         # different keys -> pydantic steps
         __dynaconf_dict__ = to_snake_case(cls.__dynaconf_settings__.to_dict())  # type: ignore
 

src/pyvarium/installers/base.py

@@ -64,7 +64,10 @@ def cmd(self, *args) -> subprocess.CompletedProcess:
         logger.debug(res)
 
         if res.returncode != 0:
-            raise RuntimeError(f"Process return code is not 0: {res=}")
+            logger.error(res.stderr.decode())
+            raise RuntimeError(
+                f"Process return code is not 0: {res.args=}, {res.returncode=}"
+            )
 
         return res
 

src/pyvarium/installers/pipenv.py

@@ -26,6 +26,7 @@ class PipenvEnvironment(Environment):
     def __post_init__(self):
         self.program.cwd = self.path
         self.program.env["VIRTUAL_ENV"] = str((self.path / ".venv").absolute())
+        self.program.env["PIPENV_VERBOSITY"] = "-1"
         self.program.env["PIPENV_PYTHON"] = str(
             (self.path / ".venv" / "bin" / "python").absolute()
         )
@@ -49,10 +50,10 @@ def new(
         return self.program.cmd(*commands)
 
     def add(self, *packages):
-        return self.program.cmd("install", *packages)
+        return self.program.cmd("--site-packages", "install", *packages)
 
     def install(self):
-        return self.program.cmd("install")
+        return self.program.cmd("--site-packages", "install")
 
     def lock(self):
         return self.program.cmd("lock")

src/pyvarium/installers/spack.py

@@ -2,7 +2,7 @@
 import shutil
 import subprocess
 from pathlib import Path
-from typing import List, Literal, Union, overload
+from typing import List, Literal, Union, overload, Dict
 
 import yaml
 from loguru import logger
@@ -21,7 +21,7 @@ def recursive_dict_update(d, u):
     return d
 
 
-def cmd_json_to_dict(cmd: subprocess.CompletedProcess) -> dict:
+def cmd_json_to_dict(cmd: subprocess.CompletedProcess) -> Dict:
     return json.loads(cmd.stdout.decode())
 
 
@@ -87,18 +87,18 @@ def install(self):
 
         return self.cmd("install", "--only-concrete", "--no-add")
 
-    # def spec(self, spec: str) -> dict:
+    # def spec(self, spec: str) -> Dict:
     #     res = self.cmd("spec", "-I", "--reuse", "--json", spec)
     #     return cmd_json_to_dict(res)
 
     def concretize(self):
         return self.cmd("concretize", "--reuse")
 
-    def find(self) -> dict:
+    def find(self) -> Dict:
         res = self.cmd("find", "--json")
         return cmd_json_to_dict(res)
 
-    # def find_missing(self) -> dict:
+    # def find_missing(self) -> Dict:
     #     res = self.cmd(
     #         "find", "--show-concretized", "--deps", "--only-missing", "--json"
     #     )
@@ -130,18 +130,40 @@ def find_python_packages(
         packages_dict: List[dict] = json.loads(packages_json)
 
         if only_names:
-            return [
-                f"{p['name']}=={p['version']}"
-                for p in packages_dict
-                if p["name"] != "pip"
-            ]
+            return [f"{p['name']}=={p['version']}" for p in packages_dict]
         else:
             return packages_dict
 
-    def get_config(self) -> dict:
+    def verify(self) -> Dict[Path, list]:
+        view_path = self.path / ".venv"
+        packages = list((view_path / ".spack").iterdir())
+
+        package_warnings = {}
+
+        for package in packages:
+            self.program.update_status(f"{package.name}")
+            manifest_file = package / "install_manifest.json"
+            manifest = json.loads(manifest_file.read_text())
+            package_path = manifest_file.resolve().parent.parent
+            files = {
+                Path(k.replace(str(package_path), str(view_path))): Path(k)
+                for k, v in manifest.items()
+                if v["type"] == "file" and ".spack" not in k and "bin" not in k
+            }
+
+            warnings = []
+            for link, target in files.items():
+                if not link.resolve() == target.resolve():
+                    warnings.append((link, target))
+
+            package_warnings[package] = warnings
+
+        return package_warnings
+
+    def get_config(self) -> Dict:
         return yaml.safe_load((self.path / "spack.yaml").read_text())
 
-    def set_config(self, config: dict) -> None:
+    def set_config(self, config: Dict) -> None:
         current_config = self.get_config()
         new_config = recursive_dict_update(current_config, config)
 

tests/cli/test_verify.py

@@ -0,0 +1,46 @@
+import os
+from pathlib import Path
+
+import pytest
+from typer.testing import CliRunner
+
+from pyvarium.cli import app
+
+runner = CliRunner()
+
+
+@pytest.fixture(autouse=True, scope="module")
+def tmp_cwd(tmp_path_factory):
+    tmpdir = tmp_path_factory.mktemp("cli.verify")
+    os.chdir(tmpdir)
+
+    res = runner.invoke(app, ["new", "test-env"])
+    assert res.exit_code == 0
+
+    os.chdir(tmpdir / "test-env")
+
+    res = runner.invoke(app, ["add", "spack", "py-numpy"])
+    assert res.exit_code == 0
+
+    yield tmpdir / "test-env"
+
+
+def test_verify_success(tmp_cwd: Path):
+    res = runner.invoke(app, ["verify"])
+    venv_site_packages = tmp_cwd / ".venv" / "lib" / "python3.8" / "site-packages"
+    assert (venv_site_packages / "numpy" / "version.py").is_symlink()
+    assert res.exit_code == 0
+    assert "correctly symlinked to spack" in res.stdout
+
+
+@pytest.mark.xfail(
+    reason="binary cache injects placeholders to path which breaks tests in image"
+)
+def test_verify_failure(tmp_cwd: Path):
+    venv_site_packages = tmp_cwd / ".venv" / "lib" / "python3.8" / "site-packages"
+    numpy_file = venv_site_packages / "numpy" / "version.py"
+    numpy_file.unlink()
+    assert not numpy_file.is_symlink()
+    res = runner.invoke(app, ["verify"])
+    print(res.stdout)
+    assert res.exit_code != 0