Source code for efrotools.buildlock

# Released under the MIT License. See LICENSE for details.
#
"""A system for sanity testing parallel build isolation."""

from __future__ import annotations

from typing import TYPE_CHECKING
import os


if TYPE_CHECKING:
    from typing import Any

LOCK_DIR_PATH = '.cache/buildlocks'


[docs] class BuildLock: """Tries to ensure a build is not getting stomped on/etc.""" def __init__(self, name: str, projroot: str) -> None: self.name = name self.projroot = projroot if '/' in name or '\\' in name: raise ValueError(f"Illegal BuildLock name: '{name}'.") self.lockpath = os.path.join(self.projroot, LOCK_DIR_PATH, name) def __enter__(self) -> None: lockdir = os.path.dirname(self.lockpath) if not os.path.exists(lockdir): os.makedirs(lockdir, exist_ok=True) # Note: we aren't doing anything super accurate/atomic here. # This is more intended as a gross check to make noise on # clearly broken build logic; it isn't important that it catch # every corner case perfectly. if os.path.exists(self.lockpath): raise RuntimeError( f"Build-lock: lock '{self.name}' exists." ' This probably means multiple builds' ' are running at once that should not be.' ) with open(self.lockpath, 'w', encoding='utf-8') as outfile: outfile.write('') def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> Any: if not os.path.exists(self.lockpath): raise RuntimeError( f"Build-lock: lock '{self.name}' not found at tear-down." ) os.unlink(self.lockpath)
# Docs-generation hack; import some stuff that we likely only forward-declared # in our actual source code so that docs tools can find it. from typing import (Coroutine, Any, Literal, Callable, Generator, Awaitable, Sequence, Self) import asyncio from concurrent.futures import Future from pathlib import Path from enum import Enum