Coverage for src/features/images.py: 100.00%

37 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-10-18 15:02 +0000

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from argparse import Namespace 

5from typing import List 

6 

7# Components 

8from ..engines.engine import Engine 

9from ..models.pipelines import Pipeline 

10from ..types.lists import Lists 

11 

12# ImagesFeature class 

13class ImagesFeature: 

14 

15 # Members 

16 __engine: Engine 

17 __images: List[str] 

18 __options: Namespace 

19 

20 # Constructor 

21 def __init__( 

22 self, 

23 pipeline: Pipeline, 

24 options: Namespace, 

25 ) -> None: 

26 

27 # Prepare pipeline 

28 self.__pipeline = pipeline 

29 

30 # Prepare options 

31 self.__options = options 

32 

33 # Prepare engine 

34 self.__engine = Engine(self.__options.engine) 

35 

36 # Prepare images 

37 self.__images = [] 

38 self.__prepare_images() 

39 

40 # Filter 

41 def __filter( 

42 self, 

43 job: str, 

44 ) -> bool: 

45 

46 # Filter pipeline jobs list, pylint: disable=duplicate-code 

47 if not self.__options.pipeline and self.__options.names and not Lists.match( 

48 self.__options.names, 

49 job, 

50 ignore_case=self.__options.ignore_case, 

51 no_regex=self.__options.no_regex, 

52 ): 

53 return False 

54 

55 # Filter pipeline stages list 

56 if self.__options.pipeline and self.__options.names and not Lists.match( 

57 self.__options.names, 

58 self.__pipeline.jobs[job].stage, 

59 ignore_case=self.__options.ignore_case, 

60 no_regex=self.__options.no_regex, 

61 ): 

62 return False 

63 

64 # Result 

65 return True 

66 

67 # Prepare images 

68 def __prepare_images(self) -> None: 

69 

70 # Iterate through images 

71 for job in self.__pipeline.jobs: 

72 

73 # Filter jobs 

74 if not self.__filter(job): 

75 continue 

76 

77 # Extract job image 

78 image = self.__pipeline.jobs[job].image 

79 if image and not self.__pipeline.jobs[ 

80 job].options.host and image not in self.__images: 

81 self.__images += [image] 

82 

83 # Extract job services 

84 for service in self.__pipeline.jobs[job].services: 

85 if service.image not in self.__images: 

86 self.__images += [service.image] 

87 

88 # Sort images 

89 self.__images.sort() 

90 

91 # Pull 

92 def pull(self) -> bool: 

93 

94 # Pull images 

95 for image in self.__images: 

96 self.__engine.pull( 

97 image, 

98 self.__options.force, 

99 ) 

100 

101 # Result 

102 return bool(self.__images) 

103 

104 # Remove images 

105 def rmi(self) -> bool: 

106 

107 # Remove images 

108 for image in self.__images: 

109 self.__engine.rmi(image) 

110 

111 # Result 

112 return bool(self.__images)