Coverage for src/types/volumes.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from typing import Dict, List, NamedTuple 

5 

6# Components 

7from ..system.platform import Platform 

8from ..types.paths import Paths 

9 

10# Volume type 

11class Volume(NamedTuple): 

12 

13 # Properties 

14 bind: str 

15 mode: str 

16 source: str 

17 

18 # Stringify 

19 def stringify(self) -> str: # pragma: podman cover 

20 

21 # Variables 

22 options = '' 

23 

24 # Extract options 

25 if self.mode == 'ro': 

26 options += ':ro' 

27 elif self.mode == 'rw': 

28 options += ':rw' 

29 

30 # Result 

31 return f'{self.source}:{self.bind}{options}' 

32 

33# Volumes class 

34class Volumes: 

35 

36 # Constants 

37 LOCAL_FLAG: str = '.local:' 

38 

39 # Members 

40 __volumes: Dict[str, Volume] 

41 

42 # Constructor 

43 def __init__(self) -> None: 

44 

45 # Initialize members 

46 self.__volumes = {} 

47 

48 # Add 

49 def add( 

50 self, 

51 source: str, 

52 target: str, 

53 mode: str, 

54 override: bool, 

55 ) -> None: 

56 

57 # Handle overrides 

58 if target in [volume.bind for _, volume in self.__volumes.items()]: 

59 if not override: 

60 return 

61 

62 # Detect duplicated volumes 

63 duplicates = [ 

64 key for key, volume in self.__volumes.items() 

65 if volume.source == source and volume.bind == target 

66 ] 

67 if duplicates: 

68 self.__volumes.pop(duplicates[0]) 

69 

70 # Adapt source to allow duplicates 

71 while source in self.__volumes: 

72 if Platform.IS_WINDOWS: # pragma: no cover 

73 source = f'{source}{Platform.PATH_SEPARATOR}.' 

74 else: 

75 source = f'{Platform.PATH_SEPARATOR}.{source}' 

76 

77 # Add volume binding 

78 self.__volumes[source] = Volume( 

79 bind=target, 

80 mode=mode, 

81 source=source, 

82 ) 

83 

84 # Get 

85 def get(self) -> Dict[str, Volume]: # pragma: podman cover 

86 

87 # Result 

88 return self.__volumes 

89 

90 # Flatten 

91 def flatten(self) -> Dict[str, Dict[str, str]]: # pragma: docker cover 

92 

93 # Variables 

94 volumes: Dict[str, Dict[str, str]] = {} 

95 

96 # Flatten volumes 

97 for source, volume in self.__volumes.items(): 

98 volumes[source] = dict(volume._asdict().items()) 

99 

100 # Result 

101 return volumes 

102 

103 # Parse 

104 @staticmethod 

105 def parse(volume: str) -> List[str]: 

106 

107 # Invalid volume 

108 if not volume: 

109 raise ValueError('Empty volume parameter cannot be parsed') 

110 

111 # Relative volume 

112 if 1 <= len(volume) <= 2: 

113 return [volume] 

114 

115 # Variables 

116 volume_node: str = '' 

117 volume_nodes: List[str] = [] 

118 

119 # Iterate through volume 

120 for char in f'{volume}\x00': 

121 

122 # Detect Windows drive 

123 if char == ':' and len(volume_node) == 1 and volume_node[0].isalpha(): 

124 volume_node += char # pragma: no cover 

125 

126 # Detect separator or end 

127 elif char in (':', ';', '\0'): 

128 volume_nodes += [Paths.translate(volume_node)] 

129 volume_node = '' 

130 

131 # Append to volume node 

132 else: 

133 volume_node += char 

134 

135 # Result 

136 return volume_nodes