Spaces:
Running
Running
| """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() | |
| 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 | |
| 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(): | |
| 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() | |