computer-playbook/utils/env/runtime.py

"""Runtime-context lookups for the .env generator.

Each helper resolves a single piece of information from the host the
generator runs on (disk free space, RAM, hostname, /proc/version,
GitHub Actions env vars, ...). They are intentionally tiny and
side-effect-free apart from the OS read they need to perform, so they
can be exercised independently in tests.
"""

from __future__ import annotations

import os
import re
import socket
import subprocess
from pathlib import Path


def is_user_writable(path: str) -> bool:
    """Return True when `path` (or its closest existing ancestor) can
    be written to by the current user without privilege escalation.

    Used to decide between a system path (e.g. ``/opt/venvs``) and a
    user-local fallback (``$HOME/.venvs``) when the system path is not
    yet prepared on a fresh dev machine."""
    p = Path(path)
    while not p.exists() and str(p) not in ("/", "."):
        p = p.parent
    return os.access(p, os.W_OK)


def df_avail_gb(path: str) -> int:
    """Return free space in GB at `path` (or the closest existing
    ancestor). 0 when `df` fails or the output is unparseable."""
    p = Path(path)
    while not p.exists() and str(p) not in ("/", "."):
        p = p.parent
    try:
        out = subprocess.check_output(
            ["df", "--output=avail", "-B1G", str(p)],
            text=True,
            stderr=subprocess.DEVNULL,
        )
    except (subprocess.CalledProcessError, OSError):
        return 0
    lines = out.strip().splitlines()
    if len(lines) < 2:
        return 0
    try:
        return int(lines[1].strip())
    except ValueError:
        return 0


def _meminfo_mb(field: str) -> int:
    """`field` from /proc/meminfo in MB. 0 on failure."""
    try:
        text = Path(
            "/proc/meminfo"
        ).read_text()  # nocheck: cache-read -- live kernel pseudo-file
    except OSError:
        return 0
    for line in text.splitlines():
        if line.startswith(field):
            try:
                kb = int(line.split()[1])
                return kb // 1024
            except (IndexError, ValueError):
                return 0
    return 0


def mem_available_mb() -> int:
    """`MemAvailable` from /proc/meminfo in MB. 0 on failure."""
    return _meminfo_mb("MemAvailable:")


def mem_total_mb() -> int:
    """`MemTotal` from /proc/meminfo in MB. 0 on failure."""
    return _meminfo_mb("MemTotal:")


def mem_stall_pct() -> float:
    """`full avg60` from /proc/pressure/memory in percent. 0.0 on failure."""
    try:
        text = Path(
            "/proc/pressure/memory"
        ).read_text()  # nocheck: cache-read -- live kernel pseudo-file
    except OSError:
        return 0.0
    for line in text.splitlines():
        if not line.startswith("full "):
            continue
        for field in line.split():
            if field.startswith("avg60="):
                try:
                    return float(field.removeprefix("avg60="))
                except ValueError:
                    return 0.0
    return 0.0


def is_wsl2() -> bool:
    """True when /proc/version mentions microsoft or wsl. False otherwise."""
    try:
        text = Path(
            "/proc/version"
        ).read_text()  # nocheck: cache-read -- live kernel pseudo-file
    except OSError:
        return False
    return bool(re.search(r"microsoft|wsl", text, re.IGNORECASE))


def detect_gha_act() -> tuple[bool, bool]:
    """Return (running_on_github, running_on_act).

    Mirrors the legacy scripts/meta/env/runtime.sh logic:
      - GITHUB_ACTIONS=true AND ACT=true  -> act (local nektos act)
      - GITHUB_ACTIONS=true AND not ACT   -> real github actions
    """
    gha = os.environ.get("GITHUB_ACTIONS") == "true"
    act = gha and os.environ.get("ACT") == "true"
    return (gha and not act, act)


def hostname() -> str:
    """`socket.gethostname()` with a `local` fallback for sandboxed hosts
    that fail the syscall."""
    try:
        name = socket.gethostname()
    except OSError:
        return "local"
    return name or "local"


def run_helper(
    cmd: list[str],
    cwd: Path,
    extra_env: dict[str, str] | None = None,
) -> str:
    """Run an existing shell helper, return stripped stdout. Empty
    string on failure. Helpers run with `cwd` as their working
    directory."""
    env = {**os.environ, **(extra_env or {})}
    try:
        out = subprocess.check_output(
            cmd, cwd=str(cwd), env=env, text=True, stderr=subprocess.DEVNULL
        )
        return out.strip()
    except (subprocess.CalledProcessError, OSError, FileNotFoundError):
        return ""