Coverage for /Users/ajo/work/jumpstarter/jumpstarter/packages/jumpstarter-driver-network/jumpstarter_driver_network/adapters/pexpect.py: 54%

1import socket 

2from contextlib import contextmanager 

3 

4from pexpect.fdpexpect import fdspawn 

5 

6from .portforward import TcpPortforwardAdapter 

7from jumpstarter.client import DriverClient 

8 

9 

10@contextmanager 

11def PexpectAdapter(*, client: DriverClient, method: str = "connect"): 

12 with TcpPortforwardAdapter(client=client, method=method) as addr: 

13 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

14 sock.connect(addr) 

15 

16 try: 

17 yield fdspawn(sock) 

18 finally: 

19 sock.close()