(self)
| 1992 | @unittest.skipIf(sys.platform == "win32", "Windows does not support SIGSTOP") |
| 1993 | @client_context.require_sync |
| 1994 | def test_sigstop_sigcont(self): |
| 1995 | test_dir = os.path.dirname(os.path.realpath(__file__)) |
| 1996 | script = os.path.join(test_dir, "sigstop_sigcont.py") |
| 1997 | p = subprocess.Popen( |
| 1998 | [sys.executable, script, client_context.uri], |
| 1999 | stdin=subprocess.PIPE, |
| 2000 | stdout=subprocess.PIPE, |
| 2001 | stderr=subprocess.STDOUT, |
| 2002 | ) |
| 2003 | self.addCleanup(p.wait, timeout=1) |
| 2004 | self.addCleanup(p.kill) |
| 2005 | time.sleep(1) |
| 2006 | # Stop the child, sleep for twice the streaming timeout |
| 2007 | # (heartbeatFrequencyMS + connectTimeoutMS), and restart. |
| 2008 | os.kill(p.pid, signal.SIGSTOP) |
| 2009 | time.sleep(2) |
| 2010 | os.kill(p.pid, signal.SIGCONT) |
| 2011 | time.sleep(0.5) |
| 2012 | # Tell the script to exit gracefully. |
| 2013 | outs, _ = p.communicate(input=b"q\n", timeout=10) |
| 2014 | self.assertTrue(outs) |
| 2015 | log_output = outs.decode("utf-8") |
| 2016 | self.assertIn("TEST STARTED", log_output) |
| 2017 | self.assertIn("ServerHeartbeatStartedEvent", log_output) |
| 2018 | self.assertIn("ServerHeartbeatSucceededEvent", log_output) |
| 2019 | self.assertIn("TEST COMPLETED", log_output) |
| 2020 | self.assertNotIn("ServerHeartbeatFailedEvent", log_output) |
| 2021 | |
| 2022 | def _test_handshake(self, env_vars, expected_env): |
| 2023 | with patch.dict("os.environ", env_vars): |
nothing calls this directly
no test coverage detected