Compare commits

..
15 changed files with 425 additions and 81 deletions
+1
View File
@@ -4,6 +4,7 @@ on:
push: push:
branches: [main, dev] branches: [main, dev]
pull_request: pull_request:
workflow_dispatch:
# Least privilege: none of the jobs write to the repo. # Least privilege: none of the jobs write to the repo.
permissions: permissions:
+1 -1
View File
@@ -3,7 +3,7 @@ name: Native test suites
on: on:
pull_request: pull_request:
push: push:
branches: [main, dev] branches: [main, dev, test/remediation-integration-validation]
workflow_dispatch: workflow_dispatch:
permissions: permissions:
+36 -6
View File
@@ -11,6 +11,38 @@ import time
from urllib.request import urlopen from urllib.request import urlopen
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
def _stop_owned_server(server, identity):
"""Close only identity-bound descendants, including MCP's new sessions.
Capture before signalling the parent: an orphaned stderr holder can no
longer be attributed by ancestry. Never search by executable name or kill
the caller's process group.
"""
from src import process_lifecycle as lifecycle
descendants = (lifecycle.bind_descendants([identity.pid])
if identity.verdict() == lifecycle.OWNED else [])
if os.name == "nt":
if identity.verdict() == lifecycle.OWNED:
lifecycle.taskkill_tree(identity.pid)
else:
outcome = lifecycle.terminate_identities(
[*(entry.identity for entry in reversed(descendants)), identity],
steps=lifecycle.term_kill_steps(10),
)
if outcome.survivors or outcome.unverified:
raise RuntimeError(f"Disposable Notes cleanup could not verify exit: {outcome}")
try:
logs, _ = server.communicate(timeout=2)
except subprocess.TimeoutExpired as error:
# Do not turn an unidentified pipe holder into an unbounded wait.
server.stdout.close()
raise RuntimeError("Disposable Notes output pipe remained open after owned cleanup") from error
return logs
def main(): def main():
@@ -36,7 +68,10 @@ def main():
server = subprocess.Popen( server = subprocess.Popen(
[sys.executable, "-m", "uvicorn", "app:app", "--host", "127.0.0.1", "--port", str(port)], [sys.executable, "-m", "uvicorn", "app:app", "--host", "127.0.0.1", "--port", str(port)],
cwd=target, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=target, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
start_new_session=(os.name != "nt"),
) )
from src.process_lifecycle import ProcessIdentity
server_identity = ProcessIdentity.capture(server.pid)
result = None result = None
try: try:
origin = f"http://127.0.0.1:{port}" origin = f"http://127.0.0.1:{port}"
@@ -62,12 +97,7 @@ def main():
print(result.stderr, file=sys.stderr) print(result.stderr, file=sys.stderr)
return result.returncode return result.returncode
finally: finally:
server.terminate() logs = _stop_owned_server(server, server_identity)
try:
logs, _ = server.communicate(timeout=10)
except subprocess.TimeoutExpired:
server.kill()
logs, _ = server.communicate()
if result is None or result.returncode: if result is None or result.returncode:
print(logs[-8000:], file=sys.stderr) print(logs[-8000:], file=sys.stderr)
+4 -2
View File
@@ -99,8 +99,10 @@ cleanup. Specifically:
`os.environ[...] = ...` that outlives the test. `os.environ[...] = ...` that outlives the test.
- **Current working directory** - never `chdir` without restoring; never assert - **Current working directory** - never `chdir` without restoring; never assert
against cwd-relative paths like `./data`. Use a temp workspace helper instead. against cwd-relative paths like `./data`. Use a temp workspace helper instead.
- **Database** - the root `conftest.py` defaults `DATABASE_URL` to an in-memory - **Database** - the root `conftest.py` defaults `DATABASE_URL` to a private,
SQLite for collection safety. A test that needs a real file-backed DB must opt worker-owned SQLite file for collection safety and consistent schema across
TestClient threads. This fallback is not a per-test data fixture. A test that
writes or depends on particular rows must opt
in explicitly via `tests.helpers.sqlite_db.make_temp_sqlite` and bind its in explicitly via `tests.helpers.sqlite_db.make_temp_sqlite` and bind its
`SessionLocal` onto the module under test. Do not rely on a persistent `SessionLocal` onto the module under test. Do not rely on a persistent
on-disk DB existing. on-disk DB existing.
+14 -1
View File
@@ -12,9 +12,17 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# File-backed databases remain fixture-owned. Cleanup restores the caller. # File-backed databases remain fixture-owned. Cleanup restores the caller.
# Keep the existing generated environment-reference locations stable. # Keep the existing generated environment-reference locations stable.
_database_environment = pytest.MonkeyPatch() _database_environment = pytest.MonkeyPatch()
_database_environment.setenv("DATABASE_URL", "sqlite:///:memory:")
from tests.helpers.worker_runtime import bootstrap_runtime, configure_runtime from tests.helpers.worker_runtime import bootstrap_runtime, configure_runtime
_runtime_environment = bootstrap_runtime() _runtime_environment = bootstrap_runtime()
# A file keeps the collection schema consistent across TestClient threads;
# SingletonThreadPool can evict the populated in-memory main-thread database.
# One patcher owns this variable, including restoration of an inherited URL.
_database_environment.setenv(
"DATABASE_URL",
f"sqlite:///{_runtime_environment.root / 'collection.db'}"
if _runtime_environment is not None else "sqlite:///:memory:",
)
_collection_database_engine = None
# Pre-import real heavy modules BEFORE any test file's module-level stubs can # Pre-import real heavy modules BEFORE any test file's module-level stubs can
# replace them with MagicMock. Some test files (e.g. test_llm_core_sanitize_*) # replace them with MagicMock. Some test files (e.g. test_llm_core_sanitize_*)
@@ -26,6 +34,7 @@ try:
import sqlalchemy.orm # noqa: F401 import sqlalchemy.orm # noqa: F401
import core.database # noqa: F401 import core.database # noqa: F401
import src.database import src.database
_collection_database_engine = core.database.engine
except ImportError: except ImportError:
pass # not installed - the stubs below will handle it pass # not installed - the stubs below will handle it
@@ -100,6 +109,10 @@ def pytest_configure(config):
only registers marker names; it imports no production module. only registers marker names; it imports no production module.
""" """
config.add_cleanup(_database_environment.undo) config.add_cleanup(_database_environment.undo)
if _collection_database_engine is not None:
# Registered after runtime cleanup, so connections close before the
# owned file is removed (also required on Windows).
config.add_cleanup(_collection_database_engine.dispose)
import pathlib import pathlib
from tests._taxonomy import discover_markers from tests._taxonomy import discover_markers
+28 -4
View File
@@ -14,12 +14,22 @@ from routes.auth_routes import SESSION_COOKIE
from tests.helpers.database import disposable_database from tests.helpers.database import disposable_database
@pytest.fixture @pytest.fixture(scope="module")
def authenticated_app(tmp_path, monkeypatch): def auth_configuration(tmp_path_factory):
monkeypatch.setenv("AUTH_ENABLED", "true") """Hash real fixture passwords once; only immutable bytes are reused."""
auth = AuthManager(str(tmp_path / "auth.json")) path = tmp_path_factory.mktemp("admin-auth-seed") / "auth.json"
auth = AuthManager(str(path))
for user, admin in (("member", False), ("admin", True)): for user, admin in (("member", False), ("admin", True)):
assert auth.create_user(user, "disposable-test-password", is_admin=admin) assert auth.create_user(user, "disposable-test-password", is_admin=admin)
return path.read_bytes()
@pytest.fixture
def authenticated_app(tmp_path, monkeypatch, auth_configuration):
monkeypatch.setenv("AUTH_ENABLED", "true")
path = tmp_path / "auth.json"
path.write_bytes(auth_configuration)
auth = AuthManager(str(path))
app = FastAPI() app = FastAPI()
app.state.auth_manager = auth app.state.auth_manager = auth
@@ -133,3 +143,17 @@ def test_diagnostics_require_real_admin_before_sensitive_work(
for operation in (stats, rag.get_stats, research.call_research_service, for operation in (stats, rag.get_stats, research.call_research_service,
extract_id, transcript, collect, read_log): extract_id, transcript, collect, read_log):
operation.assert_not_called() operation.assert_not_called()
@pytest.mark.parametrize("instance", [0, 1])
def test_authenticated_fixture_has_private_users_sessions_and_file(authenticated_app, instance):
app, client_for = authenticated_app
auth = app.state.auth_manager
assert set(auth.users) == {"member", "admin"}
assert not auth._sessions
with client_for("member") as client:
assert auth.get_username_for_token(client.cookies.get(SESSION_COOKIE)) == "member"
assert auth.create_user("fixture-local", "disposable-test-password")
# Both parameter cases mutate their own file. Any shared manager, mutable
# users/sessions or written-back seed makes the next case fail, in any order.
assert "fixture-local" in auth.users
+43 -6
View File
@@ -1929,23 +1929,60 @@ def test_place_layer_reports_false_on_queued_response(monkeypatch):
assert placed is False assert placed is False
def test_bridge_await_cancellation_handled_cleanly(): def test_bridge_await_cancellation_handled_cleanly(monkeypatch):
import asyncio import asyncio
from contextlib import suppress
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
app = _bridge_app() app = _bridge_app()
async def scenario(): async def scenario():
entered_wait = asyncio.Event()
real_wait_for = asyncio.wait_for
waits = []
async def observe_wait(awaitable, timeout):
# Observe the real wait; do not replace its result or cancellation.
waits.append(awaitable)
entered_wait.set()
return await real_wait_for(awaitable, timeout)
monkeypatch.setattr(eaar.asyncio, "wait_for", observe_wait)
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
cmd_resp = await c.post("/api/editor/bridge/command", json={"session": "cancel-sess", "op": "document.info", "wait": False}) cmd_resp = await c.post("/api/editor/bridge/command", json={"session": "cancel-sess", "op": "document.info", "wait": False})
assert cmd_resp.status_code == 200
assert cmd_resp.json()["completed"] is False
cid = cmd_resp.json()["queued"][0] cid = cmd_resp.json()["queued"][0]
await_task = asyncio.create_task(c.post("/api/editor/bridge/await", json={"id": cid, "timeout": 30})) await_task = asyncio.create_task(c.post("/api/editor/bridge/await", json={"id": cid, "timeout": 30}))
await asyncio.sleep(0.05)
await_task.cancel()
try: try:
await await_task await real_wait_for(entered_wait.wait(), 5)
except asyncio.CancelledError: assert len(waits) == 1
pass assert not waits[0].done()
assert not await_task.done(), "await returned without a queued result"
await_task.cancel()
response = await real_wait_for(await_task, 5)
assert response.status_code == 200
assert response.json() == {
"ok": False, "error": "cancelled", "status": "cancelled",
"completed": False, "id": cid,
}
finally:
if not await_task.done():
await_task.cancel()
with suppress(asyncio.CancelledError):
await await_task
# Cancelling the waiter must not cancel the queued command itself.
payload = {"width": 137, "height": 251}
delivered = await c.post("/api/editor/bridge/result", json={"id": cid, "ok": True, "result": payload})
assert delivered.status_code == 200
completed = await c.post("/api/editor/bridge/await", json={"id": cid, "timeout": 1})
assert completed.status_code == 200
assert completed.json()["status"] == "completed"
assert completed.json()["completed"] is True
assert completed.json()["ok"] is True
assert completed.json()["result"] == payload
assert len(waits) == 2
asyncio.run(scenario()) asyncio.run(scenario())
+28 -13
View File
@@ -206,35 +206,50 @@ class TestDiskConsistency:
@pytest.mark.slow @pytest.mark.slow
def test_file_always_valid_json_during_concurrent_ops(self, tmp_path): def test_file_always_valid_json_during_concurrent_ops(self, tmp_path):
mgr = _fresh_auth_manager(tmp_path) mgr = _fresh_auth_manager(tmp_path)
mgr.create_user("admin", "adminpw", is_admin=True) assert mgr.create_user("admin", "adminpw", is_admin=True)
stop_event = threading.Event() stop_event = threading.Event()
reader_ready = threading.Event()
corruption_found = [] corruption_found = []
observations = {"reads": 0, "with_writes": 0}
def reader(): def reader():
while not stop_event.is_set(): while not stop_event.is_set():
try: try:
with open(mgr.auth_path, "r") as f: with open(mgr.auth_path, "r") as f:
content = f.read() content = f.read()
json.loads(content) data = json.loads(content)
observations["reads"] += 1
if any(name.startswith("stress") for name in data["users"]):
observations["with_writes"] += 1
reader_ready.set()
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
corruption_found.append(str(e)) corruption_found.append(str(e))
break break
except FileNotFoundError:
pass
time.sleep(0.001) time.sleep(0.001)
def writer(): def writer():
assert reader_ready.wait(5), "reader never observed the initial file"
results = []
for i in range(50): for i in range(50):
mgr.create_user(f"stress{i}", f"pw{i}") results.append(mgr.create_user(f"stress{i}", f"pw{i}"))
return results
reader_thread = threading.Thread(target=reader) # Futures propagate reader/writer exceptions to the test thread.
writer_thread = threading.Thread(target=writer) with ThreadPoolExecutor(max_workers=2) as pool:
reading = pool.submit(reader)
reader_thread.start() writing = pool.submit(writer)
writer_thread.start() try:
writer_thread.join() results = writing.result(timeout=120)
stop_event.set() finally:
reader_thread.join() stop_event.set()
reading.result(timeout=5)
assert not corruption_found, f"Corrupt JSON detected: {corruption_found[0]}" assert not corruption_found, f"Corrupt JSON detected: {corruption_found[0]}"
assert len(results) == 50 and all(results), "writer operations did not succeed"
assert observations["reads"] > 0
assert observations["with_writes"] > 0, "reader saw no persisted writer activity"
expected = {"admin"} | {f"stress{i}" for i in range(50)}
with open(mgr.auth_path) as persisted:
assert set(json.load(persisted)["users"]) == expected
assert set(_fresh_auth_manager(tmp_path).users) == expected
+11 -2
View File
@@ -85,6 +85,9 @@ def test_actual_tests_restore_process_state_and_ignore_inherited_database(tmp_pa
import os import os
import sys import sys
import pytest import pytest
import tempfile
from pathlib import Path
from sqlalchemy.engine import make_url
def snapshot(): def snapshot():
import core import core
@@ -107,7 +110,11 @@ def test_actual_tests_restore_process_state_and_ignore_inherited_database(tmp_pa
class StateGuard: class StateGuard:
def pytest_sessionstart(self): def pytest_sessionstart(self):
self.initial = snapshot() self.initial = snapshot()
assert self.initial[0] == "sqlite:///:memory:" assert self.initial[0] != inherited_url
self.default_path = Path(make_url(self.initial[0]).database)
assert self.default_path.is_file()
assert self.default_path.name == "collection.db"
assert self.default_path.parent == Path(tempfile.gettempdir()).parent
def pytest_collection_finish(self): def pytest_collection_finish(self):
assert snapshot() == self.initial, "collection changed database bindings" assert snapshot() == self.initial, "collection changed database bindings"
@@ -118,9 +125,11 @@ def test_actual_tests_restore_process_state_and_ignore_inherited_database(tmp_pa
assert snapshot() == self.initial, "test leaked database or module state" assert snapshot() == self.initial, "test leaked database or module state"
inherited_url = os.environ["DATABASE_URL"] inherited_url = os.environ["DATABASE_URL"]
guard = StateGuard()
result = pytest.main(["-q", "-p", "no:cacheprovider", *sys.argv[1:]], result = pytest.main(["-q", "-p", "no:cacheprovider", *sys.argv[1:]],
plugins=[StateGuard()]) plugins=[guard])
assert os.environ["DATABASE_URL"] == inherited_url assert os.environ["DATABASE_URL"] == inherited_url
assert not guard.default_path.exists(), "collection database outlived its runtime"
raise SystemExit(result) raise SystemExit(result)
''') ''')
result = subprocess.run( result = subprocess.run(
@@ -11,6 +11,8 @@ cache from the DB, where the rows are already gone, and the token is rejected.
""" """
import asyncio import asyncio
import types import types
import pytest
from fastapi import HTTPException
from routes.auth_routes import setup_auth_routes, DeleteUserRequest from routes.auth_routes import setup_auth_routes, DeleteUserRequest
@@ -60,12 +62,9 @@ def test_refused_delete_does_not_invalidate_cache():
invalidations = [] invalidations = []
router = setup_auth_routes(_auth_manager(delete_result=False)) router = setup_auth_routes(_auth_manager(delete_result=False))
handler = _handler(router) handler = _handler(router)
try: with pytest.raises(HTTPException) as refused:
asyncio.run(handler(DeleteUserRequest(username="admin"), _fake_request(invalidations))) asyncio.run(handler(DeleteUserRequest(username="admin"), _fake_request(invalidations)))
raised = False assert refused.value.status_code == 400
except Exception:
raised = True
assert raised, "a refused delete should raise (HTTP 400)"
assert invalidations == [], "a refused delete must not touch the token cache" assert invalidations == [], "a refused delete must not touch the token cache"
@@ -93,7 +93,7 @@ def test_custom_filter_reuses_convolve_kernel():
assert "convolve(readSource(snap), k, divisor, bias, 1)" in EFFECTS assert "convolve(readSource(snap), k, divisor, bias, 1)" in EFFECTS
def test_new_filter_aliases_registered(): def test_blur_and_custom_filter_aliases_registered():
assert "'box-blur': 'blur-box'" in EDITOR assert "'box-blur': 'blur-box'" in EDITOR
assert "'surface-blur': 'blur-surface'" in EDITOR assert "'surface-blur': 'blur-surface'" in EDITOR
assert "custom: 'other-custom'" in EDITOR assert "custom: 'other-custom'" in EDITOR
@@ -56,14 +56,35 @@ def test_cold_start_with_no_snapshot_and_no_index_is_not_marked_stale(monkeypatc
# falls through to the real (blocking) path, so the response is not a # falls through to the real (blocking) path, so the response is not a
# stale-marked snapshot. # stale-marked snapshot.
endpoint = _list_endpoint() endpoint = _list_endpoint()
calls = []
class EmptyMailbox:
def select(self, folder, readonly=False):
calls.append(("select", folder, readonly))
return "OK", [b"0"]
def uid(self, command, *args):
assert command == "SEARCH"
calls.append(("search", args))
return "OK", [b""]
def logout(self):
pass
def connect(*args, **kwargs):
calls.append(("connect", args, kwargs))
return EmptyMailbox()
monkeypatch.setattr(er, "_start_poller", lambda: None)
monkeypatch.setattr(er, "_email_index_list", lambda *a, **k: ([], 0, None)) monkeypatch.setattr(er, "_email_index_list", lambda *a, **k: ([], 0, None))
monkeypatch.setattr(er, "_imap_connect", lambda *a, **k: (_ for _ in ()).throw( monkeypatch.setattr(er, "_imap_connect", connect)
RuntimeError("offline test"))) result = _call(endpoint)
try: assert [call[0] for call in calls] == ["connect", "select", "search"]
result = _call(endpoint) assert calls[1][2] is True
except Exception:
return # the cold path legitimately attempted real IMAP
assert isinstance(result, dict) assert isinstance(result, dict)
assert result["emails"] == []
assert result["total"] == 0
assert not result.get("error")
assert not (result.get("sync") or {}).get("stale") assert not (result.get("sync") or {}).get("stale")
+43
View File
@@ -0,0 +1,43 @@
"""A detached Notes descendant must not hold output open or kill a bystander."""
import os
import runpy
import subprocess
import sys
from pathlib import Path
import pytest
from src import process_lifecycle as lifecycle
ROOT = Path(__file__).resolve().parents[1]
@pytest.mark.skipif(os.name == "nt", reason="POSIX detached-session regression; Windows uses taskkill tree")
def test_notes_cleanup_reaps_detached_pipe_holder_and_preserves_bystander():
stop = runpy.run_path(str(ROOT / "scripts/verify_notes_browser_runtime.py"))["_stop_owned_server"]
child = "import time; time.sleep(60)"
program = ("import subprocess,sys,time; "
f"p=subprocess.Popen([sys.executable,'-c',{child!r}],start_new_session=True); "
"print(p.pid,flush=True); time.sleep(60)")
with subprocess.Popen([sys.executable, "-c", child]) as bystander:
server = subprocess.Popen([sys.executable, "-c", program], stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True, start_new_session=True)
root_identity = lifecycle.ProcessIdentity.capture(server.pid)
detached = None
try:
child_pid = int(server.stdout.readline())
detached = lifecycle.ProcessIdentity.capture(child_pid)
assert os.getpgid(child_pid) == child_pid
stop(server, root_identity)
assert server.poll() is not None
assert detached.exited()
assert bystander.poll() is None, "cleanup touched an unrelated process"
finally:
# Test-owned fallback cleanup also runs when the negative control
# intentionally substitutes the old parent-only teardown.
targets = [identity for identity in (detached, root_identity) if identity]
lifecycle.terminate_identities(targets, steps=lifecycle.term_kill_steps(1))
server.wait(timeout=3)
if server.stdout:
server.stdout.close()
bystander.terminate()
bystander.wait(timeout=3)
+155 -34
View File
@@ -12,6 +12,8 @@ exercising every procfs-touching code path on both a Linux and a non-Linux
host. ``test_cookbook_stop_without_procfs.py`` covers the behaviour itself. host. ``test_cookbook_stop_without_procfs.py`` covers the behaviour itself.
""" """
import ast import ast
import pytest
from pathlib import Path from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent REPO_ROOT = Path(__file__).resolve().parent.parent
@@ -75,37 +77,112 @@ def _scan_sites(tree: ast.AST) -> list[ast.Call]:
return sites return sites
def _guard_lines(tree: ast.AST) -> list[int]: def _condition_requires_procfs(node: ast.AST, truth: bool) -> bool:
"""Line numbers of calls that test whether procfs is present.""" """Whether this Boolean outcome proves a direct procfs check succeeded.
lines = []
for node in ast.walk(tree): An ignored call, another function's check, or a check on only one possible
if not isinstance(node, ast.Call): OR branch provides no proof. Keep unknown expressions conservative.
continue """
func = node.func if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not):
name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") return _condition_requires_procfs(node.operand, not truth)
if name not in _GUARD_FUNCS: if isinstance(node, ast.BoolOp):
continue facts = [_condition_requires_procfs(value, truth) for value in node.values]
subject_is_procfs = ( all_values_required = isinstance(node.op, ast.And) == truth
name in {"has_procfs", "is_wsl"} return any(facts) if all_values_required else all(facts)
or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value)) if not isinstance(node, ast.Call):
or any(_is_procfs_root(arg) for arg in node.args) return False
) func = node.func
if subject_is_procfs: name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
lines.append(node.lineno) subject_is_procfs = (
return lines name == "has_procfs"
or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value))
or any(_is_procfs_root(arg) for arg in node.args)
)
return truth and name in _GUARD_FUNCS and subject_is_procfs
def _enclosing_scope(tree: ast.AST, node: ast.AST) -> ast.AST: def _guarded_scan_ids(tree: ast.AST) -> set[int]:
"""Smallest function/module scope containing ``node``.""" """Follow conditional/early-exit paths instead of comparing line numbers.
best = tree
for candidate in ast.walk(tree): Each function starts with no procfs assumption. Facts from a loop or try
if not isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef)): body cannot escape to code which may run without that body completing.
continue This intentionally requires direct checks; arbitrary aliases/call graphs
end = getattr(candidate, "end_lineno", None) or candidate.lineno are not a proof that procfs exists at the enumeration site.
if candidate.lineno <= node.lineno <= end: """
if best is tree or candidate.lineno > best.lineno: scans = {id(site) for site in _scan_sites(tree)}
best = candidate guarded = set()
return best
def expression(node, present):
if node is None:
return
if id(node) in scans and present:
guarded.add(id(node))
if isinstance(node, ast.BoolOp):
for value in node.values:
expression(value, present)
# Later AND operands run only after a true prefix; later OR
# operands run only after a false prefix.
present = present or _condition_requires_procfs(value, isinstance(node.op, ast.And))
elif isinstance(node, ast.IfExp):
expression(node.test, present)
expression(node.body, present or _condition_requires_procfs(node.test, True))
expression(node.orelse, present or _condition_requires_procfs(node.test, False))
elif isinstance(node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp)):
for generator in node.generators:
expression(generator.iter, present)
for condition in generator.ifs:
expression(condition, present)
present = present or _condition_requires_procfs(condition, True)
for attr in ("elt", "key", "value"):
expression(getattr(node, attr, None), present)
elif isinstance(node, ast.Lambda):
expression(node.body, False)
else:
for child in ast.iter_child_nodes(node):
expression(child, present)
def block(statements, present):
for statement in statements:
if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)):
block(statement.body, False)
for value in statement.decorator_list + statement.args.defaults:
expression(value, present)
elif isinstance(statement, ast.ClassDef):
block(statement.body, present)
elif isinstance(statement, ast.If):
expression(statement.test, present)
yes = block(statement.body, present or _condition_requires_procfs(statement.test, True))
no = block(statement.orelse, present or _condition_requires_procfs(statement.test, False))
continuing = [fact for fact in (yes, no) if fact is not None]
if not continuing:
return None
present = all(continuing)
elif isinstance(statement, (ast.Return, ast.Raise, ast.Break, ast.Continue)):
for child in ast.iter_child_nodes(statement):
expression(child, present)
return None
elif isinstance(statement, (ast.For, ast.AsyncFor, ast.While)):
condition = statement.test if isinstance(statement, ast.While) else statement.iter
expression(condition, present)
body_present = present or (isinstance(statement, ast.While) and _condition_requires_procfs(condition, True))
block(statement.body, body_present)
block(statement.orelse, present)
elif isinstance(statement, (ast.Try, getattr(ast, "TryStar", ast.Try))):
block(statement.body, present)
block(statement.orelse, present)
block(statement.finalbody, present)
for handler in statement.handlers:
block(handler.body, present)
elif isinstance(statement, (ast.With, ast.AsyncWith)):
for item in statement.items:
expression(item.context_expr, present)
block(statement.body, present)
else:
expression(statement, present)
return present
block(tree.body, False)
return guarded
def _app_python_files() -> list[Path]: def _app_python_files() -> list[Path]:
@@ -139,12 +216,10 @@ def _collect_sites() -> tuple[list[str], list[str]]:
sites = _scan_sites(tree) sites = _scan_sites(tree)
if not sites: if not sites:
continue continue
guards = _guard_lines(tree) guards = _guarded_scan_ids(tree)
for site in sites: for site in sites:
scope = _enclosing_scope(tree, site)
start = getattr(scope, "lineno", 0)
label = f"{path.relative_to(REPO_ROOT)}:{site.lineno}" label = f"{path.relative_to(REPO_ROOT)}:{site.lineno}"
if any(start <= g < site.lineno for g in guards): if id(site) in guards:
guarded.append(label) guarded.append(label)
else: else:
unguarded.append(label) unguarded.append(label)
@@ -154,7 +229,7 @@ def _collect_sites() -> tuple[list[str], list[str]]:
def test_every_procfs_scan_is_guarded_by_an_existence_check(): def test_every_procfs_scan_is_guarded_by_an_existence_check():
_guarded, unguarded = _collect_sites() _guarded, unguarded = _collect_sites()
assert not unguarded, ( assert not unguarded, (
"procfs pid scans with no existence check in the enclosing function — " "procfs pid scans without an effective conditional existence guard — "
"these raise FileNotFoundError on macOS and Windows: " "these raise FileNotFoundError on macOS and Windows: "
+ ", ".join(unguarded) + ", ".join(unguarded)
) )
@@ -176,3 +251,49 @@ def test_the_guard_detector_still_sees_the_known_scans():
f"root was renamed. Found: {sorted(found)}" f"root was renamed. Found: {sorted(found)}"
) )
assert len(found) >= 3, f"expected at least 3 procfs scans, found {sorted(found)}" assert len(found) >= 3, f"expected at least 3 procfs scans, found {sorted(found)}"
@pytest.mark.parametrize("body", [
'Path("/proc").exists(); os.listdir("/proc")',
'if Path("/proc").exists():\n pass\nos.listdir("/proc")',
'if not Path("/proc").exists():\n os.listdir("/proc")',
'if ready or Path("/proc").exists():\n os.listdir("/proc")',
'if ready:\n Path("/proc").exists()\nos.listdir("/proc")',
'os.listdir("/proc"); Path("/proc").exists()',
'if Path("/tmp").exists():\n os.listdir("/proc")',
'def other():\n Path("/proc").exists()\nos.listdir("/proc")',
'if not has_procfs():\n pass\nos.listdir("/proc")',
'for x in []:\n if not has_procfs():\n return\nos.listdir("/proc")',
'if has_procfs():\n return\nelse:\n os.listdir("/proc")',
'[pid for pid in os.listdir("/proc") if has_procfs()]',
], ids=["ignored", "expired-branch", "inverted", "unsafe-or", "sibling-branch",
"too-late", "wrong-path", "other-function", "no-early-exit", "loop-may-not-run",
"wrong-else", "late-comprehension-filter"])
def test_guard_detector_rejects_ineffective_checks(body):
tree = ast.parse("def scan():\n" + "\n".join(" " + line for line in body.splitlines()))
sites = _scan_sites(tree)
assert sites, "negative control must contain a recognized enumeration"
assert any(id(site) not in _guarded_scan_ids(tree) for site in sites)
@pytest.mark.parametrize("body", [
'if Path("/proc").exists():\n os.listdir("/proc")',
'if not Path("/proc").exists():\n return\nos.listdir("/proc")',
'if not has_procfs():\n raise RuntimeError()\nos.listdir(PROC_ROOT)',
'if has_procfs():\n pass\nelse:\n return\nPROC_ROOT.iterdir()',
'if ready and has_procfs():\n os.scandir("/proc")',
'if not has_procfs() or disabled:\n return\nPROC_ROOT.glob("*")',
'if sys.platform == "linux" and has_procfs():\n PROC_ROOT.iterdir()',
'for item in items:\n if not has_procfs():\n continue\n os.listdir("/proc")',
'if os.path.isdir("/proc"):\n os.listdir("/proc")',
'if has_procfs() and list(PROC_ROOT.iterdir()):\n pass',
'if not has_procfs() or list(PROC_ROOT.iterdir()):\n pass',
'[pid for item in items if has_procfs() for pid in os.listdir("/proc")]',
], ids=["positive-branch", "early-return", "early-raise", "else-exits", "and",
"negative-or", "platform-and-procfs", "loop-continue", "isdir",
"short-circuit-and", "short-circuit-or", "guarded-comprehension"])
def test_guard_detector_preserves_effective_checks(body):
tree = ast.parse("def scan():\n" + "\n".join(" " + line for line in body.splitlines()))
sites = _scan_sites(tree)
assert sites, "positive control must contain a recognized enumeration"
assert all(id(site) in _guarded_scan_ids(tree) for site in sites)
+29
View File
@@ -79,3 +79,32 @@ def test_tmp_path_under_private_runtime_fits_unix_socket(tmp_path):
# witness. A nested pytest-of-<user> basetemp made it 110 bytes under xdist. # witness. A nested pytest-of-<user> basetemp made it 110 bytes under xdist.
with socket.socket(socket.AF_UNIX) as sock: with socket.socket(socket.AF_UNIX) as sock:
sock.bind(str(tmp_path / "tmux.sock")) sock.bind(str(tmp_path / "tmux.sock"))
def test_collection_database_survives_thread_connection_churn():
"""TestClient threads must not lose the collection database or its schema.
SingletonThreadPool's default five-connection limit used to close the main
thread's in-memory database after repeated route clients. Each new thread
also saw an empty database. Exercise more threads than that limit and then
verify both compatibility factories still see all registered tables.
"""
from concurrent.futures import ThreadPoolExecutor
from sqlalchemy import inspect
import core.database as database
import src.database as compatibility
expected = set(database.Base.metadata.tables)
def table_names(factory):
with factory() as session:
return set(inspect(session.connection()).get_table_names())
assert {"documents", "model_endpoints", "mcp_servers", "calendars"} <= expected
assert expected <= table_names(database.SessionLocal)
for _ in range(8):
# A fresh executor creates a fresh thread rather than reusing one.
with ThreadPoolExecutor(max_workers=1) as executor:
assert expected <= executor.submit(table_names, database.SessionLocal).result()
assert expected <= table_names(database.SessionLocal)
assert expected <= table_names(compatibility.SessionLocal)