fix TOCTOU races, FIPS compat, type shadows, and other code review fixes

Author: nsheff

benchmarks/benchmark_lockers.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""Benchmark: OneLocker vs ThreeLocker.
+
+Run: python benchmark_lockers.py
+"""
+
+import multiprocessing
+import os
+import sys
+import tempfile
+import time
+
+multiprocessing.set_start_method("spawn", force=True)
+
+from ubiquerg import OneLocker, ThreeLocker
+
+
+def benchmark_exclusive_access(filepath, iterations=100):
+    """Measure lock/unlock cycle speed for exclusive access."""
+    one = OneLocker(filepath)
+    start = time.perf_counter()
+    for _ in range(iterations):
+        one.write_lock()
+        one.write_unlock()
+    one_time = time.perf_counter() - start
+
+    three = ThreeLocker(filepath)
+    start = time.perf_counter()
+    for _ in range(iterations):
+        three.write_lock()
+        three.write_unlock()
+    three_time = time.perf_counter() - start
+
+    return one_time, three_time
+
+
+def reader_process(filepath, locker_type, hold_time, barrier, timestamps, idx):
+    """Subprocess: wait for barrier, acquire read lock, hold it, record times."""
+    if locker_type == "three":
+        locker = ThreeLocker(filepath)
+    else:
+        locker = OneLocker(filepath)
+
+    barrier.wait()
+    t_start = time.perf_counter()
+    locker.read_lock()
+    t_acquired = time.perf_counter()
+    time.sleep(hold_time)
+    locker.read_unlock()
+    t_done = time.perf_counter()
+
+    timestamps[idx] = (t_start, t_acquired, t_done)
+
+
+def benchmark_concurrent_reads(filepath, n_readers=4, hold_time=0.5):
+    """Spawn n_readers behind a barrier, measure overlap behavior."""
+    results = {}
+
+    for locker_type in ["one", "three"]:
+        barrier = multiprocessing.Barrier(n_readers)
+        manager = multiprocessing.Manager()
+        timestamps = manager.dict()
+
+        processes = []
+        for i in range(n_readers):
+            p = multiprocessing.Process(
+                target=reader_process,
+                args=(filepath, locker_type, hold_time, barrier, timestamps, i),
+            )
+            processes.append(p)
+
+        for p in processes:
+            p.start()
+        for p in processes:
+            p.join(timeout=60)
+
+        starts = [timestamps[i][0] for i in range(n_readers)]
+        acquireds = [timestamps[i][1] for i in range(n_readers)]
+        dones = [timestamps[i][2] for i in range(n_readers)]
+
+        wall = max(dones) - min(starts)
+        avg_wait = sum(a - s for s, a in zip(starts, acquireds)) / n_readers
+
+        results[locker_type] = {"wall": wall, "avg_wait": avg_wait}
+
+    return results
+
+
+def main():
+    tmpdir = tempfile.mkdtemp()
+    filepath = os.path.join(tmpdir, "benchmark.txt")
+    with open(filepath, "w") as f:
+        f.write("benchmark data")
+
+    iterations = 100
+    n_readers = 4
+    hold_times = [0.5, 3.0]
+
+    print("ubiquerg locker benchmark")
+    print("=" * 50)
+
+    # Benchmark 1: Exclusive access speed
+    print(f"\n1. Exclusive lock/unlock ({iterations} iterations)")
+    print("-" * 50)
+    one_time, three_time = benchmark_exclusive_access(filepath, iterations)
+    speedup = three_time / one_time
+    print(f"   OneLocker:   {one_time:.4f}s  ({one_time / iterations * 1000:.2f}ms/cycle)")
+    print(f"   ThreeLocker: {three_time:.4f}s  ({three_time / iterations * 1000:.2f}ms/cycle)")
+    print(f"   OneLocker {speedup:.1f}x faster")
+
+    # Benchmark 2: Concurrent readers at different hold times
+    for hold_time in hold_times:
+        print(f"\n2. Concurrent readers ({n_readers} readers, {hold_time}s hold each)")
+        print("-" * 50)
+        # Suppress wait_for_lock stdout dots (fd-level for child processes)
+        sys.stdout.flush()
+        devnull = os.open(os.devnull, os.O_WRONLY)
+        old_fd = os.dup(1)
+        os.dup2(devnull, 1)
+        os.close(devnull)
+        results = benchmark_concurrent_reads(filepath, n_readers, hold_time)
+        os.dup2(old_fd, 1)
+        os.close(old_fd)
+
+        one = results["one"]
+        three = results["three"]
+        print(f"   OneLocker:   {one['wall']:.2f}s wall  (avg {one['avg_wait']:.2f}s wait)")
+        print(f"   ThreeLocker: {three['wall']:.2f}s wall  (avg {three['avg_wait']:.2f}s wait)")
+
+    os.unlink(filepath)
+    os.rmdir(tmpdir)
+
+
+if __name__ == "__main__":
+    main()

changelog.md

@@ -2,28 +2,38 @@
 
 This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html) and [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) format. 
 
-## [0.9.0] -- 2026-02-10
+## [0.9.0] -- 2026-02-11
+
+### Added
+- `OneLocker` class with context manager support (`with OneLocker(path) as locker:`)
 
 ### Changed
-- Migrated packaging from setup.py/setup.cfg to pyproject.toml with hatchling backend
-- Replaced black with ruff for linting and formatting
-- Updated GitHub Actions workflows to latest versions (checkout@v6, setup-python@v6)
-- Updated publish workflow to use `python -m build` instead of `setup.py sdist`
-- Modernized type hints: replaced `Optional`/`Union` with PEP 604 `X | None` / `X | Y` syntax
-- Converted remaining Sphinx-style docstrings to Google style
-- Dropped Python 3.9 support (minimum is now 3.10)
-- Added Python 3.14 support
+- Migrated to pyproject.toml/hatchling, ruff, latest GitHub Actions
+- Modernized type hints (PEP 604) and docstrings (Google style)
+- Dropped Python 3.9; added Python 3.14
+- `is_command_callable` uses `shutil.which()` instead of `os.system()`
+- `_create_lock` is now iterative with retry limit (was unbounded recursion)
+- `wait_for_lock` progress dots go to stderr instead of stdout
+- `convert_value` passes through `None`/`bool`/`int`/`float` directly
+- `is_url` regex compiled once at module level
 
 ### Fixed
-- Deduplicated `mkabs` in `file_locking.py` — `ThreeLocker` now uses the canonical version from `paths.py` which handles None, URLs, and file-as-reldir
-- Deduplicated `_create_lock` in `file_locking.py` — consolidated to single implementation in `files.py`
-- Fixed `tarfile.extractall()` deprecation warning for Python 3.12+ (added `filter="data"`)
-- Fixed typo "assiated" → "associated" in `cli_tools.py`
-- Re-exported `uniqify` function that was accidentally dropped during explicit-exports migration
+- Shell injection in `is_command_callable`
+- Signal handler leak in `read_lock`/`write_lock` context managers
+- ThreeLocker SIGTERM handler crash
+- `read_lock()`/`_lock()` returning True when locking actually fails
+- `read_unlock()` while write lock held now raises `RuntimeError`
+- `deep_update` crash replacing scalar with nested dict
+- `TmpEnv` destroying pre-existing env vars on exit
+- `arg_defaults(unique=True)` only returning first subcommand's defaults
+- `wait_max=0` treated as falsy in `create_read_lock`/`create_write_lock`
+- `is_writable` infinite recursion and failure on nested paths
+- Mutable default argument in `parse_registry_path`
+- `tarfile.extractall()` deprecation warning on Python 3.12+
 
 ### Removed
-- Deprecated `asciify_dict()` function (was a Python 2 no-op)
-- Legacy packaging files: setup.py, setup.cfg, MANIFEST.in, requirements/
+- `build_cli_extra()` and `asciify_dict()` functions
+- Legacy packaging files (setup.py, setup.cfg, MANIFEST.in, requirements/)
 
 ## [0.8.2] -- 2025-12-01
 

pyproject.toml

@@ -32,6 +32,7 @@ test = ["pytest"]
 [tool.pytest.ini_options]
 addopts = "-rfE"
 testpaths = ["tests"]
+markers = ["benchmark: locker performance benchmark tests"]
 python_files = "test_*.py"
 python_classes = "Test* *Test *Tests *Tester"
 python_functions = "test_* test[A-Z]*"
@@ -46,3 +47,4 @@ ignore = ["E501"]
 
 [tool.ruff.lint.per-file-ignores]
 "tests/*" = ["F403", "F405"]
+"benchmarks/*" = ["E402"]

tests/test_cli_tools.py

@@ -1,17 +1,10 @@
-"""Tests for rendering CLI options and arguments"""
+"""Tests for CLI tools"""
 
 import argparse
-from collections import OrderedDict
 
 import pytest
 
-from ubiquerg import VersionInHelpParser, build_cli_extra, convert_value, powerset
-
-
-def pytest_generate_tests(metafunc):
-    """Test case generation and parameterization for this module"""
-    if "ordwrap" in metafunc.fixturenames:
-        metafunc.parametrize("ordwrap", [tuple, OrderedDict])
+from ubiquerg import VersionInHelpParser, convert_value
 
 
 def build_parser():
@@ -244,36 +237,6 @@ def add_subparser(cmd):
     return parser, msg_by_cmd
 
 
-@pytest.mark.parametrize(
-    ["optargs", "expected"],
-    [
-        (
-            [
-                ("-X", None),
-                ("--revert", 1),
-                ("-O", "outfile"),
-                ("--execute-locally", None),
-                ("-I", ["file1", "file2"]),
-            ],
-            "-X --revert 1 -O outfile --execute-locally -I file1 file2",
-        )
-    ],
-)
-def test_build_cli_extra(optargs, expected, ordwrap):
-    """Check that CLI optargs are rendered as expected."""
-    observed = build_cli_extra(ordwrap(optargs))
-    print("expected: {}".format(expected))
-    print("observed: {}".format(observed))
-    assert expected == observed
-
-
-@pytest.mark.parametrize("optargs", powerset([(None, "a"), (1, "one")], nonempty=True))
-def test_illegal_cli_extra_input_is_exceptional(optargs, ordwrap):
-    """Non-string keys are illegal and cause a TypeError."""
-    with pytest.raises(TypeError):
-        build_cli_extra(ordwrap(optargs))
-
-
 def test_dests_by_subparser_return_type():
     """Check if the return type is dict of lists keyed by subcommand name"""
     parser, msgs_by_cmd = build_parser()
@@ -341,10 +304,23 @@ def test_arg_defaults_unqiue():
     parser, _ = build_parser()
     tld = parser.arg_defaults(unique=True)
     assert isinstance(tld, dict)
-    assert len(tld) == 17
+    assert len(tld) == 19
     assert len(set(tld.keys())) == len(tld.keys())
 
 
+def test_arg_defaults_unique_includes_all_subcommands():
+    """unique=True must merge defaults from ALL subcommands, not just the first."""
+    parser = VersionInHelpParser(prog="test")
+    subs = parser.add_subparsers(dest="command")
+    sub1 = subs.add_parser("first")
+    sub1.add_argument("--only-in-first", default="val1")
+    sub2 = subs.add_parser("second")
+    sub2.add_argument("--only-in-second", default="val2")
+    result = parser.arg_defaults(unique=True)
+    assert "only_in_first" in result, f"Missing first subcommand defaults: {result}"
+    assert "only_in_second" in result, f"Missing second subcommand defaults: {result}"
+
+
 @pytest.mark.parametrize(
     ["input", "output"],
     [

tests/test_collection.py

@@ -238,3 +238,11 @@ def test_powerset_illegal_input(arbwrap, kwargs, exp_err, pool):
 @pytest.mark.parametrize(["x", "y"], [({"a": 1}, {"b": 2}), ({"a": 1, "c": 2}, {"b": 2})])
 def test_merging_dicts(x, y):
     assert "a" in list(merge_dicts(x, y).keys())
+
+
+def test_deep_update_replace_scalar_with_dict():
+    """Replacing a scalar value with a nested dict should not crash."""
+    old = {"a": "string_value", "b": 1}
+    new = {"a": {"nested": "dict"}, "b": {"also": "nested"}}
+    deep_update(old, new)
+    assert old == {"a": {"nested": "dict"}, "b": {"also": "nested"}}

tests/test_environment.py

@@ -56,6 +56,17 @@ def test_no_intersection(overwrite, envvars):
             assert v == os.getenv(k)
 
 
+def test_overwrite_restores_original_value():
+    """With overwrite=True, exiting the context should restore the original value, not delete it."""
+    os.environ["UBIQUERG_TEST_RESTORE"] = "original"
+    try:
+        with TmpEnv(overwrite=True, UBIQUERG_TEST_RESTORE="modified"):
+            assert os.environ["UBIQUERG_TEST_RESTORE"] == "modified"
+        assert os.environ.get("UBIQUERG_TEST_RESTORE") == "original"
+    finally:
+        os.environ.pop("UBIQUERG_TEST_RESTORE", None)
+
+
 def check_unset(envvars):
     """Verify that each environment variable is not set."""
     fails = [v for v in envvars if os.getenv(v) or v in os.environ]