1from test.http_test import HTTPTest 2from misc.wget_file import WgetFile 3import hashlib 4 5class Metalinkv3_XML: 6 7 """ Metalink/XML v3 object """ 8 9 # Initialize the Metalink object 10 def __init__ (self): 11 self.reset () 12 13 # Reset the Metalink object 14 def reset (self): 15 self.LocalFiles = [] # list of WgetFile objects 16 self.ServerFiles = [[]] # list of WgetFile objects 17 self.ExpectedFiles = [] # list of WgetFile objects 18 self.LocalFiles_Set = [] # used as `list (set (var))` 19 self.ServerFiles_Set = [[]] # used as `list (set (var))` 20 self.ExpectedFiles_Set = [] # used as `list (set (var))` 21 self.Xml = '' # Metalink/XML content 22 self.XmlName = '' # Metalink/XML file name 23 self.XmlFile = None # Metalink/XML WgetFile object 24 self.Xml_Header = '<?xml version="1.0" encoding="utf-8"?>\n' + \ 25 '<metalink version="3.0" xmlns="http://www.metalinker.org/">\n' + \ 26 ' <publisher>\n' + \ 27 ' <name>GNU Wget</name>\n' + \ 28 ' </publisher>\n' + \ 29 ' <license>\n' + \ 30 ' <name>GNU GPL</name>\n' + \ 31 ' <url>http://www.gnu.org/licenses/gpl.html</url>\n' + \ 32 ' </license>\n' + \ 33 ' <identity>Wget Test Files</identity>\n' + \ 34 ' <version>1.2.3</version>\n' + \ 35 ' <description>Wget Test Files description</description>\n' + \ 36 ' <files>\n' 37 self.Xml_Footer = ' </files>\n' + \ 38 '</metalink>\n' 39 40 # Print the Metalink object. 41 def print_meta (self): 42 43 print (self.Xml) 44 print ("LocalFiles = " + str (self.LocalFiles_Set)) 45 print ("ServerFiles = " + str (self.ServerFiles_Set)) 46 print ("ExpectedFiles = " + str (self.ExpectedFiles_Set)) 47 48 # Add LocalFiles as WgetFile objects 49 # 50 # ["file_name", "content"], 51 # ["file_name", "content"] 52 def add_LocalFiles (self, *local_files): 53 54 for (file_name, content) in local_files: 55 if not file_name in self.LocalFiles_Set: 56 self.LocalFiles_Set.append (file_name) 57 self.LocalFiles.append (WgetFile (file_name, content)) 58 59 # Add ServerFiles as WgetFile objects 60 # 61 # ["file_name", "content"], 62 # ["file_name", "content"] 63 def add_ServerFiles (self, *server_files): 64 65 for (file_name, content) in server_files: 66 if not file_name in self.ServerFiles_Set[0]: 67 self.ServerFiles_Set[0].append (file_name) 68 self.ServerFiles[0].append (WgetFile (file_name, content)) 69 70 # Add ExpectedFiles as WgetFile objects 71 # 72 # ["file_name", "content"], 73 # ["file_name", "content"] 74 def add_ExpectedFiles (self, *expected_files): 75 76 for (file_name, content) in expected_files: 77 if not file_name in self.ExpectedFiles_Set: 78 self.ExpectedFiles_Set.append (file_name) 79 self.ExpectedFiles.append (WgetFile (file_name, content)) 80 81 # Run a Wget HTTP test for the Metalink object. 82 def http_test (self, command_line, expected_retcode): 83 84 pre_test = { 85 "ServerFiles" : self.ServerFiles, # list of WgetFile objects as [[]] 86 "LocalFiles" : self.LocalFiles, # list of WgetFile objects as [] 87 } 88 89 test_options = { 90 "WgetCommands" : command_line, # Wget cli 91 "Urls" : [[]], # Wget urls 92 } 93 94 post_test = { 95 "ExpectedFiles" : self.ExpectedFiles, # list of WgetFile objects as [] 96 "ExpectedRetcode" : expected_retcode, # Wget return status code 97 } 98 99 http_test = HTTPTest ( 100 pre_hook=pre_test, 101 test_params=test_options, 102 post_hook=post_test, 103 ) 104 105 http_test.server_setup() 106 # Get and use dynamic server sockname 107 srv_host, srv_port = http_test.servers[0].server_inst.socket.getsockname () 108 109 self.set_srv (srv_host, srv_port) 110 111 err = http_test.begin () 112 113 return err 114 115 # Set the Wget server host and port in the Metalink/XML content. 116 def set_srv (self, srv_host, srv_port): 117 118 self.Xml = self.Xml.replace('{{SRV_HOST}}', srv_host) 119 self.Xml = self.Xml.replace('{{SRV_PORT}}', str (srv_port)) 120 121 if self.XmlFile is not None: 122 self.XmlFile.content = self.Xml 123 124 # Create the Metalink/XML file. 125 # 126 # Add the Metalink/XML file to the list of ExpectedFiles. 127 # 128 # size: 129 # True auto-compute size 130 # None no <size></size> 131 # any use this size 132 # 133 # hash_sha256: 134 # False no <verification></verification> 135 # True auto-compute sha256 136 # None no <hash></hash> 137 # any use this hash 138 # 139 # ARGUMENTS: 140 # 141 # "xml_name", # Metalink/XML file name 142 # ["file_name", "save_name", "content", size, hash_sha256, # metalink:file 143 # ["srv_file", "srv_content", utype, location, preference], # resource 144 # ["srv_file", "srv_content", utype, location, preference]], # resource 145 # ["file_name", "save_name", "content", size, hash_sha256, 146 # ["srv_file", "srv_content", utype, location, preference], 147 # ["srv_file", "srv_content", utype, location, preference]] 148 def xml (self, xml_name, *xml_data): 149 150 self.Xml = self.Xml_Header 151 152 for (file_name, save_name, content, size, hash_sha256, *resources) in xml_data: 153 self.Xml += self.file_tag (file_name, save_name, content, size, hash_sha256, resources) + '\n' 154 155 self.Xml += self.Xml_Footer 156 157 self.XmlName = xml_name 158 self.XmlFile = WgetFile (xml_name, self.Xml) 159 160 if not xml_name in self.LocalFiles_Set: 161 self.LocalFiles_Set.append (xml_name) 162 self.LocalFiles.append (self.XmlFile) 163 164 if not xml_name in self.ExpectedFiles_Set: 165 self.ExpectedFiles_Set.append (xml_name) 166 self.ExpectedFiles.append (self.XmlFile) 167 168 # Create the file tag. 169 # 170 # Add the file to be saved to the list of ExpectedFiles. 171 # 172 # size: 173 # True auto-compute size 174 # None no <size></size> 175 # any use this size 176 # 177 # hash_sha256: 178 # False no <verification></verification> 179 # True auto-compute sha256 180 # None no <hash></hash> 181 # any use this hash 182 # 183 # ARGUMENTS: 184 # 185 # ["file_name", "save_name", "content", size, hash_sha256, # metalink:file 186 # ["srv_file", "srv_content", utype, location, preference], # resource 187 # ["srv_file", "srv_content", utype, location, preference]] # resource 188 def file_tag (self, file_name, save_name, content, size, hash_sha256, resources): 189 190 Tag = ' <file name="' + file_name + '">\n' 191 192 if save_name is not None: 193 self.add_ExpectedFiles ([save_name, content]) 194 195 size_Tag = self.size_tag (content, size) 196 197 if size_Tag is not None: 198 Tag += size_Tag + '\n' 199 200 verification_Tag = self.verification_tag (content, hash_sha256) 201 202 if verification_Tag is not None: 203 Tag += verification_Tag + '\n' 204 205 Tag += self.resources_tag (resources) + '\n' 206 207 Tag += ' </file>' 208 209 return Tag 210 211 # Create the size tag. 212 # 213 # size: 214 # True auto-compute size 215 # None no <size></size> 216 # any use this size 217 # 218 # ARGUMENTS: 219 # 220 # "content", size 221 def size_tag (self, content = None, size = None): 222 223 Tag = None 224 225 if content is not None and size is True: 226 size = len (content) 227 228 if size is not None: 229 Tag = ' <size>' + str (size) + '</size>' 230 231 return Tag 232 233 # Create the verification tag. 234 # 235 # hash_sha256: 236 # False no <verification></verification> 237 # True auto-compute sha256 238 # None no <hash></hash> 239 # any use this hash 240 # 241 # ARGUMENTS: 242 # 243 # "content", hash_sha256 244 def verification_tag (self, content = None, hash_sha256 = None): 245 246 Tag = None 247 248 if hash_sha256 is not False: 249 250 if content is not None and hash_sha256 is True: 251 hash_sha256 = hashlib.sha256 (content.encode ('UTF-8')).hexdigest () 252 253 if hash_sha256 is None: 254 Tag = ' <verification>\n' + \ 255 ' </verification>' 256 else: 257 Tag = ' <verification>\n' + \ 258 ' <hash type="sha256">' + str (hash_sha256) + '</hash>\n' + \ 259 ' </verification>' 260 261 return Tag 262 263 # Create the resources tag. 264 # 265 # ARGUMENTS: 266 # 267 # ["srv_file", "srv_content", utype, location, preference], # resource 268 # ["srv_file", "srv_content", utype, location, preference] # resource 269 def resources_tag (self, resources): 270 271 Tag = ' <resources>\n' 272 273 for (srv_file, srv_content, utype, location, preference) in resources: 274 Tag += self.url_tag (srv_file, srv_content, utype, location, preference) + '\n' 275 276 Tag += ' </resources>' 277 278 return Tag 279 280 # Create the url tag. 281 # 282 # Add the file to the list of Files when there is a content. 283 # 284 # ARGUMENTS: 285 # 286 # "srv_file", "srv_content", utype, location, preference # resource 287 def url_tag (self, srv_file, srv_content = None, utype = "http", location = None, preference = 999999): 288 289 Loc = '' 290 291 if location is not None: 292 Loc = 'location="' + location + '" ' 293 294 Tag = ' ' + \ 295 '<url ' + \ 296 'type="' + utype + '" ' + \ 297 Loc + \ 298 'preference="' + str (preference) + '">' + \ 299 'http://{{SRV_HOST}}:{{SRV_PORT}}/' + srv_file + \ 300 '</url>' 301 302 if srv_content is not None: 303 self.add_ServerFiles ([srv_file, srv_content]) 304 305 return Tag 306