From fad41f46775a8b7c6bc9fe271f144485347df241 Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:23:48 -0700 Subject: [PATCH 1/2] Add adversarial regression and performance suites --- .github/workflows/tests.yml | 1 + docs/performance.md | 33 +++ docs/releasing.md | 5 +- pyproject.toml | 1 + scripts/benchmark_runtime.py | 114 ++++++++ tests/test_adversarial_regressions.py | 390 ++++++++++++++++++++++++++ tests/validate.sh | 1 + 7 files changed, 544 insertions(+), 1 deletion(-) create mode 100644 docs/performance.md create mode 100644 scripts/benchmark_runtime.py create mode 100644 tests/test_adversarial_regressions.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 2cb3a05..a096968 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -62,6 +62,7 @@ jobs: run: | python -m mypy --strict examples/typed_consumer.py python scripts/validate_docs.py + python scripts/benchmark_runtime.py --check python -m compileall -q examples - name: Run tests with coverage threshold run: python -m pytest --cov=base_cli --cov-report=term-missing --cov-fail-under=80 diff --git a/docs/performance.md b/docs/performance.md new file mode 100644 index 0000000..ca41e1d --- /dev/null +++ b/docs/performance.md @@ -0,0 +1,33 @@ +# Performance and adversarial-regression contract + +`base-cli` treats startup and filesystem behavior as part of its public +quality contract. The checked benchmark is intentionally small and runs from +the source checkout: + +```bash +python scripts/benchmark_runtime.py --check +``` + +It records fresh-process import time and the cost of an isolated production +invocation through `base_cli.testing.invoke`. The CI quality job checks the +sample p95 against these budgets: + +| Measurement | Budget | +| --- | ---: | +| Fresh `import base_cli` | 750 ms | +| Isolated invocation and runtime filesystem setup | 1,500 ms | + +The benchmark reports the median, p95, and maximum for seven samples. These +budgets are intentionally broad enough for hosted runners while still +detecting accidental quadratic startup work, unbounded metadata scans, or +unexpected dependency imports. A performance improvement should preserve the +same lifecycle and persistence assertions covered by the adversarial tests. + +The regression suite uses deterministic Hypothesis examples (`derandomize` +enabled), fixed multiprocessing workloads, and explicit seed values in every +worker payload. Property cases cover redaction and command-protocol framing; +spawned processes cover history append, private metadata replacement, logging, +extension discovery caches, and run-bundle retention. Ctrl+C is tested through +both the lifecycle boundary and a real POSIX subprocess signal. Windows keeps +the portable lifecycle and persistence checks while skipping only assertions +that require POSIX signal or descriptor semantics. diff --git a/docs/releasing.md b/docs/releasing.md index 4f4b3fd..819fc79 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -23,7 +23,10 @@ source tree on `sys.path`. Tests run across Python 3.10 through 3.14 on Linux, macOS, and Windows, with Debian, Fedora, and WSL validation retained. Blocking quality gates cover Ruff formatting/lint, strict public-sample typing, an 80% branch-coverage threshold, documentation/example checks, and dependency/static -security scans. +security scans. The [performance contract](performance.md) also checks fresh +import and isolated invocation budgets, while the adversarial suite exercises +redaction, protocol framing, persistence, concurrency, retention, and signal +cleanup. The publish job downloads that same reviewed artifact; it does not rebuild during publication. diff --git a/pyproject.toml b/pyproject.toml index 386d301..b94029b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dependencies = [ [project.optional-dependencies] dev = [ "build>=1.2", + "hypothesis>=6.100,<7", "mypy>=1.17,<2", "pytest>=8.0", ] diff --git a/scripts/benchmark_runtime.py b/scripts/benchmark_runtime.py new file mode 100644 index 0000000..46a671d --- /dev/null +++ b/scripts/benchmark_runtime.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Track import and isolated invocation costs with stable, checked budgets.""" + +from __future__ import annotations + +import argparse +import os +import statistics +import subprocess +import sys +import tempfile +import time +from pathlib import Path + + +IMPORT_P95_BUDGET_MS = 750.0 +INVOCATION_P95_BUDGET_MS = 1_500.0 +DEFAULT_ITERATIONS = 7 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--iterations", + type=int, + default=DEFAULT_ITERATIONS, + help=f"number of samples per benchmark (default: {DEFAULT_ITERATIONS})", + ) + parser.add_argument( + "--check", + action="store_true", + help="fail when the documented p95 budgets are exceeded", + ) + args = parser.parse_args() + if args.iterations < 3: + parser.error("--iterations must be at least 3") + + import_samples = _measure_import(args.iterations) + invocation_samples = _measure_invocations(args.iterations) + metrics = { + "import_ms": _summary(import_samples), + "invocation_ms": _summary(invocation_samples), + } + print("import_ms: median={median:.2f} p95={p95:.2f} max={maximum:.2f}".format(**metrics["import_ms"])) + print("invocation_ms: median={median:.2f} p95={p95:.2f} max={maximum:.2f}".format(**metrics["invocation_ms"])) + + if not args.check: + return 0 + failures = [] + if metrics["import_ms"]["p95"] > IMPORT_P95_BUDGET_MS: + failures.append(f"import p95 exceeded {IMPORT_P95_BUDGET_MS:.0f} ms") + if metrics["invocation_ms"]["p95"] > INVOCATION_P95_BUDGET_MS: + failures.append(f"invocation p95 exceeded {INVOCATION_P95_BUDGET_MS:.0f} ms") + if failures: + print("Performance budget failure: " + "; ".join(failures), file=sys.stderr) + return 1 + return 0 + + +def _measure_import(iterations: int) -> list[float]: + package_root = Path(__file__).resolve().parents[1] / "lib" / "python" + environment = dict(os.environ) + existing_path = environment.get("PYTHONPATH") + environment["PYTHONPATH"] = f"{package_root}{os.pathsep}{existing_path}" if existing_path else str(package_root) + samples: list[float] = [] + for _ in range(iterations): + started = time.perf_counter_ns() + subprocess.run( + [sys.executable, "-c", "import base_cli"], + check=True, + env=environment, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + samples.append(_elapsed_ms(started)) + return samples + + +def _measure_invocations(iterations: int) -> list[float]: + import base_cli + from base_cli.testing import invoke + + app = base_cli.App(name="benchmark-runtime") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + samples: list[float] = [] + with tempfile.TemporaryDirectory(prefix="base-cli-benchmark-") as tmpdir: + home = Path(tmpdir) + for _ in range(iterations): + started = time.perf_counter_ns() + result = invoke(app, [], home=home) + if result.exit_code != 0: + raise RuntimeError(f"benchmark invocation failed: {result.output}") + samples.append(_elapsed_ms(started)) + return samples + + +def _elapsed_ms(started_ns: int) -> float: + return (time.perf_counter_ns() - started_ns) / 1_000_000 + + +def _summary(samples: list[float]) -> dict[str, float]: + return { + "median": statistics.median(samples), + "p95": max(samples), + "maximum": max(samples), + } + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_adversarial_regressions.py b/tests/test_adversarial_regressions.py new file mode 100644 index 0000000..c2c658c --- /dev/null +++ b/tests/test_adversarial_regressions.py @@ -0,0 +1,390 @@ +from __future__ import annotations + +import io +import json +import multiprocessing as multiprocessing_module +import os +import signal +import subprocess +import sys +import tempfile +import textwrap +import time +import unittest +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Lock +from types import SimpleNamespace + +import base_cli +from hypothesis import given, settings, strategies as st + +from base_cli._private_files import write_private_json +from base_cli._runtime import prune_run_bundles +from base_cli.command_protocol import ( + BOOLEAN, + CommandProtocolError, + CommandSchemaRegistry, + NULLABLE_STRING, + STRING, + dumps_records, + loads_records, +) +from base_cli.history import append_history_line +from base_cli.redaction import REDACTED, redact_argv +from base_cli.testing import invoke + + +SEEDS = (17, 29, 41, 53) +_SAFE_TEXT = st.text( + alphabet=st.characters(blacklist_categories=("Cs",), blacklist_characters="\0"), + max_size=80, +) + + +@st.composite +def _record_strategy(draw: st.DrawFn) -> dict[str, str | bool | None]: + return { + "name": draw(_SAFE_TEXT), + "enabled": draw(st.booleans()), + "note": draw(st.one_of(st.none(), _SAFE_TEXT)), + "command": draw(_SAFE_TEXT), + } + + +def _append_history_worker(path_text: str, seed: int, count: int) -> None: + path = Path(path_text) + for ordinal in range(count): + payload = json.dumps({"seed": seed, "ordinal": ordinal}, sort_keys=True) + append_history_line(path, payload + "\n") + + +def _write_metadata_worker(path_text: str, seed: int, count: int) -> None: + path = Path(path_text) + for ordinal in range(count): + write_private_json(path, {"seed": seed, "ordinal": ordinal, "status": "ok"}) + + +def _write_log_worker(path_text: str, seed: int, count: int) -> None: + from base_cli.logging import configure_logger + + stream = io.StringIO() + logger = configure_logger( + f"adversarial-{seed}", + Path(path_text), + debug=True, + quiet=True, + stream=stream, + run_id=f"seed-{seed}", + ) + for ordinal in range(count): + logger.info("seed=%s ordinal=%s", seed, ordinal) + for handler in list(logger.handlers): + handler.flush() + handler.close() + logger.removeHandler(handler) + + +def _prune_worker(runs_root_text: str) -> None: + prune_run_bundles( + Path(runs_root_text), + policy=base_cli.RetentionPolicy(max_bundles=2), + ) + + +def _run_processes(target: object, args_list: list[tuple[object, ...]]) -> None: + context = multiprocessing_module.get_context("spawn") + processes = [context.Process(target=target, args=args) for args in args_list] # type: ignore[arg-type] + try: + for process in processes: + process.start() + for process in processes: + process.join(30) + if process.is_alive(): + process.terminate() + process.join(5) + raise AssertionError("adversarial worker exceeded its 30-second budget") + if process.exitcode != 0: + raise AssertionError(f"adversarial worker exited with {process.exitcode}") + finally: + for process in processes: + if process.is_alive(): + process.terminate() + process.join(5) + + +class ProtocolAndRedactionPropertyTests(unittest.TestCase): + @settings(max_examples=80, deadline=None, derandomize=True) + @given(st.lists(_record_strategy(), max_size=5)) + def test_protocol_round_trip_is_total_for_generated_records( + self, + records: list[dict[str, str | bool | None]], + ) -> None: + registry = CommandSchemaRegistry() + registry.register( + "fuzz-record", + {"name": STRING, "enabled": BOOLEAN, "note": NULLABLE_STRING, "command": STRING}, + ) + + payload = dumps_records("fuzz-record", records, registry=registry) + + self.assertEqual( + loads_records(payload, expected_record_type="fuzz-record", registry=registry), + ("fuzz-record", tuple(records)), + ) + + @settings(max_examples=120, deadline=None, derandomize=True) + @given(st.text(alphabet=st.characters(blacklist_categories=("Cs",)), max_size=256)) + def test_protocol_fuzzer_rejects_or_decodes_without_leaking_unexpected_errors(self, payload: str) -> None: + registry = CommandSchemaRegistry() + registry.register("fuzz-record", {"name": STRING}) + + try: + loads_records(payload, registry=registry) + except CommandProtocolError: + return + + @settings(max_examples=100, deadline=None, derandomize=True) + @given( + secret=_SAFE_TEXT.filter(lambda value: value not in {"", REDACTED}), + visible=_SAFE_TEXT, + ) + def test_redaction_property_preserves_argv_shape_and_hides_generated_secrets( + self, + secret: str, + visible: str, + ) -> None: + argv = ["tool", "--token", secret, "--label", visible] + + redacted = redact_argv(argv, {"token"}) + + self.assertEqual(len(redacted), len(argv)) + self.assertEqual(redacted[2], REDACTED) + self.assertNotEqual(redacted[2], secret) + + +class MultiprocessingRegressionTests(unittest.TestCase): + def test_history_appends_remain_complete_across_spawned_processes(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "history.jsonl" + _run_processes( + _append_history_worker, + [(str(path), seed, 16) for seed in SEEDS], + ) + + records = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + + self.assertEqual(len(records), len(SEEDS) * 16) + self.assertEqual( + {(record["seed"], record["ordinal"]) for record in records}, + {(seed, ordinal) for seed in SEEDS for ordinal in range(16)}, + ) + + def test_atomic_metadata_writers_leave_one_valid_snapshot(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "run.json" + _run_processes( + _write_metadata_worker, + [(str(path), seed, 20) for seed in SEEDS], + ) + + payload = json.loads(path.read_text(encoding="utf-8")) + temporary_files = tuple(path.parent.glob(f".{path.name}.*.tmp")) + + self.assertEqual(payload["status"], "ok") + self.assertIn(payload["seed"], SEEDS) + self.assertIn(payload["ordinal"], range(20)) + self.assertEqual(temporary_files, ()) + + def test_log_file_writers_do_not_leave_partial_records(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "parallel.log" + _run_processes( + _write_log_worker, + [(str(path), seed, 16) for seed in SEEDS], + ) + + lines = path.read_text(encoding="utf-8").splitlines() + + self.assertEqual(len(lines), len(SEEDS) * 16) + self.assertTrue(all(" seed=" in line and " ordinal=" in line for line in lines)) + + def test_extension_discovery_metadata_cache_is_thread_safe_and_single_snapshot(self) -> None: + entry_point = SimpleNamespace( + group=base_cli.COMMAND_ENTRY_POINT_GROUP, + name="adversarial", + value="sample:register", + extras=(), + dist=SimpleNamespace(name="sample", version="1.0"), + ) + calls = 0 + calls_lock = Lock() + + def provider() -> tuple[SimpleNamespace]: + nonlocal calls + with calls_lock: + calls += 1 + return (entry_point,) + + discovery = base_cli.ExtensionDiscovery(entry_points=provider) + with ThreadPoolExecutor(max_workers=8) as executor: + snapshots = list(executor.map(lambda _index: discovery.list_commands(), range(32))) + + self.assertEqual(calls, 1) + self.assertTrue(all(snapshot == snapshots[0] for snapshot in snapshots)) + self.assertEqual(snapshots[0][0].key, "base_cli.commands:adversarial") + + @unittest.skipUnless(os.name != "nt", "POSIX retention locking is covered on Windows through single-process tests") + def test_run_bundle_retention_remains_bounded_across_processes(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + runs_root = Path(tmpdir) / "runs" + runs_root.mkdir() + for index in range(8): + bundle = runs_root / f"run-{index}" + (bundle / "logs").mkdir(parents=True) + write_private_json( + bundle / "run.json", + { + "run_id": bundle.name, + "status": "ok", + "started_at": f"2020-01-{index + 1:02d}T00:00:00Z", + "preserve": False, + }, + ) + + _run_processes(_prune_worker, [(str(runs_root),) for _seed in SEEDS]) + + bundles = [path for path in runs_root.iterdir() if path.is_dir() and not path.name.startswith(".")] + index = json.loads((runs_root / ".base-cli-run-index.json").read_text(encoding="utf-8")) + + self.assertLessEqual(len(bundles), 2) + self.assertEqual(index["version"], 1) + self.assertLessEqual(len(index["bundles"]), 2) + + +class SignalRegressionTests(unittest.TestCase): + def test_keyboard_interrupt_finishes_lifecycle_without_leaked_context(self) -> None: + app = base_cli.App(name="adversarial-interrupt") + seen: dict[str, object] = {} + + @app.command() + def main(ctx: base_cli.Context) -> None: + seen["temp_dir"] = ctx.temp_dir + seen["logger"] = ctx.log + raise KeyboardInterrupt() + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + result = invoke(app, [], home=home) + metadata_paths = tuple((home / ".cache").rglob("run.json")) + payload = json.loads(metadata_paths[0].read_text(encoding="utf-8")) + temp_dir = Path(seen["temp_dir"]) + temp_contents = tuple(temp_dir.iterdir()) if temp_dir.is_dir() else () + logger_handlers = list(seen["logger"].handlers) # type: ignore[union-attr] + + self.assertEqual(result.exit_code, base_cli.ExitCode.INTERRUPTED) + self.assertIn("Interrupted.", result.stderr) + self.assertEqual(len(metadata_paths), 1) + self.assertEqual(payload["outcome"], "interrupted") + self.assertEqual(payload["status"], "error") + self.assertEqual(temp_contents, ()) + self.assertEqual(logger_handlers, []) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_click_abort_is_recorded_as_a_terminal_aborted_outcome(self) -> None: + import click + + app = base_cli.App(name="adversarial-abort") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise click.Abort() + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + result = invoke(app, [], home=home) + metadata_paths = tuple((home / ".cache").rglob("run.json")) + payload = json.loads(metadata_paths[0].read_text(encoding="utf-8")) + + self.assertEqual(result.exit_code, base_cli.ExitCode.FAILURE) + self.assertIn("Aborted!", result.stderr) + self.assertEqual(payload["outcome"], "aborted") + self.assertEqual(payload["status"], "error") + + @unittest.skipUnless(os.name != "nt", "SIGINT subprocess semantics are platform-specific") + def test_real_sigint_returns_130_and_persists_interrupted_metadata(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache = home / ".cache" + ready = root / "ready" + script = textwrap.dedent( + """ + import os + import time + from pathlib import Path + + import base_cli + + app = base_cli.App(name="signal-child") + ready = Path(os.environ["BASE_CLI_READY"]) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + ready.write_text("ready", encoding="utf-8") + while True: + time.sleep(1) + + raise SystemExit(base_cli.run_app(app, [])) + """ + ) + environment = dict(os.environ) + package_root = Path(__file__).resolve().parents[1] / "lib" / "python" + environment["PYTHONPATH"] = f"{package_root}{os.pathsep}{environment.get('PYTHONPATH', '')}" + environment.update( + { + "HOME": str(home), + "BASE_CLI_CACHE_DIR": str(cache), + "BASE_CLI_READY": str(ready), + } + ) + process = subprocess.Popen( + [sys.executable, "-c", script], + cwd=Path(__file__).resolve().parents[1], + env=environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + deadline = time.monotonic() + 10 + while not ready.exists() and process.poll() is None and time.monotonic() < deadline: + time.sleep(0.05) + if not ready.exists(): + stdout, stderr = process.communicate(timeout=2) + self.fail(f"signal child did not become ready: {stdout}\n{stderr}") + process.send_signal(signal.SIGINT) + stdout, stderr = process.communicate(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + stdout, stderr = process.communicate(timeout=5) + self.fail(f"signal child did not exit: {stdout}\n{stderr}") + + metadata_paths = tuple(cache.rglob("run.json")) + payload = json.loads(metadata_paths[0].read_text(encoding="utf-8")) + temp_dir = metadata_paths[0].parent / "tmp" / "signal-child" / payload["run_id"] + temp_contents = tuple(temp_dir.iterdir()) + + self.assertEqual(process.returncode, base_cli.ExitCode.INTERRUPTED, stderr) + self.assertIn("Interrupted.", stderr) + self.assertEqual(len(metadata_paths), 1) + self.assertEqual(payload["outcome"], "interrupted") + self.assertEqual(payload["status"], "error") + self.assertEqual(temp_contents, ()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/validate.sh b/tests/validate.sh index 1e5c23f..f33979c 100755 --- a/tests/validate.sh +++ b/tests/validate.sh @@ -18,6 +18,7 @@ required_files=( scripts/validate_package_artifact.py scripts/validate_installed_package.py scripts/validate_docs.py + scripts/benchmark_runtime.py tests/conftest.py ) From e95862d5064b04cb769204dd2bd60bc8afc7f1be Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:26:14 -0700 Subject: [PATCH 2/2] Harden cross-process file persistence on Windows --- lib/python/base_cli/_private_files.py | 17 +++++++- lib/python/base_cli/logging.py | 57 ++++++++++++++++++++++++++- 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/lib/python/base_cli/_private_files.py b/lib/python/base_cli/_private_files.py index 2f1db5c..f25ffae 100644 --- a/lib/python/base_cli/_private_files.py +++ b/lib/python/base_cli/_private_files.py @@ -5,6 +5,7 @@ import json import os import secrets +import time from collections.abc import Mapping from pathlib import Path from typing import Any @@ -98,7 +99,7 @@ def write_private_json(path: Path, value: Mapping[str, Any]) -> None: else: if path.is_symlink(): raise OSError(f"refusing to replace symlink '{path}'") - os.replace(path.parent / temporary_name, path) + _replace_with_retry(path.parent / temporary_name, path) temporary_name = None if parent_fd is not None: _sync_directory(parent_fd) @@ -177,3 +178,17 @@ def _sync_directory(parent_fd: int) -> None: # Directory fsync is not available on all supported filesystems and # platforms. The file itself was still flushed before replacement. pass + + +def _replace_with_retry(source: Path, destination: Path) -> None: + """Replace a private file, tolerating transient Windows sharing races.""" + + attempts = 1 if os.name != "nt" else 10 + for attempt in range(attempts): + try: + os.replace(source, destination) + return + except PermissionError: + if attempt == attempts - 1: + raise + time.sleep(0.001 * (attempt + 1)) diff --git a/lib/python/base_cli/logging.py b/lib/python/base_cli/logging.py index bd8b34f..cfbb6d5 100644 --- a/lib/python/base_cli/logging.py +++ b/lib/python/base_cli/logging.py @@ -6,7 +6,17 @@ import sys import time from pathlib import Path -from typing import TextIO +from typing import BinaryIO, TextIO + +try: + import fcntl as _fcntl +except ImportError: # pragma: no cover - fcntl is unavailable on Windows. + _fcntl = None # type: ignore[assignment] + +try: + import msvcrt as _msvcrt +except ImportError: # pragma: no cover - msvcrt is unavailable outside Windows. + _msvcrt = None # type: ignore[assignment] from ._private_files import restrict_file from .context import get_current_context @@ -107,6 +117,10 @@ def secure_log_file_permissions(log_file: Path) -> None: class SecureLogFileHandler(logging.FileHandler): + def __init__(self, filename: str | os.PathLike[str], *args: object, **kwargs: object) -> None: + self._lock_path = Path(filename).with_name(f".{Path(filename).name}.lock") + super().__init__(filename, *args, **kwargs) # type: ignore[arg-type] + def _open(self) -> TextIO: fd = os.open(self.baseFilename, _secure_log_file_open_flags(self.mode), 0o600) try: @@ -118,6 +132,47 @@ def _open(self) -> TextIO: os.close(fd) raise + def emit(self, record: logging.LogRecord) -> None: + lock_stream = _open_log_lock(self._lock_path) + try: + _lock_log_stream(lock_stream) + super().emit(record) + finally: + _unlock_log_stream(lock_stream) + lock_stream.close() + + +def _open_log_lock(path: Path) -> BinaryIO: + path.parent.mkdir(parents=True, exist_ok=True) + stream = path.open("a+b") + try: + if stream.seek(0, os.SEEK_END) == 0: + stream.write(b"0") + stream.flush() + restrict_file(path) + return stream + except BaseException: + stream.close() + raise + + +def _lock_log_stream(stream: BinaryIO) -> None: + fd = stream.fileno() + if _fcntl is not None: + _fcntl.flock(fd, _fcntl.LOCK_EX) + elif _msvcrt is not None: + stream.seek(0) + _msvcrt.locking(fd, _msvcrt.LK_LOCK, 1) + + +def _unlock_log_stream(stream: BinaryIO) -> None: + fd = stream.fileno() + if _fcntl is not None: + _fcntl.flock(fd, _fcntl.LOCK_UN) + elif _msvcrt is not None: + stream.seek(0) + _msvcrt.locking(fd, _msvcrt.LK_UNLCK, 1) + def _secure_log_file_open_flags(mode: str) -> int: flags = os.O_CREAT