.gitignore

@@ -22,7 +22,6 @@ test.*
 pre-drv
 post-drv
 post-diff
-*.nixoutput
 
 # ruff cache
 .ruff_cache

flupdt/__init__.py

@@ -1 +0,0 @@
-"""Tools to build, evaluate, and parse a nix flake."""

flupdt/cli.py

@@ -1,16 +1,9 @@
-"""Manages the CLI component of the tool."""
-
 import argparse
 
 
 def parse_inputs() -> argparse.Namespace:
-    """Parse inputs from argparse.
-
-    :returns the argparse Namespace to be evaluated
-    """
     parser = argparse.ArgumentParser()
     parser.add_argument("flake_path", metavar="flake-path", help="path to flake to evaluate")
-    parser.add_argument("--keep-hydra", action="store_true", help="retain Hydra jobs")
+    parser.add_argument("--keep-hydra", action="store_true", help="allow evaluating Hydra jobs")
-    parser.add_argument("--build", action="store_true", help="allow building Hydra jobs")
+    args = parser.parse_args()
-    parser.add_argument("--evaluate", action="store_true", help="allow evaluating Hydra jobs")
+    return args
-    return parser.parse_args()

flupdt/common.py

@@ -1,11 +1,9 @@
-"""Common utilities."""
+"""common."""
 
 import itertools
 import logging
 import sys
-from collections.abc import Iterable
+from subprocess import Popen,PIPE
-from subprocess import PIPE, Popen
-from types import FunctionType
 
 
 def configure_logger(level: str = "INFO") -> None:
@@ -22,7 +20,7 @@ def configure_logger(level: str = "INFO") -> None:
     )
 
 
-def partition(predicate: FunctionType, iterable: Iterable) -> tuple[Iterable, Iterable]:
+def partition(predicate, iterable):
     """Partition entries into false entries and true entries.
 
     If *predicate* is slow, consider wrapping it with functools.lru_cache().

flupdt/flake_build.py

@@ -1,31 +0,0 @@
-"""Provides components to build nix components and process the result."""
-
-from __future__ import annotations
-
-import logging
-import re
-
-from flupdt.common import bash_wrapper
-
-drv_re = re.compile(r".*(/nix/store/.*\.drv).*")
-
-
-def build_output(path: str, output: str) -> str | None:
-    """Builds a given output in a flake.
-
-    :param path: path to flake
-    :param output: flake output to be built
-    :returns the .drv path on success or None on failure
-    """
-    logging.info(f"build {output}")
-    out = bash_wrapper(f"nix build {path}#{output} -o {output}.nixoutput")
-    logging.debug("output")
-    logging.debug(out[0])
-    logging.debug("error")
-    logging.debug(out[1])
-    logging.debug("statuscode")
-    logging.debug(out[2])
-    if out[2] != 0:
-        logging.warning(f"output {output} did not build correctly")
-        return None
-    return ""

flupdt/flake_eval.py

@@ -1,23 +1,14 @@
-"""Provides components to evaluate nix components and process the result."""
-
 from __future__ import annotations
 
 import logging
-import re
+from typing import Optional
-
 from flupdt.common import bash_wrapper
+import re
 
 drv_re = re.compile(r".*(/nix/store/.*\.drv).*")
 
 
-def evaluate_output(path: str, output: str) -> str | None:
+def evaluate_output(path: str, output: str) -> Optional[str]:
-    """Evaluates a given output in a flake.
-
-    :param path: path to flake
-    :param output: flake output to be evaluated
-    :returns the .drv path on success or None on failure
-    :raises RuntimeError: evaluation succeeds but no derivation is found
-    """
     logging.info(f"evaluating {output}")
     out = bash_wrapper(f"nix eval {path}#{output}")
     logging.debug(out[0])
@@ -26,6 +17,7 @@ def evaluate_output(path: str, output: str) -> str | None:
     if out[2] != 0:
         logging.warning(f"output {output} did not evaluate correctly")
         return None
+    else:
         drv_match = drv_re.match(out[0])
         if drv_match is None:
             out_msg = "derivation succeeded but output derivation does not contain a derivation"

flupdt/flake_show.py

@@ -1,10 +1,11 @@
-"""Utility to extract flake output info using nix flake (show|check)."""
+#!/usr/bin/env python3
 
 import json
 import logging
 import re
 import shutil
 import typing
+from subprocess import Popen
 
 from flupdt.common import bash_wrapper
 
@@ -15,12 +16,6 @@ output_regexes = [
 
 
 def traverse_json_base(json_dict: dict[str, typing.Any], path: list[str]) -> list[str]:
-    """Crawls through the flake outputs to get nixos-configuration and derivation types.
-
-    :param json_dict: dict of flake outputs to check
-    :param path: a list of outputs constructed so far
-    :returns the output path list, plus any new paths found
-    """
     final_paths = []
     for key, value in json_dict.items():
         if isinstance(value, dict):
@@ -37,21 +32,10 @@ def traverse_json_base(json_dict: dict[str, typing.Any], path: list[str]) -> lis
 
 
 def traverse_json(json_dict: dict) -> list[str]:
-    """Crawls through the flake outputs to get nixos-configuration and derivation types.
-
-    :param json_dict: dict of flake outputs to check
-    :returns a list of outputs that can be evaluated
-    """
     return traverse_json_base(json_dict, [])
 
 
 def get_derivations_from_check(nix_path: str, path_to_flake: str) -> list[str]:
-    """Gets all derivations in a flake, using check instead of show.
-
-    :param nix_path: path to nix binary
-    :param path_to_flake: path to flake to be checked
-    :returns a list of all valid derivations in the flake
-    """
     flake_check = bash_wrapper(f"{nix_path} flake check --verbose --keep-going", path=path_to_flake)
     if flake_check[2] != 0:
         logging.warning(
@@ -71,17 +55,10 @@ def get_derivations_from_check(nix_path: str, path_to_flake: str) -> list[str]:
 
 
 def get_derivations(path_to_flake: str) -> list[str]:
-    """Gets all derivations present in a flake.
-
-    :param path_to_flake: path to flake to be checked
-    :returns a list of all valid derivations in the flake
-    :raises RuntimeError: fails if nix is not present in the PATH
-    """
     nix_path = shutil.which("nix")
     derivations = []
     if nix_path is None:
-        status_msg = "nix is not available in the PATH, please verify that it is installed"
+        raise RuntimeError("nix is not available in the PATH, please verify that it is installed")
-        raise RuntimeError(status_msg)
     flake_show = bash_wrapper(f"{nix_path} flake show --json", path=path_to_flake)
     if flake_show[2] != 0:
         logging.error("flake show returned non-zero exit code")

flupdt/main.py

@@ -1,30 +1,10 @@
 #!/usr/bin/env python3
 
-"""Default processing of flake outputs for evaluating flake updates."""
-
-import logging
-from argparse import Namespace
-
-from flupdt.cli import parse_inputs
-from flupdt.common import configure_logger, partition
-from flupdt.flake_build import build_output
-from flupdt.flake_eval import evaluate_output
 from flupdt.flake_show import get_derivations
-
+from flupdt.cli import parse_inputs
-
+from flupdt.flake_eval import evaluate_output
-def batch_eval(args: Namespace, flake_path: str, derivations: list[str]) -> None:
+from flupdt.common import configure_logger, partition
-    """Bulk run evaluations or builds on a derivation set.
+import logging
-
-    :params args: argument namespace to check against
-    :params flake_path: path to flake to be evaluated
-    :params derivations: list of derivations to run against
-    :returns None
-    """
-    for d in derivations:
-        if args.evaluate:
-            evaluate_output(flake_path, d)
-        if args.build:
-            build_output(flake_path, d)
 
 
 def main() -> None:
@@ -33,20 +13,23 @@ def main() -> None:
     :returns: None
 
     """
-    configure_logger("DEBUG")
+    configure_logger(logging.DEBUG)
     args = parse_inputs()
     flake_path = args.flake_path
     derivations, hydra_jobs = partition(
         lambda s: s.startswith("hydraJobs"), get_derivations(flake_path)
     )
-    derivations, hydra_jobs = list(derivations), list(hydra_jobs)
     logging.info(f"derivations: {list(derivations)}")
-    batch_eval(args, flake_path, derivations)
+    for d in derivations:
+        evaluate_output(flake_path, d)
 
     if not args.keep_hydra:
         logging.info("--keep-hydra flag is not specified, removing Hydra jobs")
     else:
-        batch_eval(args, flake_path, hydra_jobs)
+        hydra_jobs = list(hydra_jobs)
+        logging.info(f"hydraJobs: {hydra_jobs}")
+        for d in hydra_jobs:
+            evaluate_output(flake_path, d)
 
 
 if __name__ == "__main__":