Spaces:
Running
Running
File size: 2,316 Bytes
436ab72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """No external API calls or engine dependencies: verify the shared-state boundary."""
import ast
import inspect
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Barrier, Lock
from time import sleep
import pytest
from calculation_lock import serialize_calculation
def test_mixed_chart_requests_do_not_interleave_shared_settings():
start = Barrier(8)
state = {"profile": None, "active": 0, "max_active": 0}
stats_lock = Lock()
@serialize_calculation
def western(profile):
state["profile"] = profile
with stats_lock:
state["active"] += 1
state["max_active"] = max(state["max_active"], state["active"])
sleep(0.003)
assert state["profile"] == profile
with stats_lock:
state["active"] -= 1
return profile
@serialize_calculation
def vedic(profile):
# Reentrant calls share the same lock, including engine helpers.
return western(profile)
def request(index):
start.wait()
return (western if index % 2 else vedic)(str(index))
with ThreadPoolExecutor(max_workers=8) as pool:
assert list(pool.map(request, range(8))) == list(map(str, range(8)))
assert state["max_active"] == 1
def test_errors_release_the_calculation_lock_and_preserve_signature():
@serialize_calculation
def calculate(value: str) -> str:
if value == "invalid":
raise ValueError("invalid")
return value
with pytest.raises(ValueError):
calculate("invalid")
assert str(inspect.signature(calculate)) == "(value: str) -> str"
with ThreadPoolExecutor(max_workers=1) as pool:
assert pool.submit(calculate, "next request").result(timeout=1) == "next request"
def test_every_shared_engine_entry_point_is_serialized():
tree = ast.parse(Path(__file__).with_name("app.py").read_text())
routes = {node.name: node for node in tree.body if isinstance(node, ast.FunctionDef)}
for name in ["calculate_chart", "generate_chart_svg", "calculate_vedic_chart"]:
assert any(isinstance(d, ast.Name) and d.id == "serialize_calculation" for d in routes[name].decorator_list)
assert "COPY app.py calculation_lock.py ./" in Path(__file__).with_name("Dockerfile").read_text()
|