Compare commits

..
6 changed files with 186 additions and 137 deletions
+6 -43
View File
@@ -1929,60 +1929,23 @@ def test_place_layer_reports_false_on_queued_response(monkeypatch):
assert placed is False
def test_bridge_await_cancellation_handled_cleanly(monkeypatch):
def test_bridge_await_cancellation_handled_cleanly():
import asyncio
from contextlib import suppress
from httpx import ASGITransport, AsyncClient
app = _bridge_app()
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:
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]
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:
await real_wait_for(entered_wait.wait(), 5)
assert len(waits) == 1
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
await await_task
except asyncio.CancelledError:
pass
asyncio.run(scenario())
+13 -28
View File
@@ -206,50 +206,35 @@ class TestDiskConsistency:
@pytest.mark.slow
def test_file_always_valid_json_during_concurrent_ops(self, tmp_path):
mgr = _fresh_auth_manager(tmp_path)
assert mgr.create_user("admin", "adminpw", is_admin=True)
mgr.create_user("admin", "adminpw", is_admin=True)
stop_event = threading.Event()
reader_ready = threading.Event()
corruption_found = []
observations = {"reads": 0, "with_writes": 0}
def reader():
while not stop_event.is_set():
try:
with open(mgr.auth_path, "r") as f:
content = f.read()
data = json.loads(content)
observations["reads"] += 1
if any(name.startswith("stress") for name in data["users"]):
observations["with_writes"] += 1
reader_ready.set()
json.loads(content)
except json.JSONDecodeError as e:
corruption_found.append(str(e))
break
except FileNotFoundError:
pass
time.sleep(0.001)
def writer():
assert reader_ready.wait(5), "reader never observed the initial file"
results = []
for i in range(50):
results.append(mgr.create_user(f"stress{i}", f"pw{i}"))
return results
mgr.create_user(f"stress{i}", f"pw{i}")
# Futures propagate reader/writer exceptions to the test thread.
with ThreadPoolExecutor(max_workers=2) as pool:
reading = pool.submit(reader)
writing = pool.submit(writer)
try:
results = writing.result(timeout=120)
finally:
stop_event.set()
reading.result(timeout=5)
reader_thread = threading.Thread(target=reader)
writer_thread = threading.Thread(target=writer)
reader_thread.start()
writer_thread.start()
writer_thread.join()
stop_event.set()
reader_thread.join()
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,8 +11,6 @@ cache from the DB, where the rows are already gone, and the token is rejected.
"""
import asyncio
import types
import pytest
from fastapi import HTTPException
from routes.auth_routes import setup_auth_routes, DeleteUserRequest
@@ -62,9 +60,12 @@ def test_refused_delete_does_not_invalidate_cache():
invalidations = []
router = setup_auth_routes(_auth_manager(delete_result=False))
handler = _handler(router)
with pytest.raises(HTTPException) as refused:
try:
asyncio.run(handler(DeleteUserRequest(username="admin"), _fake_request(invalidations)))
assert refused.value.status_code == 400
raised = False
except Exception:
raised = True
assert raised, "a refused delete should raise (HTTP 400)"
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
def test_new_filter_aliases_registered():
def test_blur_and_custom_filter_aliases_registered():
assert "'box-blur': 'blur-box'" in EDITOR
assert "'surface-blur': 'blur-surface'" in EDITOR
assert "custom: 'other-custom'" in EDITOR
@@ -56,35 +56,14 @@ 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
# stale-marked snapshot.
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, "_imap_connect", connect)
result = _call(endpoint)
assert [call[0] for call in calls] == ["connect", "select", "search"]
assert calls[1][2] is True
monkeypatch.setattr(er, "_imap_connect", lambda *a, **k: (_ for _ in ()).throw(
RuntimeError("offline test")))
try:
result = _call(endpoint)
except Exception:
return # the cold path legitimately attempted real IMAP
assert isinstance(result, dict)
assert result["emails"] == []
assert result["total"] == 0
assert not result.get("error")
assert not (result.get("sync") or {}).get("stale")
+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.
"""
import ast
import pytest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
@@ -75,37 +77,112 @@ def _scan_sites(tree: ast.AST) -> list[ast.Call]:
return sites
def _guard_lines(tree: ast.AST) -> list[int]:
"""Line numbers of calls that test whether procfs is present."""
lines = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
if name not in _GUARD_FUNCS:
continue
subject_is_procfs = (
name in {"has_procfs", "is_wsl"}
or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value))
or any(_is_procfs_root(arg) for arg in node.args)
)
if subject_is_procfs:
lines.append(node.lineno)
return lines
def _condition_requires_procfs(node: ast.AST, truth: bool) -> bool:
"""Whether this Boolean outcome proves a direct procfs check succeeded.
An ignored call, another function's check, or a check on only one possible
OR branch provides no proof. Keep unknown expressions conservative.
"""
if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not):
return _condition_requires_procfs(node.operand, not truth)
if isinstance(node, ast.BoolOp):
facts = [_condition_requires_procfs(value, truth) for value in node.values]
all_values_required = isinstance(node.op, ast.And) == truth
return any(facts) if all_values_required else all(facts)
if not isinstance(node, ast.Call):
return False
func = node.func
name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
subject_is_procfs = (
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:
"""Smallest function/module scope containing ``node``."""
best = tree
for candidate in ast.walk(tree):
if not isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
end = getattr(candidate, "end_lineno", None) or candidate.lineno
if candidate.lineno <= node.lineno <= end:
if best is tree or candidate.lineno > best.lineno:
best = candidate
return best
def _guarded_scan_ids(tree: ast.AST) -> set[int]:
"""Follow conditional/early-exit paths instead of comparing line numbers.
Each function starts with no procfs assumption. Facts from a loop or try
body cannot escape to code which may run without that body completing.
This intentionally requires direct checks; arbitrary aliases/call graphs
are not a proof that procfs exists at the enumeration site.
"""
scans = {id(site) for site in _scan_sites(tree)}
guarded = set()
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]:
@@ -139,12 +216,10 @@ def _collect_sites() -> tuple[list[str], list[str]]:
sites = _scan_sites(tree)
if not sites:
continue
guards = _guard_lines(tree)
guards = _guarded_scan_ids(tree)
for site in sites:
scope = _enclosing_scope(tree, site)
start = getattr(scope, "lineno", 0)
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)
else:
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():
_guarded, unguarded = _collect_sites()
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: "
+ ", ".join(unguarded)
)
@@ -176,3 +251,49 @@ def test_the_guard_detector_still_sees_the_known_scans():
f"root was renamed. 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)