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