Coverage for src/engines/pexpect_popen.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Coverage 

4# pragma: windows cover file 

5 

6# Standard libraries 

7from signal import SIGTERM 

8 

9# Modules libraries 

10from pexpect import EOF, TIMEOUT # pylint: disable=import-error 

11from pexpect.popen_spawn import PopenSpawn # pylint: disable=import-error 

12 

13# Components 

14from .base import Base 

15 

16# PexpectPopen class 

17class PexpectPopen(Base): 

18 

19 # Constructor 

20 def __init__(self, command: str) -> None: # pylint: disable=super-init-not-called 

21 

22 # Spawn command 

23 self._child = PopenSpawn(command) 

24 

25 # Read 

26 def _read(self) -> bytes | str: 

27 

28 # Acquire output, pylint: disable=duplicate-code 

29 assert self._child is not None 

30 try: 

31 result = self._child.read_nonblocking(size=1024, timeout=1) 

32 assert isinstance(result, (bytes, str)) 

33 return result 

34 except (EOF, TIMEOUT): 

35 pass 

36 

37 # Empty output 

38 return '' 

39 

40 # Is alive 

41 def isalive(self) -> bool: 

42 

43 # Result 

44 assert self._child is not None 

45 return not self._child.terminated 

46 

47 # Terminate 

48 def terminate(self, force: bool = False) -> None: 

49 

50 # Terminate process 

51 assert self._child is not None 

52 if force: 

53 try: 

54 self._child.kill(sig=SIGTERM) 

55 except PermissionError: # pragma: no cover 

56 pass 

57 

58 # Wait process 

59 self._child.wait()