refine testing and code
This commit is contained in:
parent
9e05a7d1eb
commit
2b45ace3ba
@ -30,7 +30,7 @@ METADATA_TOKEN_KEY = "devicetoken"
|
||||
|
||||
class Notifier:
|
||||
CONNECTION_TIMEOUT = 60.0 # seconds
|
||||
NOTIFICATION_RETRY_DELAY = 8.0 # seconds
|
||||
NOTIFICATION_RETRY_DELAY = 8.0 # seconds, with exponential backoff
|
||||
MAX_NUMBER_OF_TRIES = 6
|
||||
# exponential backoff means we try for 8^5 seconds, approximately 10 hours
|
||||
|
||||
@ -69,15 +69,17 @@ class Notifier:
|
||||
self.add_token_for_retry(token)
|
||||
|
||||
def add_token_for_retry(self, token, numtries=0):
|
||||
# backup exponentially with number of retries
|
||||
when = time.time() + pow(self.NOTIFICATION_RETRY_DELAY, numtries)
|
||||
when = time.time()
|
||||
if numtries > 0:
|
||||
# backup exponentially with number of retries
|
||||
when += pow(self.NOTIFICATION_RETRY_DELAY, numtries)
|
||||
self.retry_queues[numtries].put((when, token))
|
||||
|
||||
def start_notification_threads(self):
|
||||
for token_path in self.notification_dir.iterdir():
|
||||
self.add_token_for_retry(token_path.name)
|
||||
|
||||
# we start a thread for each retry-queue
|
||||
# we start a thread for each retry-queue bucket
|
||||
for numtries in range(len(self.retry_queues)):
|
||||
t = Thread(target=self.thread_retry_loop, args=(numtries,))
|
||||
t.setDaemon(True)
|
||||
@ -86,12 +88,15 @@ class Notifier:
|
||||
def thread_retry_loop(self, numtries):
|
||||
requests_session = requests.Session()
|
||||
while True:
|
||||
retry_queue = self.retry_queues[numtries]
|
||||
when, token = retry_queue.get()
|
||||
wait_time = when - time.time()
|
||||
if wait_time > 0:
|
||||
time.sleep(wait_time)
|
||||
self.notify_one(requests_session, token, numtries)
|
||||
self.thread_retry_one(requests_session, numtries)
|
||||
|
||||
def thread_retry_one(self, requests_session, numtries):
|
||||
retry_queue = self.retry_queues[numtries]
|
||||
when, token = retry_queue.get()
|
||||
wait_time = when - time.time()
|
||||
if wait_time > 0:
|
||||
time.sleep(wait_time)
|
||||
self.notify_one(requests_session, token, numtries)
|
||||
|
||||
def notify_one(self, requests_session, token, numtries=0):
|
||||
try:
|
||||
@ -123,7 +128,9 @@ class Notifier:
|
||||
if numtries < self.MAX_NUMBER_OF_TRIES:
|
||||
self.add_token_for_retry(token, numtries=numtries)
|
||||
else:
|
||||
logging.warning("giving up on token after %d tries: %r", numtries - 1, token)
|
||||
logging.warning(
|
||||
"giving up on token after %d tries: %r", numtries - 1, token
|
||||
)
|
||||
|
||||
|
||||
def handle_dovecot_protocol(rfile, wfile, notifier):
|
||||
|
@ -31,6 +31,24 @@ def token():
|
||||
return "01234"
|
||||
|
||||
|
||||
def get_mocked_requests(statuslist):
|
||||
class ReqMock:
|
||||
requests = []
|
||||
|
||||
def post(self, url, data, timeout):
|
||||
self.requests.append((url, data, timeout))
|
||||
res = statuslist.pop(0)
|
||||
if isinstance(res, Exception):
|
||||
raise res
|
||||
|
||||
class Result:
|
||||
status_code = res
|
||||
|
||||
return Result()
|
||||
|
||||
return ReqMock()
|
||||
|
||||
|
||||
def test_notifier_persistence(tmp_path, testaddr, testaddr2):
|
||||
notifier1 = Notifier(tmp_path)
|
||||
notifier2 = Notifier(tmp_path)
|
||||
@ -152,82 +170,50 @@ def test_handle_dovecot_protocol_iterate(notifier):
|
||||
|
||||
|
||||
def test_notifier_thread_firstrun(notifier, testaddr):
|
||||
requests = []
|
||||
|
||||
class ReqMock:
|
||||
def post(self, url, data, timeout):
|
||||
requests.append((url, data, timeout))
|
||||
|
||||
class Result:
|
||||
status_code = 200
|
||||
|
||||
return Result()
|
||||
|
||||
notifier.add_token(testaddr, "01234")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
when, token = notifier.retry_queues[0].get()
|
||||
notifier.notify_one(ReqMock(), token, numtries=0)
|
||||
url, data, timeout = requests[0]
|
||||
assert data == "01234"
|
||||
assert notifier.get_tokens(testaddr) == ["01234"]
|
||||
|
||||
|
||||
def get_mocked_requests(statuslist):
|
||||
class ReqMock:
|
||||
requests = []
|
||||
|
||||
def post(self, url, data, timeout):
|
||||
self.requests.append((url, data, timeout))
|
||||
res = statuslist.pop(0)
|
||||
if isinstance(res, Exception):
|
||||
raise res
|
||||
|
||||
class Result:
|
||||
status_code = res
|
||||
|
||||
return Result()
|
||||
|
||||
return ReqMock()
|
||||
|
||||
|
||||
def test_notifier_thread_run(notifier, testaddr):
|
||||
notifier.add_token(testaddr, "01234")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
token = notifier.retry_queues[0].get()[1]
|
||||
|
||||
reqmock = get_mocked_requests([200])
|
||||
notifier.notify_one(reqmock, token=token, numtries=0)
|
||||
notifier.add_token(testaddr, "01234")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
url, data, timeout = reqmock.requests[0]
|
||||
assert data == "01234"
|
||||
assert notifier.get_tokens(testaddr) == ["01234"]
|
||||
|
||||
|
||||
def test_notifier_thread_connection_exceptions(notifier, testaddr):
|
||||
reqmock = get_mocked_requests([requests.exceptions.RequestException()])
|
||||
def test_notifier_thread_run(notifier, testaddr):
|
||||
notifier.add_token(testaddr, "01234")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
token = notifier.retry_queues[0].get()[1]
|
||||
reqmock = get_mocked_requests([200])
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
url, data, timeout = reqmock.requests[0]
|
||||
assert data == "01234"
|
||||
assert notifier.get_tokens(testaddr) == ["01234"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", [requests.exceptions.RequestException(), 404, 500])
|
||||
def test_notifier_thread_connection_failures(notifier, testaddr, status):
|
||||
notifier.add_token(testaddr, "01234")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
reqmock = get_mocked_requests([status])
|
||||
notifier.NOTIFICATION_RETRY_DELAY = 0.1
|
||||
notifier.notify_one(reqmock, token)
|
||||
assert notifier.retry_queues[1].get()[1] == token
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
assert notifier.retry_queues[1].get()[1] == "01234"
|
||||
assert notifier.retry_queues[0].qsize() == 0
|
||||
|
||||
|
||||
def test_multi_device_notifier(notifier, testaddr):
|
||||
notifier.add_token(testaddr, "01234")
|
||||
notifier.add_token(testaddr, "56789")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
token1 = notifier.retry_queues[0].get()[1]
|
||||
token2 = notifier.retry_queues[0].get()[1]
|
||||
|
||||
reqmock = get_mocked_requests([200, 200])
|
||||
notifier.notify_one(reqmock, token1)
|
||||
notifier.notify_one(reqmock, token2)
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
assert notifier.retry_queues[0].qsize() == 0
|
||||
assert notifier.retry_queues[1].qsize() == 0
|
||||
url, data, timeout = reqmock.requests[0]
|
||||
assert data == token1
|
||||
assert data == "01234"
|
||||
url, data, timeout = reqmock.requests[1]
|
||||
assert data == token2
|
||||
assert data == "56789"
|
||||
assert notifier.get_tokens(testaddr) == ["01234", "56789"]
|
||||
|
||||
|
||||
@ -236,12 +222,12 @@ def test_notifier_thread_run_gone_removes_token(notifier, testaddr):
|
||||
notifier.add_token(testaddr, "45678")
|
||||
notifier.new_message_for_addr(testaddr)
|
||||
reqmock = get_mocked_requests([410, 200])
|
||||
for token in notifier.get_tokens(testaddr):
|
||||
notifier.notify_one(reqmock, token, numtries=0)
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
notifier.thread_retry_one(reqmock, numtries=0)
|
||||
url, data, timeout = reqmock.requests[0]
|
||||
assert data == "01234"
|
||||
url, data, timeout = reqmock.requests[1]
|
||||
assert data == "45678"
|
||||
assert notifier.get_tokens(testaddr) == ["45678"]
|
||||
assert notifier.retry_queues[0].qsize() == 2
|
||||
assert notifier.retry_queues[0].qsize() == 0
|
||||
assert notifier.retry_queues[1].qsize() == 0
|
||||
|
Loading…
Reference in New Issue
Block a user