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

1#!/usr/bin/env python3 

2 

3# Coverage 

4# pragma: windows cover file 

5 

6# Standard libraries 

7# from os import environ 

8from typing import List 

9 

10# Modules libraries 

11# environ['WEXPECT_SPAWN_CLASS'] = 'SpawnPipe' 

12from wexpect import EOF, spawn, TIMEOUT # pylint: disable=import-error 

13 

14# Components 

15from .base import Base 

16 

17# Wexpect class 

18class Wexpect(Base): 

19 

20 # Constructor 

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

22 

23 # Spawn command 

24 self._child = spawn(command) 

25 

26 # Acquire initial output 

27 assert self._child is not None 

28 try: 

29 self._child.expect(EOF, timeout=0) 

30 except TIMEOUT: # pragma: no cover 

31 pass 

32 

33 # Configure timeout 

34 self._child.timeout = 1 

35 

36 # Read 

37 def _read(self) -> bytes | List[bytes | str] | str: 

38 

39 # Acquire output 

40 assert self._child is not None 

41 try: 

42 output = self._child.read_nonblocking(size=1024) 

43 if self._child.before: # pragma: no cover 

44 output = self._child.before + output 

45 except EOF: 

46 output = self._child.before 

47 except TIMEOUT: # pragma: no cover 

48 output = self._child.before 

49 

50 # Handle output 

51 if output: # pragma: no cover 

52 

53 # Stream line 

54 assert isinstance(output, (bytes, str)) 

55 if '\n' in output: 

56 return output 

57 

58 # Adapt lines 

59 length = 80 

60 return [output[i:i + length] for i in range(0, len(output), length)] 

61 

62 # Empty output 

63 return [] 

64 

65 # Send 

66 def send(self, key: bytes | str) -> None: 

67 

68 # Send key 

69 assert self._child is not None 

70 try: 

71 self._child.send(key) 

72 except EOF: 

73 pass 

74 

75 # Status 

76 def status(self) -> int: 

77 

78 # Process terminated 

79 assert self._child is not None 

80 if self._child.exitstatus is None: 

81 return 0 if self._child.flag_eof else 1 

82 

83 # Process status 

84 return int(self._child.exitstatus) # pragma: no cover 

85 

86 # Terminate 

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

88 

89 # Terminate process 

90 if force: 

91 assert self._child is not None 

92 self._child.terminate() 

93 

94 # Wait process 

95 else: 

96 assert self._child is not None 

97 self._child.wait()