Merge pull request #171 from fmi-faim/pixi-helpers

Add pixi utility modules

pyproject.toml

@@ -81,6 +81,7 @@ python = ["3.10", "3.11", "3.12"]
 dependencies = [
   "coverage[toml]>=6.5",
   "pytest",
+  "pytest-mock",
 ]
 
 [tool.hatch.envs.default.scripts]

src/faim_ipa/pixi/cache_status.py

@@ -0,0 +1,37 @@
+import json
+import shutil
+import subprocess
+import sys
+from inspect import cleandoc
+
+GB = 1024 * 1024 * 1024
+
+
+def min_cache_size_gb(gb: int = 2):
+    info = subprocess.run(
+        ["pixi", "info", "--json"],  # noqa: S607
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    cache_dir = json.loads(info.stdout)["cache_dir"]
+
+    if shutil.disk_usage(cache_dir).free < gb * GB:
+        sys.tracebacklimit = 0
+        message = cleandoc(
+            f"""
+        Disk space in cache directory is below {gb} GB.
+
+        PIXI_CACHE_DIR: {cache_dir}
+
+        Did you initialize your session correctly ('source ./init.sh') ?
+        """
+        )
+        raise RuntimeError(message)
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        min_cache_size_gb(gb=int(sys.argv[1]))
+    min_cache_size_gb()

src/faim_ipa/pixi/log_commit.py

@@ -0,0 +1,22 @@
+import os
+import subprocess
+import sys
+
+from faim_ipa.utils import create_logger
+
+
+def log_commit(task: str = ""):
+    command = ["git", "rev-parse", "--verify", "HEAD"]
+    info = subprocess.run(command, capture_output=True, text=True, check=False)
+    hash_string = info.stdout.strip()
+
+    os.chdir(path=os.getenv("WD", default="."))
+    logger = create_logger(name="githash", include_timestamp=False)
+    logger.info("[Executing] %s [git-commit] %s", task, hash_string)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        log_commit(task=str(sys.argv[1]))
+    else:
+        log_commit()

src/faim_ipa/pixi/src_status.py

@@ -0,0 +1,27 @@
+import subprocess
+import sys
+from inspect import cleandoc
+
+
+def git_status_clean(src_dir: str = "src"):
+    command = ["git", "status", "--porcelain", src_dir]
+    info = subprocess.run(command, capture_output=True, text=True, check=False)
+    changes = len(info.stdout.splitlines())
+
+    if changes > 0:
+        sys.tracebacklimit = 0
+        message = cleandoc(
+            f"""
+        There are {changes} untracked changes in {src_dir}.
+
+        Please commit or stash before proceeding.
+        """
+        )
+        raise RuntimeError(message)
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        git_status_clean(src_dir=str(sys.argv[1]))
+    git_status_clean()

src/faim_ipa/utils.py

@@ -1,7 +1,7 @@
 import logging
-import os.path
-import pathlib
+import os
 from datetime import datetime
+from pathlib import Path
 
 import pydantic
 from pydantic import BaseModel
@@ -60,7 +60,7 @@ def rgb_to_hex(r, g, b):
     return f"{r:02x}{g:02x}{b:02x}"
 
 
-def create_logger(name: str) -> logging.Logger:
+def create_logger(name: str, *, include_timestamp: bool = True) -> logging.Logger:
     """
     Create logger which logs to <timestamp>-<name>.log inside the current
     working directory.
@@ -70,19 +70,22 @@ def create_logger(name: str) -> logging.Logger:
     name
         Name of the logger instance.
     """
-    logger = logging.getLogger(name.capitalize())
+    logger = logging.getLogger(name=name)
     now = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
-    handler = logging.FileHandler(f"{now}-{name}.log")
+    handler = logging.FileHandler(
+        f"{now}-{name}.log" if include_timestamp else f"{name}.log"
+    )
     formatter = logging.Formatter(
-        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+        fmt="{asctime} - {name} - {levelname} - {message}",
+        style="{",
     )
     handler.setFormatter(formatter)
     logger.addHandler(handler)
     logger.setLevel(logging.INFO)
     return logger
 
 
-def get_git_root() -> pathlib.Path:
+def get_git_root() -> Path:
     """
     Recursively search for the directory containing the .git folder.
 
@@ -91,14 +94,14 @@ def get_git_root() -> pathlib.Path:
     Path
         Path to the root of the git repository.
     """
-    parent_dir = pathlib.Path(__file__).parent
+    parent_dir = Path(__file__).parent
     while not (parent_dir / ".git").exists():
         parent_dir = parent_dir.parent
 
     return parent_dir
 
 
-def resolve_with_git_root(relative_path: pathlib.Path) -> pathlib.Path:
+def resolve_with_git_root(relative_path: Path) -> Path:
     """
     Takes a relative path and resolves it relative to the git_root directory.
 
@@ -116,7 +119,7 @@ def resolve_with_git_root(relative_path: pathlib.Path) -> pathlib.Path:
     return (git_root / relative_path).resolve()
 
 
-def make_relative_to_git_root(path: pathlib.Path) -> pathlib.Path:
+def make_relative_to_git_root(path: Path) -> Path:
     """
     Convert an absolute path to a path relative to the git_root directory.
 
@@ -136,7 +139,7 @@ def make_relative_to_git_root(path: pathlib.Path) -> pathlib.Path:
         return path.relative_to(git_root, walk_up=True)
     except (ValueError, TypeError):
         # fallback for Python < 3.12
-        return pathlib.Path(os.path.relpath(path, git_root))
+        return Path(os.path.relpath(path, git_root))
 
 
 class IPAConfig(BaseModel):
@@ -145,7 +148,7 @@ def make_paths_absolute(self):
         """
         Convert all `pathlib.Path` fields to absolute paths.
 
-        The paths are assumed to be relative to a git-root directory somewhere
+        The paths are assumed to be relative to a git root directory somewhere
         in the parent directories of the class implementing `IPAConfig`.
         """
         fields = (
@@ -156,7 +159,7 @@ def make_paths_absolute(self):
 
         for f in fields:
             attr = getattr(self, f)
-            if isinstance(attr, pathlib.Path) and not attr.is_absolute():
+            if isinstance(attr, Path) and not attr.is_absolute():
                 setattr(self, f, resolve_with_git_root(attr))
 
     def make_paths_relative(self):
@@ -175,5 +178,5 @@ def make_paths_relative(self):
 
         for f in fields:
             attr = getattr(self, f)
-            if isinstance(attr, pathlib.Path) and attr.is_absolute():
+            if isinstance(attr, Path) and attr.is_absolute():
                 setattr(self, f, make_relative_to_git_root(attr))

tests/pixi/test_pixi_utils.py

@@ -0,0 +1,123 @@
+import os
+import re
+from inspect import cleandoc
+from typing import NamedTuple
+
+import pytest
+
+from faim_ipa import pixi
+from faim_ipa.pixi.cache_status import min_cache_size_gb
+from faim_ipa.pixi.log_commit import log_commit
+from faim_ipa.pixi.src_status import git_status_clean
+
+
+def test_src_status_fail(mocker):
+    mock_result = mocker.Mock()
+    mock_result.stdout = cleandoc(
+        """
+        A some/file
+        M some/other/file
+        """
+    )
+    mock_result.returncode = 1
+
+    mocker.patch("faim_ipa.pixi.src_status.subprocess.run", return_value=mock_result)
+
+    with pytest.raises(
+        RuntimeError, match="There are 2 untracked changes in mock_source"
+    ):
+        git_status_clean(src_dir="mock_source")
+
+    pixi.src_status.subprocess.run.assert_called_once_with(
+        ["git", "status", "--porcelain", "mock_source"],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def test_src_status_succeed(mocker):
+    mock_result = mocker.Mock()
+    mock_result.stdout = ""
+    mock_result.returncode = 0
+
+    mocker.patch("faim_ipa.pixi.src_status.subprocess.run", return_value=mock_result)
+
+    with pytest.raises(SystemExit) as excinfo:
+        git_status_clean(src_dir="mock_source")
+    assert excinfo.value.code == 0
+
+    pixi.src_status.subprocess.run.assert_called_once_with(
+        ["git", "status", "--porcelain", "mock_source"],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def test_cache_status_fail(mocker):
+    mock_result_pixi = mocker.Mock()
+    mock_result_pixi.stdout = '{"cache_dir": "/some/folder"}'
+    mock_result_pixi.returncode = 0
+
+    class Usage(NamedTuple):
+        free: int
+
+    mock_usage = Usage
+    mock_disk_usage = mocker.Mock(return_value=mock_usage(free=0))
+
+    mocker.patch(
+        "faim_ipa.pixi.cache_status.subprocess.run", return_value=mock_result_pixi
+    )
+    mocker.patch(
+        "faim_ipa.pixi.cache_status.shutil.disk_usage", side_effect=mock_disk_usage
+    )
+
+    with pytest.raises(
+        RuntimeError, match="Disk space in cache directory is below 1 GB"
+    ):
+        min_cache_size_gb(gb=1)
+
+    pixi.cache_status.subprocess.run.assert_called_once_with(
+        ["pixi", "info", "--json"], text=True, capture_output=True, check=False
+    )
+    pixi.cache_status.shutil.disk_usage.assert_called_once_with("/some/folder")
+
+
+def test_cache_status_succeed(mocker):
+    mock_result_pixi = mocker.Mock()
+    mock_result_pixi.stdout = '{"cache_dir": "/some/folder"}'
+    mock_result_pixi.returncode = 0
+
+    class Usage(NamedTuple):
+        free: int
+
+    mock_usage = Usage
+    mock_disk_usage = mocker.Mock(return_value=mock_usage(free=2 * 1024 * 1024 * 1024))
+
+    mocker.patch(
+        "faim_ipa.pixi.cache_status.subprocess.run", return_value=mock_result_pixi
+    )
+    mocker.patch(
+        "faim_ipa.pixi.cache_status.shutil.disk_usage", side_effect=mock_disk_usage
+    )
+
+    with pytest.raises(SystemExit) as excinfo:
+        min_cache_size_gb()
+    assert excinfo.value.code == 0
+
+    pixi.cache_status.subprocess.run.assert_called_once_with(
+        ["pixi", "info", "--json"], text=True, capture_output=True, check=False
+    )
+    pixi.cache_status.shutil.disk_usage.assert_called_once_with("/some/folder")
+
+
+def test_log_commit(tmp_path):
+    os.environ["WD"] = str(tmp_path)
+    log_commit(task="TESTING")
+    log_path = tmp_path / "githash.log"
+    assert log_path.exists()
+    pattern = r"^\d{4}-\d{2}-\d{2}.*githash.*INFO.*Executing.*TESTING.*git-commit\] [0-9a-f]*$"
+    with open(log_path) as log:
+        assert re.match(pattern=pattern, string=log.readline())
+        assert not log.readline()

tests/test_utils.py

@@ -20,7 +20,7 @@ def test_create_logger(tmp_path_factory):
     os.chdir(tmp_path_factory.mktemp("logs"))
     logger = create_logger("test")
     logger.info("Test")
-    assert logger.name == "Test"
+    assert logger.name == "test"
     assert basename(logger.handlers[0].baseFilename).endswith("-test.log")
 
     with open(logger.handlers[0].baseFilename) as f: