(mounts: list[Path])
| 62 | |
| 63 | |
| 64 | def _verify_across(mounts: list[Path]) -> int: |
| 65 | where = str(mounts[0]) if len(mounts) == 1 else f"{len(mounts)} mounts of {mounts[0]} .. {mounts[-1]}" |
| 66 | print(f"verifying mutual exclusion across {where} ({_PROCESSES} processes x {_HOLDS} holds)") |
| 67 | gated = _gated_locks() |
| 68 | failures = 0 |
| 69 | for name in _ALL_LOCKS: |
| 70 | counts, overlaps, reason = _run_one(name, mounts) |
| 71 | expected = _PROCESSES * _HOLDS |
| 72 | # A correct lock never overlaps, finishes every hold, and raises no error the retries could not clear. |
| 73 | ok = overlaps == 0 and sum(counts) == expected and reason is None |
| 74 | failures += (not ok) and name in gated |
| 75 | note = "" if name in gated else " (ungated)" |
| 76 | tail = f"{note}{f' {reason}' if reason else ''}" |
| 77 | stats = f"held={sum(counts)}/{expected} overlaps={overlaps}" |
| 78 | print(f" {name:20} {'PASS' if ok else 'FAIL'} {stats}{tail}") |
| 79 | return 1 if failures else 0 |
| 80 | |
| 81 | |
| 82 | def _gated_locks() -> frozenset[str]: |
no test coverage detected