lib

Reusable functions for parsing @nix protocol and running nix commands.

_current_nix_system() cached

Return the Nix system identifier for the current host machine.

Delegates to nix eval --impure --json --expr builtins.currentSystem. The result is cached because it never changes during a process lifetime.

Returns:

Type	Description
str	Nix system string (e.g. "x86_64-linux").

Source code in src/junix/lib.py

@cache
def _current_nix_system() -> str:
    """Return the Nix system identifier for the current host machine.

    Delegates to `nix eval --impure --json --expr builtins.currentSystem`.
    The result is cached because it never changes during a process lifetime.

    Returns:
        Nix system string (e.g. `"x86_64-linux"`).
    """
    return json.loads(
        local["nix"]("eval", "--impure", "--json", "--expr", "builtins.currentSystem")
    )

_ensure_builds(builds, process_successful, default_name='nix build', expected_count=0)

Ensure the build list has at least expected_count entries.

Nix only emits actBuild protocol activities when it actually compiles something. Cached/substituted builds produce no actBuild events, so the build list may be shorter than the number of attributes requested. This helper pads with synthetic skipped builds so the JUnit report reflects every requested attribute.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	Collected build events from the protocol handler.	required
process_successful	bool	Whether the overall Nix process succeeded.	required
default_name	str	Name to use for the synthetic build when no builds exist and no expected_count is given.	'nix build'
expected_count	int	Minimum number of builds to return. When 0 (default), at least one build is guaranteed.	0

Returns:

Type	Description
list[BuildEvent]	Build list padded to at least expected_count entries.

Source code in src/junix/lib.py

def _ensure_builds(
    builds: list[BuildEvent],
    process_successful: bool,
    default_name: str = "nix build",
    expected_count: int = 0,
) -> list[BuildEvent]:
    """Ensure the build list has at least ``expected_count`` entries.

    Nix only emits ``actBuild`` protocol activities when it actually
    compiles something.  Cached/substituted builds produce no ``actBuild``
    events, so the build list may be shorter than the number of
    attributes requested.  This helper pads with synthetic skipped
    builds so the JUnit report reflects every requested attribute.

    Args:
        builds: Collected build events from the protocol handler.
        process_successful: Whether the overall Nix process succeeded.
        default_name: Name to use for the synthetic build when no
            builds exist and no expected_count is given.
        expected_count: Minimum number of builds to return.  When 0
            (default), at least one build is guaranteed.

    Returns:
        Build list padded to at least ``expected_count`` entries.
    """
    if not builds:
        builds = [
            BuildEvent(
                name=default_name,
                drv_path=default_name,
                started=True,
                stopped=True,
                success=process_successful,
                cached=True,
            )
        ]
    while len(builds) < expected_count:
        builds.append(
            BuildEvent(
                name=f"{default_name} #{len(builds) + 1}",
                drv_path=default_name,
                started=True,
                stopped=True,
                success=True,
                cached=True,
            )
        )
    return builds

_normalize_flake_path(path)

Normalize a flake path so Nix can resolve it.

Nix requires relative filesystem paths to start with ./ or /. Bare paths like tests/sample-flake are not recognised as flake references. This helper prepends ./ when the path looks like a relative filesystem path (no scheme, not already prefixed).

Parameters:

Name	Type	Description	Default
path	str	Raw flake path from the user (e.g. ".", "tests/sample-flake").	required

Returns:

Type	Description
str	Normalised path suitable for nix flake show and flake URIs.

Source code in src/junix/lib.py

def _normalize_flake_path(path: str) -> str:
    """Normalize a flake path so Nix can resolve it.

    Nix requires relative filesystem paths to start with ``./`` or ``/``.
    Bare paths like ``tests/sample-flake`` are not recognised as flake
    references.  This helper prepends ``./`` when the path looks like a
    relative filesystem path (no scheme, not already prefixed).

    Args:
        path: Raw flake path from the user (e.g. ``"."``, ``"tests/sample-flake"``).

    Returns:
        Normalised path suitable for ``nix flake show`` and flake URIs.
    """
    if path in (".", ".."):
        return path
    if path.startswith(("./", "/")):
        return path
    if ":" in path:
        return path
    return f"./{path}"

_overall_success(builds)

Return True when every build completed successfully.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	List of build events to check.	required

Returns:

Type	Description
bool	True if all builds stopped and succeeded.

Source code in src/junix/lib.py

def _overall_success(builds: list[BuildEvent]) -> bool:
    """Return True when every build completed successfully.

    Args:
        builds: List of build events to check.

    Returns:
        True if all builds stopped and succeeded.
    """
    return all(b.stopped and b.success for b in builds)

_parse_nix_stderr(stderr, handler, log_level=0)

Feed stderr lines to handler, forwarding non-@nix lines to stderr.

Parameters:

Name	Type	Description	Default
stderr	str	Raw stderr text from a nix invocation.	required
handler	NixEventHandler	The event handler to feed @nix lines to.	required
log_level	int	Maximum Nix log level to forward to stderr (default 0 = errors only; higher = more verbose).	0

Source code in src/junix/lib.py

def _parse_nix_stderr(
    stderr: str,
    handler: NixEventHandler,
    log_level: int = 0,
) -> None:
    """Feed stderr lines to handler, forwarding non-`@nix` lines to stderr.

    Args:
        stderr: Raw stderr text from a nix invocation.
        handler: The event handler to feed `@nix` lines to.
        log_level: Maximum Nix log level to forward to stderr
            (default 0 = errors only; higher = more verbose).
    """
    for line in stderr.splitlines(keepends=True):
        line = line.rstrip("\n").rstrip("\r")
        if not handler.handle_line(line):
            print(line, file=sys.stderr)

    for msg in handler.get_messages(threshold=log_level):
        print(msg, file=sys.stderr)

_setup_colors()

Configure ANSI color output based on stderr and NO_COLOR.

Disables color when stderr is not a TTY or when the NO_COLOR environment variable is set (per https://no-color.org/).

Source code in src/junix/lib.py

def _setup_colors() -> None:
    """Configure ANSI color output based on stderr and NO_COLOR.

    Disables color when stderr is not a TTY or when the ``NO_COLOR``
    environment variable is set (per https://no-color.org/).
    """
    if not sys.stderr.isatty() or "NO_COLOR" in os.environ:
        ANSIStyle.use_color = 0

_write_report(xml, output_path)

Write the JUnit XML to output_path or stdout.

Parameters:

Name	Type	Description	Default
xml	str	The JUnit XML string to write.	required
output_path	str | None	Path to write to, or None for stdout.	required

Source code in src/junix/lib.py

def _write_report(xml: str, output_path: str | None) -> None:
    """Write the JUnit XML to output_path or stdout.

    Args:
        xml: The JUnit XML string to write.
        output_path: Path to write to, or None for stdout.
    """
    if output_path:
        with open(output_path, "w") as f:
            f.write(xml)
    else:
        sys.stdout.write(xml)

discover_checks(path, eval_arch=None) async

Discover flake check attribute URIs for path.

Uses nix flake show --json and filters by eval_arch (default: current system only).

Parameters:

Name	Type	Description	Default
path	str	Flake path (e.g. "." or "github:owner/repo").	required
eval_arch	list[str] | None	Architectures to evaluate (default: current system).	None

Returns:

Type	Description
list[str]	List of flake attribute URIs (e.g. [".#checks.x86_64-linux.test-a"]).

Source code in src/junix/lib.py

async def discover_checks(
    path: str,
    eval_arch: list[str] | None = None,
) -> list[str]:
    """Discover flake check attribute URIs for path.

    Uses `nix flake show --json` and filters by eval_arch
    (default: current system only).

    Args:
        path: Flake path (e.g. ``"."`` or ``"github:owner/repo"``).
        eval_arch: Architectures to evaluate (default: current system).

    Returns:
        List of flake attribute URIs (e.g. ``[".#checks.x86_64-linux.test-a"]``).
    """
    path = _normalize_flake_path(path)
    result = await _nix.run(["flake", "show", path, "--json"], retcode=None)

    if result.returncode != 0:
        return []

    flake = json.loads(result.stdout)
    checks = flake.get("checks", {})

    eval_systems: set[str] = set(eval_arch) if eval_arch else {_current_nix_system()}

    check_attrs: list[str] = []
    for system, sys_checks in checks.items():
        if system not in eval_systems:
            continue
        check_attrs.extend(
            f"{path}#checks.{system}.{check_name}" for check_name in sys_checks
        )

    return check_attrs

print_summary(builds, downloaded=0)

Print a human-readable, color-coded summary of build results to stderr.

Output includes the number of paths downloaded from cache, cached builds, passed builds, and failed builds. On failure the names of failing builds are listed individually.

Parameters:

Name	Type	Description	Default
builds	list[BuildEvent]	List of build events to summarize.	required
downloaded	int	Number of paths downloaded from binary caches.	0

Source code in src/junix/lib.py

def print_summary(
    builds: list[BuildEvent],
    downloaded: int = 0,
) -> None:
    """Print a human-readable, color-coded summary of build results to stderr.

    Output includes the number of paths downloaded from cache, cached
    builds, passed builds, and failed builds.  On failure the names of
    failing builds are listed individually.

    Args:
        builds: List of build events to summarize.
        downloaded: Number of paths downloaded from binary caches.
    """
    _setup_colors()

    successful = [b for b in builds if b.success and b.stopped]
    failed = [b for b in builds if not b.success]
    cached = [b for b in builds if b.cached]

    print(file=sys.stderr)

    if downloaded:
        print(
            str(colors.cyan | f"  \u2b07 {downloaded} path(s) downloaded"),
            file=sys.stderr,
        )

    if cached:
        print(
            str(colors.yellow | f"  \u007e {len(cached)} build(s) cached"),
            file=sys.stderr,
        )

    for b in successful:
        if not b.cached:
            print(str(colors.green | f"  \u2713 {b.name}"), file=sys.stderr)

    for b in failed:
        print(str(colors.red | f"  \u2717 {b.name}"), file=sys.stderr)

    total = len(builds)
    passed = len(successful)
    failed_count = len(failed)

    if failed_count:
        print(
            str(colors.bold & colors.red | f"\n  FAILED ({failed_count}/{total}):"),
            file=sys.stderr,
        )
        for b in failed:
            print(
                str(colors.red | f"    - {b.name}"),
                file=sys.stderr,
            )
    else:
        print(
            str(colors.bold & colors.green | f"\n  PASSED ({passed}/{total})"),
            file=sys.stderr,
        )

run_nix_build(attrs, suite_name='nix', store=None, default_name='nix build', log_level=0, expected_count=0) async

Run nix build and produce JUnit XML from its @nix protocol output.

Uses --keep-going so Nix attempts every attribute even if some fail. Cached/substituted builds produce no actBuild protocol events; when expected_count is set, the report is padded with synthetic skipped entries so every requested attribute is accounted for.

Parameters:

Name	Type	Description	Default
attrs	list[str]	Flake attribute paths to build (e.g. ["nixpkgs#hello"]).	required
suite_name	str	Value for the <testsuite name="..."> attribute.	'nix'
store	str | None	Remote store URL (e.g. "ssh-ng://eu.nixbuild.net").	None
default_name	str	Name for the synthetic build when no actBuild activities were received.	'nix build'
log_level	int	Maximum Nix log level to forward to stderr (default 0 = errors only; higher = more verbose).	0
expected_count	int	Minimum number of builds to report. When Nix caches some builds they produce no protocol events, so this pads the report with synthetic skipped entries.	0

Returns:

Type	Description
tuple[str, int]	Tuple of (xml_string, exit_code).

Source code in src/junix/lib.py

async def run_nix_build(
    attrs: list[str],
    suite_name: str = "nix",
    store: str | None = None,
    default_name: str = "nix build",
    log_level: int = 0,
    expected_count: int = 0,
) -> tuple[str, int]:
    """Run nix build and produce JUnit XML from its `@nix` protocol output.

    Uses ``--keep-going`` so Nix attempts every attribute even if some
    fail.  Cached/substituted builds produce no ``actBuild`` protocol
    events; when ``expected_count`` is set, the report is padded with
    synthetic skipped entries so every requested attribute is accounted
    for.

    Args:
        attrs: Flake attribute paths to build (e.g. ``["nixpkgs#hello"]``).
        suite_name: Value for the ``<testsuite name="...">`` attribute.
        store: Remote store URL (e.g. ``"ssh-ng://eu.nixbuild.net"``).
        default_name: Name for the synthetic build when no ``actBuild``
            activities were received.
        log_level: Maximum Nix log level to forward to stderr
            (default 0 = errors only; higher = more verbose).
        expected_count: Minimum number of builds to report.  When Nix
            caches some builds they produce no protocol events, so this
            pads the report with synthetic skipped entries.

    Returns:
        Tuple of (xml_string, exit_code).
    """
    handler = NixEventHandler()

    cmd = ["build", "--log-format", "internal-json", "--keep-going"]
    if store:
        cmd.extend(["--store", store])
    cmd.extend(attrs)

    result = await _nix.run(cmd, retcode=None)
    nix_success = result.returncode == 0

    _parse_nix_stderr(result.stderr, handler, log_level=log_level)

    handler.finalize(process_successful=nix_success)
    builds = _ensure_builds(
        handler.builds,
        process_successful=nix_success,
        default_name=default_name,
        expected_count=expected_count,
    )
    print_summary(builds, downloaded=handler.downloaded_count)
    xml = report_to_xml(
        builds,
        suite_name=suite_name,
        error_messages=handler._errors or None,
    )
    exit_code = 0 if nix_success and _overall_success(builds) else 1
    return xml, exit_code

translate_stream(stream, suite_name='nix', process_successful=False, default_name='nix build', log_level=0)

Read @nix JSON lines from stream and produce JUnit XML.

Non-@nix lines and Nix log messages are forwarded to stderr.

Parameters:

Name	Type	Description	Default
stream	IOBase	Text stream to read (e.g. sys.stdin).	required
suite_name	str	Value for the <testsuite name="..."> attribute.	'nix'
process_successful	bool	Passed to NixEventHandler.finalize.	False
default_name	str	Name for the synthetic build when no actBuild activities were received.	'nix build'
log_level	int	Maximum Nix log level to forward to stderr (default 0 = errors only; higher = more verbose).	0

Returns:

Type	Description
tuple[str, int]	Tuple of (xml_string, exit_code).

Source code in src/junix/lib.py

def translate_stream(
    stream: IOBase,
    suite_name: str = "nix",
    process_successful: bool = False,
    default_name: str = "nix build",
    log_level: int = 0,
) -> tuple[str, int]:
    """Read `@nix` JSON lines from stream and produce JUnit XML.

    Non-`@nix` lines and Nix log messages are forwarded to stderr.

    Args:
        stream: Text stream to read (e.g. sys.stdin).
        suite_name: Value for the `<testsuite name="...">` attribute.
        process_successful: Passed to NixEventHandler.finalize.
        default_name: Name for the synthetic build when no `actBuild`
            activities were received.
        log_level: Maximum Nix log level to forward to stderr
            (default 0 = errors only; higher = more verbose).

    Returns:
        Tuple of (xml_string, exit_code).
    """
    handler = NixEventHandler()
    non_nix = handler.handle_stream(stream)

    for line in non_nix:
        print(line, end="", file=sys.stderr)

    for msg in handler.get_messages(threshold=log_level):
        print(msg, file=sys.stderr)

    handler.finalize(process_successful=process_successful)
    builds = _ensure_builds(
        handler.builds,
        process_successful=not handler._errors,
        default_name=default_name,
    )
    print_summary(builds, downloaded=handler.downloaded_count)
    xml = report_to_xml(
        builds,
        suite_name=suite_name,
        error_messages=handler._errors or None,
    )
    exit_code = 0 if _overall_success(builds) else 1
    return xml, exit_code