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

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from abc import ABC, abstractmethod 

5from typing import Dict, Iterator, List, NamedTuple 

6 

7# Components 

8from ..models.images import Entrypoint 

9from ..types.volumes import Volumes 

10 

11# Commands type 

12Commands = List[str] 

13 

14# ContainerName type 

15ContainerName = str 

16 

17# _ExecResult type 

18class _ExecResult(NamedTuple): 

19 

20 # Properties 

21 exit_code: int 

22 output: str 

23 

24# _LogsResult type 

25_LogsResult = Iterator[bytes] 

26 

27# Base engine class, pylint: disable=unused-argument 

28class BaseEngine(ABC): # pragma: no cover 

29 

30 # Command exec 

31 @abstractmethod 

32 def cmd_exec(self) -> str: 

33 return '' 

34 

35 # Container 

36 @property 

37 @abstractmethod 

38 def container(self) -> ContainerName: 

39 return '' 

40 

41 # Exec 

42 @abstractmethod 

43 def exec(self, commands: Commands) -> _ExecResult: 

44 return _ExecResult( 

45 0, 

46 '', 

47 ) 

48 

49 # Get 

50 @abstractmethod 

51 def get(self, image: str) -> None: 

52 pass 

53 

54 # Logs 

55 @abstractmethod 

56 def logs(self) -> _LogsResult: 

57 return iter() # type: ignore[call-overload,no-any-return] 

58 

59 # Pull 

60 @abstractmethod 

61 def pull( 

62 self, 

63 image: str, 

64 force: bool = False, 

65 ) -> None: 

66 pass 

67 

68 # Remove 

69 @abstractmethod 

70 def remove(self) -> None: 

71 pass 

72 

73 # Remove image 

74 @abstractmethod 

75 def rmi(self, image: str) -> None: 

76 pass 

77 

78 # Run, pylint: disable=too-many-arguments,too-many-positional-arguments 

79 @abstractmethod 

80 def run( 

81 self, 

82 image: str, 

83 commands: Commands, 

84 entrypoint: Entrypoint, 

85 variables: Dict[str, str], 

86 network: str, 

87 option_privileged: bool, 

88 option_sockets: bool, 

89 services: bool, 

90 volumes: Volumes, 

91 directory: str, 

92 temp_folder: str, 

93 ) -> None: 

94 pass 

95 

96 # Stop 

97 @abstractmethod 

98 def stop(self, timeout: int) -> None: 

99 pass 

100 

101 # Supports 

102 @abstractmethod 

103 def supports(self, binary: str) -> bool: 

104 return False 

105 

106 # Wait 

107 @abstractmethod 

108 def wait(self) -> bool: 

109 return False