perl_d
1cd8102b69
Docs build and publish / docker (push) Successful in 19s
CI / lint (push) Successful in 45s
CI / test (3.12) (push) Canceled after 52s
CI / test (3.13) (push) Canceled after 48s
CI / test (3.14) (push) Canceled after 47s
CI / test-with-beamline-plugins (pxi_bec) (push) Canceled after 43s
CI / test-with-beamline-plugins (pxii_bec) (push) Canceled after 42s
CI / test-with-beamline-plugins (pxiii_bec) (push) Canceled after 38s
CI / test-with-coverage (push) Canceled after 37s
CI / coverage-analysis (push) Canceled after 0s
Build and Publish / release (push) Successful in 19s
74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
from unittest.mock import patch
|
|
|
|
import fakeredis
|
|
import pytest
|
|
from redis.lock import Lock as RedisLock
|
|
|
|
LOCK_NAME = "SIMULATED:hardware_busy_lock"
|
|
AUTH = {"Authorization": "Bearer fake-token"}
|
|
|
|
THREAD_LOCAL = False
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_redis():
|
|
return fakeredis.FakeRedis(decode_responses=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def hw_lock(server_module, client, fake_redis, monkeypatch):
|
|
"""This worker's lock.
|
|
|
|
Depends on ``client`` so that TestClient's lifespan (which overwrites
|
|
``hardware_busy_lock``) has already run by the time we patch.
|
|
"""
|
|
lock = RedisLock(fake_redis, name=LOCK_NAME, thread_local=THREAD_LOCAL)
|
|
monkeypatch.setattr(server_module, "hardware_busy_lock", lock)
|
|
return lock
|
|
|
|
|
|
@pytest.fixture
|
|
def other_worker(fake_redis):
|
|
"""A second uvicorn worker's view of the same lock in Redis."""
|
|
return RedisLock(fake_redis, name=LOCK_NAME, thread_local=THREAD_LOCAL)
|
|
|
|
|
|
def test_endpoint_acquires_and_releases_the_lock(client, hw_lock, other_worker):
|
|
with patch("aare.daq.auth.check_jwt_rw"), patch("aare.daq.server.daq"):
|
|
response = client.put("/beamline/omega?val=10.5", headers=AUTH)
|
|
|
|
assert response.status_code == 200
|
|
# Released on the way out, so a second worker can take it.
|
|
assert not hw_lock.owned()
|
|
assert other_worker.acquire(blocking=False)
|
|
|
|
|
|
def test_endpoint_is_503_while_another_worker_holds_the_lock(client, hw_lock, other_worker):
|
|
assert other_worker.acquire(blocking=False)
|
|
|
|
with patch("aare.daq.auth.check_jwt_rw"), patch("aare.daq.server.daq"):
|
|
response = client.put("/beamline/omega?val=10.5", headers=AUTH)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json()["exception_class"] == "BeamlineBusyException"
|
|
# The rejected request must not have stolen or released the other worker's lock.
|
|
assert other_worker.owned()
|
|
assert not hw_lock.owned()
|
|
|
|
|
|
async def test_lock_hw_is_reentrant_for_the_owning_worker(server_module, hw_lock):
|
|
assert hw_lock.acquire(blocking=False)
|
|
|
|
async with server_module._lock_hw():
|
|
assert hw_lock.owned()
|
|
|
|
assert hw_lock.owned(), "re-entrant exit released a lock it did not acquire"
|
|
|
|
|
|
async def test_lock_is_released_when_the_handler_raises(server_module, hw_lock):
|
|
with pytest.raises(RuntimeError):
|
|
async with server_module._lock_hw():
|
|
raise RuntimeError("handler blew up")
|
|
|
|
assert not hw_lock.owned(), "hardware lock leaked after an exception"
|