Skip to content

Commit

Permalink
Add unit test;
Browse files Browse the repository at this point in the history
credits to @Novakov at syrusakbary#87
  • Loading branch information
Alex committed Jul 15, 2020
1 parent 14e7af8 commit e98f616
Showing 1 changed file with 81 additions and 2 deletions.
83 changes: 81 additions & 2 deletions tests/test_thread_safety.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,13 @@
from promise import Promise
from promise.dataloader import DataLoader
import threading
from queue import Queue
from sys import setswitchinterval
from threading import Thread, Barrier
from traceback import print_exc, format_exc

import pytest

from promise import Promise
from promise.dataloader import DataLoader


def test_promise_thread_safety():
Expand Down Expand Up @@ -113,3 +119,76 @@ def do():

assert assert_object['is_same_thread_1']
assert assert_object['is_same_thread_2']


@pytest.mark.parametrize('num_threads', [1])
@pytest.mark.parametrize('count', [1000])
def test_with_process_loop(num_threads, count):
"""
Start a Promise in one thread, but resolve it in another.
"""
items = Queue()
barrier = Barrier(num_threads)

force_stop = False
asserts = []
timeouts = []

def event_loop():
stop_count = num_threads
while True:
item = items.get()
if item[0] == 'STOP':
stop_count -= 1
if stop_count == 0:
break
if item[0] == 'ABORT':
break
if item[0] == 'ITEM':
(_, resolve, i) = item
resolve(i)

def worker():
nonlocal force_stop
barrier.wait()
# Force fast switching of threads, this is NOT used in real world case. However without this
# I was unable to reproduce the issue.
setswitchinterval(0.000001)
for i in range(0, count):
if force_stop:
break

def do(resolve, reject):
items.put(('ITEM', resolve, i))

p = Promise(do)
try:
p.get(timeout=1)
except AssertionError as e:
print("ASSERT", e)
print_exc()
force_stop = True
items.put(('ABORT',))
asserts.append(format_exc())
except Exception as e:
print("Timeout", e)
print_exc()
force_stop = True
items.put(('ABORT',))
timeouts.append(format_exc())

items.put(('STOP',))

loop_thread = Thread(target=event_loop)
loop_thread.start()

worker_threads = [Thread(target=worker) for i in range(0, num_threads)]
for t in worker_threads:
t.start()

loop_thread.join()
for t in worker_threads:
t.join()

assert asserts == []
assert timeouts == []

0 comments on commit e98f616

Please sign in to comment.