Coverage for r11k/puppetmodule/forge.py: 84%

1""" 

2Modules form the [Puppet Forge][forge]. 

3 

4The Forge notes a few installation methods, where we (currently) 

5aren't listed. The manual install method 

6 

7```sh 

8puppet module install puppetlabs-stdlib --version 8.4.0 

9``` 

10maps unto us as 

11>>> ForgePuppetModule(name='puppetlabs-stdlib', version='8.4.0') 

12 

13[forge]: https://forge.puppet.com/ 

14""" 

15 

16import logging 

17import os.path 

18import hashlib 

19import threading 

20import shutil 

21 

22from semver import VersionInfo 

23import tarfile 

24 

25from r11k.puppetmodule.base import PuppetModule 

26from r11k.puppet import PuppetMetadata 

27from r11k.forge import FullForgeModule, CurrentRelease 

28from r11k import util 

29from r11k.util import ( 

30 unfix_name, 

31) 

32 

33 

34logger = logging.getLogger(__name__) 

35 

36 

37class ForgePuppetModule(PuppetModule): 

38 """ 

39 Puppet module backed by the Puppet Forge. 

40 

41 :param name: Module name, as noted on the Forge for installation. 

42 :param version: Semver formatted string, as per the Puppet Forge 

43 """ 

44 

45 def _fetch_metadata(self) -> PuppetMetadata: 

46 """Return metadata for this module.""" 

47 try: 

48 version = self.version or self.latest() 

49 return self.get_versioned_forge_module_metadata(self.name, version).metadata 

50 except Exception as e: 

51 logger.error(self) 

52 raise e 

53 

54 def _fetch_latest_metadata(self) -> FullForgeModule: 

55 """ 

56 Get complete metadata for module. 

57 

58 Possibly download the module from puppet forge, or simple look it 

59 up in the module cache. 

60 

61 :param module_name: Name of the module to look up. 

62 """ 

63 unfixed_module_name = unfix_name(self.name) 

64 

65 url = f'{self.config.api_base}/v3/modules/{unfixed_module_name}' 

66 data = self.config.httpcache.get(url) 

67 if not data: 67 ↛ 68line 67 didn't jump to line 68, because the condition on line 67 was never true

68 raise ValueError('No data') 

69 

70 return FullForgeModule(**data) 

71 

72 def fetch(self) -> str: 

73 """ 

74 Download module from puppet forge. 

75 

76 Return path to a tarball. 

77 """ 

78 name = self.name 

79 version = self.version 

80 

81 # TODO this should use the forges file_uri instead (when available) 

82 url = f"{self.config.api_base}/v3/files/{name}-{version}.tar.gz" 

83 

84 # TODO this should use the HTTP cache, but currently HTTPcache 

85 # is hardcoded to expect JSON responses 

86 filename = os.path.join(self.config.tar_cache, f'{name}-{version}.tar.gz') 

87 

88 if not os.path.exists(filename): 

89 util.download_file(url, filename) 

90 

91 return filename 

92 

93 def publish(self, path: str) -> None: 

94 """ 

95 Extract this module to path. 

96 

97 Does some checks if the source and target differs, and does 

98 nothing if they are the same. 

99 """ 

100 tar_filepath = self.fetch() 

101 with tarfile.open(tar_filepath) as archive: 

102 archive_root = archive.next() 

103 if not archive_root: 103 ↛ 104line 103 didn't jump to line 104, because the condition on line 103 was never true

104 raise ValueError(f"Empty tarfile: {tar_filepath}") 

105 if not archive_root.isdir(): 105 ↛ 106line 105 didn't jump to line 106, because the condition on line 105 was never true

106 raise ValueError(f"Tar root not a directory: {tar_filepath}, {archive_root.name}") 

107 

108 unlink_old: bool = False 

109 if os.path.exists(path): 

110 if os.path.isdir(path): 110 ↛ 124line 110 didn't jump to line 124, because the condition on line 110 was never false

111 with open(os.path.join(path, 'metadata.json'), 'rb') as f: 

112 file_meta_chk = hashlib.sha256(f.read()) 

113 

114 member = archive.getmember(archive_root.name + '/metadata.json') 

115 if not archive.fileobj: 115 ↛ 116line 115 didn't jump to line 116, because the condition on line 115 was never true

116 raise ValueError('Underlying tar-file non-existant') 

117 archive.fileobj.seek(member.offset_data) 

118 tar_meta_chk = hashlib.sha256(archive.fileobj.read(member.size)) 

119 

120 if tar_meta_chk == file_meta_chk: 120 ↛ 121line 120 didn't jump to line 121, because the condition on line 120 was never true

121 return 

122 unlink_old = True 

123 else: 

124 os.unlink(path) 

125 

126 archive.extractall(path=os.path.dirname(path)) 

127 extract_path = os.path.join(os.path.dirname(path), 

128 archive_root.name) 

129 tmp_path = os.path.join(os.path.dirname(path), '.old-' + os.path.basename(path)) 

130 # This will hopefully cause the update to be 

131 # unnoticable to any observers 

132 with threading.Lock(): 

133 if unlink_old: 

134 os.rename(path, tmp_path) 

135 os.rename(extract_path, path) 

136 if os.path.exists(tmp_path): 

137 shutil.rmtree(tmp_path) 

138 

139 def versions(self) -> list[VersionInfo]: 

140 """Return available versions.""" 

141 j = self._fetch_latest_metadata() 

142 # return [o['version'] for o in j['releases']] 

143 return [o.version for o in j.releases] 

144 

145 def __repr__(self) -> str: 

146 return f"ForgePuppetModule('{self.name}', '{self.version}')" 

147 

148 def get_versioned_forge_module_metadata(self, module_name: str, version: str) -> CurrentRelease: 

149 """Retrieve release data from puppet forge.""" 

150 unfixed_module_name = unfix_name(module_name) 

151 url = f'{self.config.api_base}/v3/releases/{unfixed_module_name}-{version}' 

152 data = self.config.httpcache.get(url) 

153 if not data: 153 ↛ 154line 153 didn't jump to line 154, because the condition on line 153 was never true

154 raise ValueError('No Data') 

155 return CurrentRelease(**data)