| 1498 | self._create_script() |
| 1499 | |
| 1500 | def _create_script(self, script=None): |
| 1501 | # Create a file for subprocess script |
| 1502 | script = textwrap.dedent( |
| 1503 | f""" |
| 1504 | import socket |
| 1505 | import time |
| 1506 | |
| 1507 | def foo(): |
| 1508 | return bar() |
| 1509 | |
| 1510 | def bar(): |
| 1511 | return baz() |
| 1512 | |
| 1513 | def baz(): |
| 1514 | x = 1 |
| 1515 | # Trigger attach |
| 1516 | sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 1517 | sock.connect(('127.0.0.1', {self.port})) |
| 1518 | sock.close() |
| 1519 | count = 0 |
| 1520 | while x == 1 and count < 100: |
| 1521 | count += 1 |
| 1522 | time.sleep(0.1) |
| 1523 | return x |
| 1524 | |
| 1525 | result = foo() |
| 1526 | print(f"Function returned: {{result}}") |
| 1527 | """ |
| 1528 | ) |
| 1529 | |
| 1530 | self.script_path = TESTFN + "_connect_test.py" |
| 1531 | with open(self.script_path, 'w') as f: |
| 1532 | f.write(script) |
| 1533 | |
| 1534 | def tearDown(self): |
| 1535 | self.sock.close() |