MCPcopy Index your code
hub / github.com/RustPython/RustPython / test_stress_delivery_simultaneous

Method test_stress_delivery_simultaneous

Lib/test/test_signal.py:1315–1346  ·  view source on GitHub ↗

This test uses simultaneous signal handlers.

(self)

Source from the content-addressed store, hash-verified

1313 @unittest.skipUnless(hasattr(signal, "setitimer"),
1314 "test needs setitimer()")
1315 def test_stress_delivery_simultaneous(self):
1316 """
1317 This test uses simultaneous signal handlers.
1318 """
1319 N = self.decide_itimer_count()
1320 sigs = []
1321
1322 def handler(signum, frame):
1323 sigs.append(signum)
1324
1325 # On Android, SIGUSR1 is unreliable when used in close proximity to
1326 # another signal – see Android/testbed/app/src/main/python/main.py.
1327 # So we use a different signal.
1328 self.setsig(signal.SIGUSR2, handler)
1329 self.setsig(signal.SIGALRM, handler) # for ITIMER_REAL
1330
1331 expected_sigs = 0
1332 while expected_sigs < N:
1333 # Hopefully the SIGALRM will be received somewhere during
1334 # initial processing of SIGUSR2.
1335 signal.setitimer(signal.ITIMER_REAL, 1e-6 + random.random() * 1e-5)
1336 os.kill(os.getpid(), signal.SIGUSR2)
1337
1338 expected_sigs += 2
1339 # Wait for handlers to run to avoid signal coalescing
1340 for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
1341 if len(sigs) >= expected_sigs:
1342 break
1343
1344 # All ITIMER_REAL signals should have been delivered to the
1345 # Python handler
1346 self.assertEqual(len(sigs), N, "Some signals were lost")
1347
1348 @support.requires_gil_enabled("gh-121065: test is flaky on free-threaded build")
1349 @unittest.skipIf(is_apple, "crashes due to system bug (FB13453490)")

Callers

nothing calls this directly

Calls 6

decide_itimer_countMethod · 0.95
setsigMethod · 0.95
lenFunction · 0.85
randomMethod · 0.45
killMethod · 0.45
assertEqualMethod · 0.45

Tested by

no test coverage detected