From 6d8fba03201a73bf4d208d173ceec112f5ca85bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A0=20Casaj=C3=BAs?= Date: Thu, 27 Oct 2022 14:04:03 +0200 Subject: [PATCH] Added too many exceptions test (#1378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Adrià Casajús --- tests/test_prarallel_limiter.py | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/tests/test_prarallel_limiter.py b/tests/test_prarallel_limiter.py index 05133cd1..84b93e04 100644 --- a/tests/test_prarallel_limiter.py +++ b/tests/test_prarallel_limiter.py @@ -1,3 +1,8 @@ +import threading +import time +from typing import Optional + +import werkzeug.exceptions from flask_login import login_user from app.parallel_limiter import _InnerLock @@ -11,3 +16,27 @@ def test_parallel_limiter(flask_app): pl = _InnerLock("test", max_wait_secs=1) for loop_id in range(10): assert pl(lambda x: x)(loop_id) == loop_id + + +def sleep_thread(pl: _InnerLock, sem: Optional[threading.Semaphore] = None): + if sem is not None: + sem.release() + pl(time.sleep)(1) + + +def test_too_many_requests(flask_app): + user = create_new_user() + with flask_app.test_request_context(): + login_user(user) + sem = threading.Semaphore(0) + pl = _InnerLock("test", max_wait_secs=5) + t = threading.Thread(target=sleep_thread, args=(pl, sem)) + t.daemon = True + t.start() + sem.acquire() + try: + got_exception = False + pl(sleep_thread)(pl) + except werkzeug.exceptions.TooManyRequests: + got_exception = True + assert got_exception