forked from retoor/devplacepy
retoor
9cfaddfc40
The creation and run quotas were checked and then acted on, so two concurrent create_task calls or two schedulers could both pass the check and overshoot the limit. Both are now a single conditional INSERT decided on the driver rowcount: reserve_run takes a run slot after the claim and releases the claim by deferring when the quota is spent, and insert_task_within_quota does the same for the task row itself. Racing twelve and sixteen processes now yields exactly the limit. The atomic insert names its columns, and dataset skips a None valued key when it creates a table lazily, so the store declares the full task column set up front. Both the column and index ensures now tolerate a concurrent duplicate, since several processes build a store at once and SQLite DDL is not idempotent. Adds the quota, task-run context, guard, store and scheduler test suites, and documents the chokepoints and the unhackable task-run flag.
106 lines
2.7 KiB
Python
106 lines
2.7 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
import asyncio
|
|
|
|
from devplacepy.database import invalidate_admins_cache
|
|
from devplacepy.services.devii.tasks.context import in_task_run, task_run_scope
|
|
from devplacepy.services.devii.tasks.guards import nesting_allowed
|
|
from devplacepy.utils import generate_uid
|
|
|
|
SIGNUP_AFTER_PRIMARY_ADMIN = "2099-01-01T00:00:00"
|
|
from tests.conftest import run_async
|
|
|
|
|
|
def _account(local_db, role):
|
|
uid = generate_uid()
|
|
local_db["users"].insert(
|
|
{
|
|
"uid": uid,
|
|
"username": f"ctx-{uid[-10:]}",
|
|
"role": role,
|
|
"deleted_at": None,
|
|
"created_at": SIGNUP_AFTER_PRIMARY_ADMIN,
|
|
}
|
|
)
|
|
invalidate_admins_cache()
|
|
return uid
|
|
|
|
|
|
def test_outside_a_task_run_the_flag_is_off():
|
|
assert in_task_run() is False
|
|
|
|
|
|
def test_scope_sets_and_restores_the_flag():
|
|
assert in_task_run() is False
|
|
with task_run_scope():
|
|
assert in_task_run() is True
|
|
assert in_task_run() is False
|
|
|
|
|
|
def test_scope_restores_the_flag_after_an_exception():
|
|
try:
|
|
with task_run_scope():
|
|
raise RuntimeError("boom")
|
|
except RuntimeError:
|
|
pass
|
|
assert in_task_run() is False
|
|
|
|
|
|
def test_nested_scopes_restore_correctly():
|
|
with task_run_scope():
|
|
with task_run_scope():
|
|
assert in_task_run() is True
|
|
assert in_task_run() is True
|
|
assert in_task_run() is False
|
|
|
|
|
|
def test_the_flag_reaches_work_spawned_inside_the_run():
|
|
seen = []
|
|
|
|
async def inner():
|
|
seen.append(in_task_run())
|
|
|
|
async def run():
|
|
with task_run_scope():
|
|
await inner()
|
|
task = asyncio.create_task(inner())
|
|
await task
|
|
|
|
run_async(run())
|
|
assert seen == [True, True]
|
|
|
|
|
|
def test_a_concurrent_turn_outside_the_run_never_sees_the_flag():
|
|
seen = {}
|
|
|
|
async def scheduled(gate):
|
|
with task_run_scope():
|
|
seen["inside"] = in_task_run()
|
|
gate.set()
|
|
await asyncio.sleep(0.05)
|
|
seen["inside_after"] = in_task_run()
|
|
|
|
async def interactive(gate):
|
|
await gate.wait()
|
|
seen["outside"] = in_task_run()
|
|
|
|
async def run():
|
|
gate = asyncio.Event()
|
|
await asyncio.gather(scheduled(gate), interactive(gate))
|
|
|
|
run_async(run())
|
|
assert seen["inside"] is True
|
|
assert seen["inside_after"] is True
|
|
assert seen["outside"] is False
|
|
|
|
|
|
def test_members_may_not_nest_but_administrators_may(local_db):
|
|
member = _account(local_db, "Member")
|
|
admin = _account(local_db, "Admin")
|
|
assert nesting_allowed(member) is True
|
|
assert nesting_allowed(admin) is True
|
|
with task_run_scope():
|
|
assert nesting_allowed(member) is False
|
|
assert nesting_allowed(admin) is True
|
|
assert nesting_allowed(member) is True
|