with pipe(duplex=True) as (cend, pend):
- p = start_process(duplchild_time_sync
_child, args=(cend, ))
+ p = start_process(duplchild_time_sync, args=(cend, ))
assert pend.get() == "ACK"
assert abs(ptime - ctime) < 0.01
+ def test_circular_forward(self):
+ with pipe(True) as (p11, p12):
+ with pipe(True) as (p21, p22):
+ with pipe(True) as (p31, p32):
+ # Spawn two forwarders.
+ forwarder1 = start_process(
+ duplchild_circular_forward, (p12, p21))
+ forwarder2 = start_process(
+ duplchild_circular_forward, (p22, p31))
+ # Make sure that only 2 of 6 handles are usable.
+ for h in (p12, p21, p22, p31):
+ with raises(GIPCClosed):
+ # Send messages on their journey through children.
+ assert p32.get() == "BABUUUZ"
+def duplchild_circular_forward(receiver, sender):
assert cend.get() == "SYN"