Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051"""One scratch directory per test process, removed when the process exits.
Every Python suite works inside ``<root>/<name>-<pid>``, where root is``$ALBEDO_TEST_TMP`` or ``/tmp/albedo-tests``. Claiming it points tempfile'sdefault directory and ``TMPDIR`` there, so daemons, their kernels and anytemporary directory a test forgets are removed together instead of piling up in/tmp or in each nix shell's own TMPDIR. A run that died before cleaning up isswept by the next one. The root is fixed rather than taken from TMPDIR, solocks under it hold across shells and unix socket paths below it stay short."""
import atexitimport osfrom pathlib import Pathimport shutilimport tempfile
ROOT = Path(os.environ.get("ALBEDO_TEST_TMP", "/tmp/albedo-tests"))
def claim(name: str) -> Path: """This process's scratch directory, created on first use.""" path = ROOT / f"{name}-{os.getpid()}" if tempfile.tempdir == str(path): return path ROOT.mkdir(parents=True, exist_ok=True) sweep() path.mkdir(exist_ok=True) tempfile.tempdir = str(path) os.environ["TMPDIR"] = str(path) atexit.register(shutil.rmtree, path, ignore_errors=True) return path
def sweep() -> None: """Remove the directories of test processes that no longer exist.""" for entry in ROOT.iterdir(): pid = entry.name.rpartition("-")[2] if entry.is_dir() and pid.isdigit() and not alive(int(pid)): shutil.rmtree(entry, ignore_errors=True)
def alive(pid: int) -> bool: try: os.kill(pid, 0) except ProcessLookupError: return False except PermissionError: return True return True