Question: I need to add a threaded test to the enum test module [1] -- is there anything extra I need to worry about besides the test itself? Setting or resetting or using a tool library, etc? -- ~Ethan~ [1] The test to be added: def test_unique_composite(self): # override __eq__ to be identity only class TestFlag(IntFlag): one = auto() two = auto() three = auto() four = auto() five = auto() six = auto() seven = auto() eight = auto() def __eq__(self, other): return self is other def __hash__(self): return hash(self._value_) # have multiple threads competing to complete the composite members seen = set() failed = False def cycle_enum(): nonlocal failed try: for i in range(256): seen.add(TestFlag(i)) except (Exception, RuntimeError): failed = True threads = [] for i in range(8): threads.append(threading.Thread(target=cycle_enum)) for t in threads: t.start() for t in threads: t.join() # check that only 248 members were created self.assertFalse( failed, 'at least one thread failed while creating composite members') self.assertEqual(256, len(seen), 'too many composite members created')
There is @support.reap_thread which can help. Victor Le dim. 22 janv. 2017 à 21:04, Ethan Furman <ethan@stoneleaf.us> a écrit :
Question: I need to add a threaded test to the enum test module [1] -- is there anything extra I
need to worry about besides the test itself? Setting or resetting or using a tool library, etc?
--
~Ethan~
[1] The test to be added:
def test_unique_composite(self):
# override __eq__ to be identity only
class TestFlag(IntFlag):
one = auto()
two = auto()
three = auto()
four = auto()
five = auto()
six = auto()
seven = auto()
eight = auto()
def __eq__(self, other):
return self is other
def __hash__(self):
return hash(self._value_)
# have multiple threads competing to complete the composite members
seen = set()
failed = False
def cycle_enum():
nonlocal failed
try:
for i in range(256):
seen.add(TestFlag(i))
except (Exception, RuntimeError):
failed = True
threads = []
for i in range(8):
threads.append(threading.Thread(target=cycle_enum))
for t in threads:
t.start()
for t in threads:
t.join()
# check that only 248 members were created
self.assertFalse(
failed,
'at least one thread failed while creating composite members')
self.assertEqual(256, len(seen), 'too many composite members created')
_______________________________________________
Python-Dev mailing list
Python-Dev@python.org
https://mail.python.org/mailman/listinfo/python-dev
Unsubscribe: https://mail.python.org/mailman/options/python-dev/victor.stinner%40gmail.co...
Le dim. 22 janv. 2017 à 21:04, Ethan Furman <ethan@stoneleaf.us> a écrit :
Question: I need to add a threaded test to the enum test module [1] -- is there anything extra I need to worry about besides the test itself? Setting or resetting or using a tool library, etc?
threads = [] for i in range(8): threads.append(threading.Thread(target=cycle_enum)) for t in threads: t.start() for t in threads: t.join()
On 22 January 2017 at 20:17, Victor Stinner <victor.stinner@gmail.com> wrote:
There is @support.reap_thread which can help.
As I understand, @reap_threads basically does a join() on each background thread, with a total timeout of 1 s. So since your test is unlikely to fail between starting threads and joining them, I don’t think you need to use @reap_threads.
On Sun, Jan 22, 2017 at 2:39 PM, Martin Panter <vadmium+py@gmail.com> wrote:
As I understand, @reap_threads basically does a join() on each background thread, with a total timeout of 1 s. So since your test is unlikely to fail between starting threads and joining them, I don’t think you need to use @reap_threads.
reap_threads is meant as a failsafe, in case your test case doesn't clean up after itself properly. Most of the time, reap_threads shouldn't actually *do* anything. -- Zach
On Sun, Jan 22, 2017 at 2:02 PM, Ethan Furman <ethan@stoneleaf.us> wrote:
Question: I need to add a threaded test to the enum test module [1] -- is there anything extra I need to worry about besides the test itself? Setting or resetting or using a tool library, etc?
As far as I know, the only extras to worry about are to use the support.reap_threads decorator, and be sure to skip the test if threading is not available. Search through the tests for 'reap_threads' if you want other examples of how to handle threaded tests. -- Zach
On 01/22/2017 12:02 PM, Ethan Furman wrote:
Question: I need to add a threaded test to the enum test module [1] -- is there anything extra I need to worry about besides the test itself? Setting or resetting or using a tool library, etc?
Thanks everyone. @support.reap_threads and skipping it is. -- ~Ethan~
On 22.01.17 22:02, Ethan Furman wrote:
Question: I need to add a threaded test to the enum test module [1] -- is there anything extra I need to worry about besides the test itself? Setting or resetting or using a tool library, etc?
You can use the test.support.start_threads() context manager.
participants (5)
-
Ethan Furman -
Martin Panter -
Serhiy Storchaka -
Victor Stinner -
Zachary Ware