1
-# Test utilities for fetching & caching assets
1
+# SPDX-License-Identifier: GPL-2.0-or-later
2
#
3
# Copyright 2024 Red Hat, Inc.
4
#
5
# This work is licensed under the terms of the GNU GPL, version 2 or
6
# later. See the COPYING file in the top-level directory.
7
+'''
8
+Test utilities for fetching & caching assets
9
+'''
10
11
import hashlib
12
import logging
21
from shutil import copyfileobj
22
from urllib.error import HTTPError, URLError
23
24
+
25
class AssetError(Exception):
26
+ '''This exception will be raised if an asset is not usable'''
27
def __init__(self, asset, msg, transient=False):
28
self.url = asset.url
29
self.msg = msg
30
self.transient = transient
31
32
def __str__(self):
28
- return "%s: %s" % (self.url, self.msg)
33
+ return f"{self.url}: {self.msg}"
34
30
-# Instances of this class must be declared as class level variables
31
-# starting with a name "ASSET_". This enables the pre-caching logic
32
-# to easily find all referenced assets and download them prior to
33
-# execution of the tests.
34
-class Asset:
35
36
+class Asset:
37
+ '''
38
+ This class is used to represent an asset that gets downloaded from
39
+ the internet and will be stored in the local asset cache.
40
+ Instances of this class must be declared as class level variables
41
+ starting with a name "ASSET_". This enables the pre-caching logic
42
+ to easily find all referenced assets and download them prior to
43
+ execution of the tests.
44
+ '''
45
def __init__(self, url, hashsum):
46
self.url = url
47
self.hash = hashsum
55
self.log = logging.getLogger('qemu-test')
56
57
def __repr__(self):
49
- return "Asset: url=%s hash=%s cache=%s" % (
50
- self.url, self.hash, self.cache_file)
58
+ return f"Asset: url={self.url} hash={self.hash} cache={self.cache_file}"
59
60
def __str__(self):
61
return str(self.cache_file)
79
return self.hash == hl.hexdigest()
80
81
def valid(self):
82
+ '''Check whether the file exists in the cache and has the right hash'''
83
if os.getenv("QEMU_TEST_REFRESH_CACHE", None) is not None:
84
self.log.info("Force refresh of asset %s", self.url)
85
return False
87
return self.cache_file.exists() and self._check(self.cache_file)
88
89
def fetchable(self):
90
+ '''Check whether we are allowed to download assets from the internet'''
91
return not os.environ.get("QEMU_TEST_NO_DOWNLOAD", False)
92
93
def available(self):
94
+ '''Check whether the asset is either in the cache or fetchable'''
95
return self.valid() or self.fetchable()
96
97
def _wait_for_other_download(self, tmp_cache_file):
134
self.cache_file.with_suffix(".stamp").write_text(f"{int(time.time())}")
135
136
def fetch(self):
137
+ '''Download the asset from the internet'''
138
if not self.cache_dir.exists():
139
self.cache_dir.mkdir(parents=True, exist_ok=True)
140
191
# server or networking problem
192
if e.code == 404:
193
raise AssetError(self, "Unable to download: "
182
- "HTTP error %d" % e.code) from e
194
+ f"HTTP error {e.code}") from e
195
continue
196
except URLError as e:
197
# This is typically a network/service level error
199
tmp_cache_file.unlink()
200
self.log.error("Unable to download %s: URL error %s",
201
self.url, e.reason)
190
- raise AssetError(self, "Unable to download: URL error %s" %
191
- e.reason, transient=True) from e
202
+ raise AssetError(self,
203
+ f"Unable to download: URL error{e.reason}",
204
+ transient=True) from e
205
except ConnectionError as e:
206
# A socket connection failure, such as dropped conn
207
# or refused conn
211
continue
212
except Exception as e:
213
tmp_cache_file.unlink()
201
- raise AssetError(self, "Unable to download: %s" % e,
214
+ raise AssetError(self, f"Unable to download: {e}",
215
transient=True) from e
216
217
if not os.path.exists(tmp_cache_file):
223
self.url.encode('utf8'))
224
os.setxattr(str(tmp_cache_file), "user.qemu-asset-hash",
225
self.hash.encode('utf8'))
213
- except Exception as e:
226
+ except OSError as e:
227
self.log.debug("Unable to set xattr on %s: %s", tmp_cache_file, e)
228
229
if not self._check(tmp_cache_file):
230
tmp_cache_file.unlink()
218
- raise AssetError(self, "Hash does not match %s" % self.hash)
231
+ raise AssetError(self, f"Hash does not match {self.hash}")
232
tmp_cache_file.replace(self.cache_file)
233
self._save_time_stamp()
234
# Remove write perms to stop tests accidentally modifying them
239
240
@staticmethod
241
def precache_test(test):
242
+ '''
243
+ Look for variables starting with "ASSET_" and try to fetch the asset
244
+ that is specified there.
245
+ '''
246
log = logging.getLogger('qemu-test')
247
log.setLevel(logging.DEBUG)
248
handler = logging.StreamHandler(sys.stdout)
264
265
@staticmethod
266
def precache_suite(suite):
267
+ '''
268
+ Iterate through all tests/suites in a suite and precache their assets
269
+ '''
270
for test in suite:
271
if isinstance(test, unittest.TestSuite):
272
Asset.precache_suite(test)
275
276
@staticmethod
277
def precache_suites(path, cache_tstamp):
278
+ '''
279
+ Get the available test suite and precache their assets
280
+ '''
281
loader = unittest.loader.defaultTestLoader
282
tests = loader.loadTestsFromNames([path], None)
283