Function recv_wait

Lib/test/test__interpchannels.py:31–39 · view source on GitHub ↗

(cid)

Source from the content-addressed store, hash-verified

29	# helpers
30
31	def recv_wait(cid):
32	while True:
33	try:
34	obj, unboundop = _channels.recv(cid)
35	except _channels.ChannelEmptyError:
36	time.sleep(0.1)
37	else:
38	assert unboundop is None, repr(unboundop)
39	return obj
40
41
42	def recv_nowait(cid, *args, unbound=False):

fMethod · 0.85

test_send_recv_different_threadsMethod · 0.85

test_send_recv_different_interpreters_and_threadsMethod · 0.85

recvMethod · 0.45

sleepMethod · 0.45

no test coverage detected

searching dependent graphs…