4
import http.server as SimpleHTTPServer
14
from mock import patch
15
from multiprocessing import Process
17
sys.path.insert(0, "data")
18
sys.path.insert(1, "../data")
19
import package_data_downloader # nopep8
22
class TestHttpd(Process):
23
def __init__(self, port, servdir):
24
super(TestHttpd, self).__init__()
26
self.servdir = servdir
29
os.chdir(self.servdir)
30
Handler = SimpleHTTPServer.SimpleHTTPRequestHandler
31
httpd = socketserver.TCPServer(("localhost", self.port), Handler)
35
class DownloadTestCase(unittest.TestCase):
38
self.tempdir = tempfile.mkdtemp()
39
self._setup_canary_file()
40
self.PORT = random.randint(1025, 60000)
41
self.p = TestHttpd(self.PORT, self.tempdir)
46
shutil.rmtree(self.tempdir)
48
def _setup_canary_file(self):
49
self.canary_text = "meep"
50
self.canary_file = os.path.join(self.tempdir, "canary-file.txt")
51
with open(self.canary_file, "w") as f:
52
f.write(self.canary_text)
53
self.canary_hashsum = hashlib.sha256(
54
self.canary_text.encode('utf-8')).hexdigest()
56
def test_download_file(self):
57
package_data_downloader.STAMPDIR = self.tempdir
58
os.makedirs(os.path.join(self.tempdir, "partial"))
59
uri = "http://localhost:%s/%s" % (
60
self.PORT, os.path.basename(self.canary_file))
61
destfile = package_data_downloader.download_file(
62
uri, self.canary_hashsum)
63
self.assertNotEqual(destfile, None)
64
self.assertEqual(hashlib.sha256(
65
open(destfile).read().encode('utf-8')).hexdigest(),
69
class ProcessDownloadRequestsTestCase(unittest.TestCase):
72
self.tempdir = tempfile.mkdtemp()
73
self._setup_canary_file()
74
self.PORT = random.randint(1025, 60000)
75
self.p = TestHttpd(self.PORT, self.tempdir)
77
self.tempdir = tempfile.mkdtemp()
79
stampdir = os.path.join(self.tempdir, "stampdir")
81
package_data_downloader.STAMPDIR = stampdir
82
os.makedirs(os.path.join(stampdir, "partial"))
84
datadir = os.path.join(self.tempdir, "datadir")
86
package_data_downloader.DATADIR = datadir
87
# override the location of the notifier files
88
notifierdir = os.path.join(self.tempdir, "notifierdir")
89
os.makedirs(notifierdir)
90
package_data_downloader.NOTIFIER_FILE = \
91
os.path.join(notifierdir, "data-downloads-failed")
92
package_data_downloader.NOTIFIER_PERMANENT_FILE = \
93
package_data_downloader.NOTIFIER_FILE + "-permanently"
97
shutil.rmtree(self.tempdir)
99
def _setup_canary_file(self):
100
self.canary_text = "meep"
101
self.canary_file = os.path.join(self.tempdir, "canary-file.txt")
102
with open(self.canary_file, "w") as f:
103
f.write(self.canary_text)
104
self.canary_hashsum = hashlib.sha256(
105
self.canary_text.encode('utf-8')).hexdigest()
107
def _setup_hook_file(self, filename, script="/bin/true"):
108
with open(os.path.join(package_data_downloader.DATADIR,
109
filename), "w") as foo:
110
foo.write("Url: http://localhost:%s/%s\n" %
111
(self.PORT, os.path.basename(self.canary_file)))
112
foo.write("Sha256: %s\n" % self.canary_hashsum)
114
foo.write("Script: %s" % script)
116
def test_hookfile_older_than_stampfile(self):
117
# hook file older than stamp file nothing happens
118
# create the hook file
119
hookfile = "older-hookfile"
120
self._setup_hook_file(hookfile)
122
# create the stamp file
123
stampfile = os.path.join(package_data_downloader.STAMPDIR,
125
with open(stampfile, "w"):
127
orig_stamp_time = os.stat(stampfile).st_mtime
128
package_data_downloader.process_download_requests()
129
new_stamp_time = os.stat(stampfile).st_mtime
130
self.assertEqual(orig_stamp_time, new_stamp_time)
131
# confirm failure files not created
132
self.assertFalse(os.path.exists(
133
os.path.join(package_data_downloader.STAMPDIR,
134
"%s.permanent-failure" % hookfile)))
135
self.assertFalse(os.path.exists(package_data_downloader.NOTIFIER_FILE))
137
def test_stampfile_older_than_hookfile(self):
138
# hook file newer than stampfile, download, run script, update
140
hookfile = "older-stampfile"
141
stampfile = os.path.join(package_data_downloader.STAMPDIR,
143
with open(stampfile, "w"):
145
orig_stamp_date = os.stat(stampfile).st_mtime
147
# create the hook file
148
self._setup_hook_file(hookfile)
149
package_data_downloader.process_download_requests()
150
# the download succeeded so the stamp file is touched
151
new_stamp_date = os.stat(stampfile).st_mtime
152
self.assertGreater(new_stamp_date, orig_stamp_date)
153
# confirm failure files not created
154
self.assertFalse(os.path.exists(
155
os.path.join(package_data_downloader.STAMPDIR,
156
"%s.permanent-failure" % hookfile)))
157
self.assertFalse(os.path.exists(package_data_downloader.NOTIFIER_FILE))
159
def test_only_retry_failure(self):
160
# two hook files but only has failed
162
stampfile = os.path.join(package_data_downloader.STAMPDIR,
164
with open(stampfile, "w"):
166
orig_stamp_date = os.stat(stampfile).st_mtime
168
# create the hook file
169
self._setup_hook_file(hookfile)
170
fail_hookfile = "failure"
171
# the stampfile indicates a failure
172
fail_stampfile = os.path.join(package_data_downloader.STAMPDIR,
173
"%s.failed" % fail_hookfile)
174
with open(fail_stampfile, "w"):
177
# create the hook file
178
self._setup_hook_file(fail_hookfile)
179
# create an empty notifier file
180
with open(package_data_downloader.NOTIFIER_FILE, "w"):
182
package_data_downloader.process_download_requests()
183
new_stamp_date = os.stat(stampfile).st_mtime
184
# if the downloaded succeeded it shouldn't be done again
185
self.assertEqual(new_stamp_date, orig_stamp_date)
187
def test_hookfile_script_fails(self):
188
# script in the hook file fails, failure recorded as permanent
189
hookfile = "script-failure"
190
stampfile = os.path.join(package_data_downloader.STAMPDIR,
192
with open(stampfile, "w"):
195
# create the hook file
196
self._setup_hook_file(hookfile, "/bin/false")
197
package_data_downloader.process_download_requests()
198
# script failures are considered permanent
199
self.assertTrue(os.path.exists(
200
os.path.join(package_data_downloader.STAMPDIR,
201
"%s.permanent-failure" % hookfile)))
203
def test_stampfile_and_notifierfile(self):
204
# notifier file removed on successful download
205
hookfile = "stampfile_and_notifierfile"
206
stampfile = os.path.join(package_data_downloader.STAMPDIR,
208
with open(stampfile, "w"):
211
self._setup_hook_file(hookfile)
212
# create an empty notifier file
213
with open(package_data_downloader.NOTIFIER_FILE, "w"):
215
package_data_downloader.process_download_requests()
216
self.assertFalse(os.path.exists(package_data_downloader.NOTIFIER_FILE))
218
def test_stampfile_notifierfile_and_notifierpermanent(self):
219
# both notifier files removed on successful download
220
hookfile = "stampfile_notifierfile_and_notifierpermanentfile"
221
stampfile = os.path.join(package_data_downloader.STAMPDIR,
223
with open(stampfile, "w"):
226
self._setup_hook_file(hookfile)
227
# create empty notifier files
228
with open(package_data_downloader.NOTIFIER_FILE, "w"):
230
with open(package_data_downloader.NOTIFIER_PERMANENT_FILE, "w"):
232
package_data_downloader.process_download_requests()
233
self.assertFalse(os.path.exists(package_data_downloader.NOTIFIER_FILE))
234
self.assertFalse(os.path.exists(
235
package_data_downloader.NOTIFIER_PERMANENT_FILE))
237
def test_stampfile_and_notifierpermanent(self):
238
# permanent notifier file removed on successful download
239
hookfile = "stampfile_and_permanentnotifierfile"
240
stampfile = os.path.join(package_data_downloader.STAMPDIR,
242
with open(stampfile, "w"):
245
self._setup_hook_file(hookfile)
246
# create an empty permanent notifier file
247
with open(package_data_downloader.NOTIFIER_PERMANENT_FILE, "w"):
249
package_data_downloader.process_download_requests()
250
self.assertFalse(os.path.exists(
251
package_data_downloader.NOTIFIER_PERMANENT_FILE))
253
def test_hookfile_download_failure(self):
254
# can't download the file, non-permanent failure created
255
hookfile = "download-failure"
256
stampfile = os.path.join(package_data_downloader.STAMPDIR,
258
with open(stampfile, "w"):
261
# overwrite canary file to create a failure
262
self.canary_file = "not-there.txt"
263
# create the hook file
264
self._setup_hook_file(hookfile)
265
package_data_downloader.process_download_requests()
266
self.assertTrue(os.path.exists(
267
os.path.join(package_data_downloader.STAMPDIR,
268
"%s.failed" % hookfile)))
269
self.assertFalse(os.path.exists(
270
os.path.join(package_data_downloader.STAMPDIR,
271
"%s.permanent-failure" % hookfile)))
273
def test_hookfile_notifierfile_download_failure(self):
274
# can't download the file, notifier file stays around
275
hookfile = "download-failure-with-notifierfile"
276
# overwrite canary file to create a failure
277
self.canary_file = "not-there.txt"
278
# create the hook file
279
self._setup_hook_file(hookfile)
280
# create an empty notifier file
281
with open(package_data_downloader.NOTIFIER_FILE, "w"):
283
package_data_downloader.process_download_requests()
284
# because there was a failure it shouldn't be removed
285
self.assertTrue(os.path.exists(
286
package_data_downloader.NOTIFIER_FILE))
289
class PackageDataDownloaderTestCase(unittest.TestCase):
292
self.tmpdir = tempfile.mkdtemp()
294
stampdir = os.path.join(self.tmpdir, "stampdir")
295
os.makedirs(stampdir)
296
package_data_downloader.STAMPDIR = stampdir
298
datadir = os.path.join(self.tmpdir, "datadir")
300
package_data_downloader.DATADIR = datadir
303
shutil.rmtree(self.tmpdir)
305
def test_wrong_template_translations(self):
306
package_data_downloader.NOTIFIER_SOURCE_FILE = \
307
'data/package-data-downloads-failed.in'
308
package_data_downloader.NOTIFIER_FILE = \
309
self.tmpdir + "/data-downloads-failed"
310
package_data_downloader.NOTIFIER_PERMANENT_SOURCE_FILE = \
311
'data/package-data-downloads-failed-permanently.in'
312
package_data_downloader.NOTIFIER_PERMANENT_FILE = \
313
package_data_downloader.NOTIFIER_FILE + '-permanently'
314
package_data_downloader.trigger_update_notifier([], False)
315
package_data_downloader.trigger_update_notifier([], True)
317
def test_permanently_failed(self):
318
# create a bunch of files using the provided mechanism
319
test_files = ["foo.permanent-failure", "bar.failure", "baz"]
321
package_data_downloader.create_or_update_stampfile(
322
os.path.join(package_data_downloader.STAMPDIR, f))
324
sorted(os.listdir(package_data_downloader.STAMPDIR)),
326
# test hook_is_permanently_failed()
328
package_data_downloader.hook_is_permanently_failed("foo"))
330
package_data_downloader.hook_is_permanently_failed("bar"))
332
package_data_downloader.hook_is_permanently_failed("baz"))
333
# existing_permanent_failures()
335
package_data_downloader.existing_permanent_failures(),
338
def test_hook_aged_out(self):
339
# test to see if a hook has failed for more than 3 days
340
# create a failure file
341
with open(os.path.join(package_data_downloader.STAMPDIR,
342
"aged.failed"), "w"):
344
from datetime import datetime, timedelta
345
# patch datetime so we think its the future
346
with patch('package_data_downloader.datetime') as mock_datetime:
347
thefuture = (datetime.now() + timedelta(days=3))
348
mock_datetime.now.return_value = thefuture
349
mock_datetime.fromtimestamp.side_effect = \
350
lambda *args, **kw: datetime.fromtimestamp(*args, **kw)
351
self.assertEqual(package_data_downloader.datetime.now(),
353
self.assertTrue(package_data_downloader.hook_aged_out("aged"))
355
def test_mark_hook_failed(self):
357
package_data_downloader.create_or_update_stampfile(
358
os.path.join(package_data_downloader.STAMPDIR, "foo"))
360
package_data_downloader.mark_hook_failed("foo")
361
self.assertEqual(os.listdir(package_data_downloader.STAMPDIR),
364
package_data_downloader.hook_is_permanently_failed("foo"))
366
package_data_downloader.mark_hook_failed("foo", permanent=True)
367
self.assertEqual(os.listdir(package_data_downloader.STAMPDIR),
368
["foo.permanent-failure"])
370
package_data_downloader.hook_is_permanently_failed("foo"))
372
def test_get_hook_file_names(self):
373
# test that .dpkg-* is ignored
374
test_datadir_files = ["foo.dpkg-new", "bar"]
375
for name in test_datadir_files:
376
with open(os.path.join(package_data_downloader.DATADIR, name),
379
self.assertEqual(package_data_downloader.get_hook_file_names(),
382
def test_trigger_update_notifier(self):
384
package_data_downloader.NOTIFIER_FILE = os.path.join(
385
self.tmpdir, "data-downloads-failed")
386
# point to local repo file
387
package_data_downloader.NOTIFIER_SOURCE_FILE = \
388
"data/package-data-downloads-failed.in"
389
package_data_downloader.trigger_update_notifier(
390
failures=["foo", "bar"], permanent=False)
391
data = open(package_data_downloader.NOTIFIER_FILE).read()
392
self.assertEqual(data, """Priority: High
393
_Name: Failure to download extra data files
395
Command: pkexec /usr/lib/update-notifier/package-data-downloader
397
The following packages requested additional data downloads after package
398
installation, but the data could not be downloaded or could not be
405
The download will be attempted again later, or you can try the download
406
again now. Running this command requires an active Internet connection.
410
if __name__ == "__main__":
added
removed
tests/test_package-data-downloader.py
