Open files in binary mode for Windows portability
[zeroinstall/solver.git] / tests / testdownload.py
blobfc9fb135eac5fec6f2c44510740512c3a4269d32
1 #!/usr/bin/env python
2 from __future__ import with_statement
3 from basetest import BaseTest
4 import sys, tempfile, os
5 from StringIO import StringIO
6 import unittest, signal
7 from logging import getLogger, WARN, ERROR
8 from contextlib import contextmanager
10 sys.path.insert(0, '..')
12 os.environ["http_proxy"] = "localhost:8000"
14 from zeroinstall import helpers
15 from zeroinstall.injector import model, gpg, download, trust, background, arch, selections, qdom, run
16 from zeroinstall.injector.requirements import Requirements
17 from zeroinstall.injector.driver import Driver
18 from zeroinstall.zerostore import Store, NotStored; Store._add_with_helper = lambda *unused: False
19 from zeroinstall.support import basedir, tasks, ro_rmtree
20 from zeroinstall.injector import fetch
21 import data
22 import my_dbus
24 import server
26 ran_gui = False
27 def raise_gui(*args):
28 global ran_gui
29 ran_gui = True
30 background._detach = lambda: False
32 local_hello = """<?xml version="1.0" ?>
33 <selections command="run" interface="http://example.com:8000/Hello.xml" xmlns="http://zero-install.sourceforge.net/2004/injector/interface">
34 <selection id="." local-path='.' interface="http://example.com:8000/Hello.xml" version="0.1"><command name="run" path="foo"/></selection>
35 </selections>"""
37 @contextmanager
38 def output_suppressed():
39 old_stdout = sys.stdout
40 old_stderr = sys.stderr
41 try:
42 sys.stdout = StringIO()
43 sys.stderr = StringIO()
44 try:
45 yield
46 except Exception:
47 raise
48 except BaseException as ex:
49 # Don't abort unit-tests if someone raises SystemExit
50 raise Exception(str(type(ex)) + " " + str(ex))
51 finally:
52 sys.stdout = old_stdout
53 sys.stderr = old_stderr
55 @contextmanager
56 def trapped_exit(expected_exit_status):
57 pid = os.getpid()
58 old_exit = os._exit
59 def my_exit(code):
60 # The background handler runs in the same process
61 # as the tests, so don't let it abort.
62 if os.getpid() == pid:
63 raise SystemExit(code)
64 # But, child download processes are OK
65 old_exit(code)
66 os._exit = my_exit
67 try:
68 try:
69 yield
70 assert False
71 except SystemExit as ex:
72 assert ex.code == expected_exit_status
73 finally:
74 os._exit = old_exit
76 class Reply:
77 def __init__(self, reply):
78 self.reply = reply
80 def readline(self):
81 return self.reply
83 def download_and_execute(driver, prog_args, main = None):
84 downloaded = driver.solve_and_download_impls()
85 if downloaded:
86 tasks.wait_for_blocker(downloaded)
87 run.execute_selections(driver.solver.selections, prog_args, stores = driver.config.stores, main = main)
89 class NetworkManager:
90 def state(self):
91 return 3 # NM_STATUS_CONNECTED
93 server_process = None
94 def kill_server_process():
95 global server_process
96 if server_process is not None:
97 os.kill(server_process, signal.SIGTERM)
98 os.waitpid(server_process, 0)
99 server_process = None
101 def run_server(*args):
102 global server_process
103 assert server_process is None
104 server_process = server.handle_requests(*args)
106 real_get_selections_gui = helpers.get_selections_gui
108 class TestDownload(BaseTest):
109 def setUp(self):
110 BaseTest.setUp(self)
112 self.config.handler.allow_downloads = True
113 self.config.key_info_server = 'http://localhost:3333/key-info'
115 self.config.fetcher = fetch.Fetcher(self.config)
117 stream = tempfile.TemporaryFile()
118 stream.write(data.thomas_key)
119 stream.seek(0)
120 gpg.import_key(stream)
121 stream.close()
123 trust.trust_db.watchers = []
125 helpers.get_selections_gui = raise_gui
127 global ran_gui
128 ran_gui = False
130 def tearDown(self):
131 helpers.get_selections_gui = real_get_selections_gui
132 BaseTest.tearDown(self)
133 kill_server_process()
135 def testRejectKey(self):
136 with output_suppressed():
137 run_server('Hello', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B')
138 driver = Driver(requirements = Requirements('http://localhost:8000/Hello'), config = self.config)
139 assert driver.need_download()
140 sys.stdin = Reply("N\n")
141 try:
142 download_and_execute(driver, ['Hello'])
143 assert 0
144 except model.SafeException as ex:
145 if "has no usable implementations" not in str(ex):
146 raise ex
147 if "Not signed with a trusted key" not in str(self.config.handler.ex):
148 raise ex
149 self.config.handler.ex = None
151 def testRejectKeyXML(self):
152 with output_suppressed():
153 run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B')
154 driver = Driver(requirements = Requirements('http://example.com:8000/Hello.xml'), config = self.config)
155 assert driver.need_download()
156 sys.stdin = Reply("N\n")
157 try:
158 download_and_execute(driver, ['Hello'])
159 assert 0
160 except model.SafeException as ex:
161 if "has no usable implementations" not in str(ex):
162 raise ex
163 if "Not signed with a trusted key" not in str(self.config.handler.ex):
164 raise
165 self.config.handler.ex = None
167 def testImport(self):
168 from zeroinstall.injector import cli
170 rootLogger = getLogger()
171 rootLogger.disabled = True
172 try:
173 try:
174 cli.main(['--import', '-v', 'NO-SUCH-FILE'], config = self.config)
175 assert 0
176 except model.SafeException as ex:
177 assert 'NO-SUCH-FILE' in str(ex)
178 finally:
179 rootLogger.disabled = False
180 rootLogger.setLevel(WARN)
182 hello = self.config.iface_cache.get_feed('http://localhost:8000/Hello')
183 self.assertEqual(None, hello)
185 with output_suppressed():
186 run_server('6FCF121BE2390E0B.gpg')
187 sys.stdin = Reply("Y\n")
189 assert not trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
190 cli.main(['--import', 'Hello'], config = self.config)
191 assert trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
193 # Check we imported the interface after trusting the key
194 hello = self.config.iface_cache.get_feed('http://localhost:8000/Hello', force = True)
195 self.assertEqual(1, len(hello.implementations))
197 self.assertEqual(None, hello.local_path)
199 # Shouldn't need to prompt the second time
200 sys.stdin = None
201 cli.main(['--import', 'Hello'], config = self.config)
203 def testSelections(self):
204 from zeroinstall.injector import cli
205 root = qdom.parse(open("selections.xml"))
206 sels = selections.Selections(root)
207 class Options: dry_run = False
209 with output_suppressed():
210 run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
211 sys.stdin = Reply("Y\n")
212 try:
213 self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
214 assert False
215 except NotStored:
216 pass
217 cli.main(['--download-only', 'selections.xml'], config = self.config)
218 path = self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
219 assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
221 assert sels.download_missing(self.config) is None
223 def testHelpers(self):
224 from zeroinstall import helpers
226 with output_suppressed():
227 run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
228 sys.stdin = Reply("Y\n")
229 sels = helpers.ensure_cached('http://example.com:8000/Hello.xml', config = self.config)
230 path = self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
231 assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
232 assert sels.download_missing(self.config) is None
234 def testSelectionsWithFeed(self):
235 from zeroinstall.injector import cli
236 root = qdom.parse(open("selections.xml"))
237 sels = selections.Selections(root)
239 with output_suppressed():
240 run_server('Hello.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
241 sys.stdin = Reply("Y\n")
243 tasks.wait_for_blocker(self.config.fetcher.download_and_import_feed('http://example.com:8000/Hello.xml', self.config.iface_cache))
245 cli.main(['--download-only', 'selections.xml'], config = self.config)
246 path = self.config.stores.lookup_any(sels.selections['http://example.com:8000/Hello.xml'].digests)
247 assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
249 assert sels.download_missing(self.config) is None
251 def testAcceptKey(self):
252 with output_suppressed():
253 run_server('Hello', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
254 driver = Driver(requirements = Requirements('http://localhost:8000/Hello'), config = self.config)
255 assert driver.need_download()
256 sys.stdin = Reply("Y\n")
257 try:
258 download_and_execute(driver, ['Hello'], main = 'Missing')
259 assert 0
260 except model.SafeException as ex:
261 if "HelloWorld/Missing" not in str(ex):
262 raise
264 def testAutoAcceptKey(self):
265 self.config.auto_approve_keys = True
266 with output_suppressed():
267 run_server('Hello', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
268 driver = Driver(requirements = Requirements('http://localhost:8000/Hello'), config = self.config)
269 assert driver.need_download()
270 sys.stdin = Reply("")
271 try:
272 download_and_execute(driver, ['Hello'], main = 'Missing')
273 assert 0
274 except model.SafeException as ex:
275 if "HelloWorld/Missing" not in str(ex):
276 raise
278 def testDistro(self):
279 with output_suppressed():
280 native_url = 'http://example.com:8000/Native.xml'
282 # Initially, we don't have the feed at all...
283 master_feed = self.config.iface_cache.get_feed(native_url)
284 assert master_feed is None, master_feed
286 trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
287 run_server('Native.xml', '6FCF121BE2390E0B.gpg', '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B')
288 driver = Driver(requirements = Requirements(native_url), config = self.config)
289 assert driver.need_download()
291 solve = driver.solve_with_downloads()
292 tasks.wait_for_blocker(solve)
293 tasks.check(solve)
295 master_feed = self.config.iface_cache.get_feed(native_url)
296 assert master_feed is not None
297 assert master_feed.implementations == {}
299 distro_feed_url = master_feed.get_distro_feed()
300 assert distro_feed_url is not None
301 distro_feed = self.config.iface_cache.get_feed(distro_feed_url)
302 assert distro_feed is not None
303 assert len(distro_feed.implementations) == 2, distro_feed.implementations
305 def testWrongSize(self):
306 with output_suppressed():
307 run_server('Hello-wrong-size', '6FCF121BE2390E0B.gpg',
308 '/key-info/key/DE937DD411906ACF7C263B396FCF121BE2390E0B', 'HelloWorld.tgz')
309 driver = Driver(requirements = Requirements('http://localhost:8000/Hello-wrong-size'), config = self.config)
310 assert driver.need_download()
311 sys.stdin = Reply("Y\n")
312 try:
313 download_and_execute(driver, ['Hello'], main = 'Missing')
314 assert 0
315 except model.SafeException as ex:
316 if "Downloaded archive has incorrect size" not in str(ex):
317 raise ex
319 def testRecipe(self):
320 old_out = sys.stdout
321 try:
322 sys.stdout = StringIO()
323 run_server(('HelloWorld.tar.bz2', 'redirect/dummy_1-1_all.deb', 'dummy_1-1_all.deb'))
324 driver = Driver(requirements = Requirements(os.path.abspath('Recipe.xml')), config = self.config)
325 try:
326 download_and_execute(driver, [])
327 assert False
328 except model.SafeException as ex:
329 if "HelloWorld/Missing" not in str(ex):
330 raise ex
331 finally:
332 sys.stdout = old_out
334 def testSymlink(self):
335 old_out = sys.stdout
336 try:
337 sys.stdout = StringIO()
338 run_server(('HelloWorld.tar.bz2', 'HelloSym.tgz'))
339 driver = Driver(requirements = Requirements(os.path.abspath('RecipeSymlink.xml')), config = self.config)
340 try:
341 download_and_execute(driver, [])
342 assert False
343 except model.SafeException as ex:
344 if 'Attempt to unpack dir over symlink "HelloWorld"' not in str(ex):
345 raise
346 self.assertEqual(None, basedir.load_first_cache('0install.net', 'implementations', 'main'))
347 finally:
348 sys.stdout = old_out
350 def testAutopackage(self):
351 old_out = sys.stdout
352 try:
353 sys.stdout = StringIO()
354 run_server('HelloWorld.autopackage')
355 driver = Driver(requirements = Requirements(os.path.abspath('Autopackage.xml')), config = self.config)
356 try:
357 download_and_execute(driver, [])
358 assert False
359 except model.SafeException as ex:
360 if "HelloWorld/Missing" not in str(ex):
361 raise
362 finally:
363 sys.stdout = old_out
365 def testRecipeFailure(self):
366 old_out = sys.stdout
367 try:
368 sys.stdout = StringIO()
369 run_server('*')
370 driver = Driver(requirements = Requirements(os.path.abspath('Recipe.xml')), config = self.config)
371 try:
372 download_and_execute(driver, [])
373 assert False
374 except download.DownloadError as ex:
375 if "Connection" not in str(ex):
376 raise
377 finally:
378 sys.stdout = old_out
380 def testMirrors(self):
381 old_out = sys.stdout
382 try:
383 sys.stdout = StringIO()
384 getLogger().setLevel(ERROR)
385 trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
386 run_server(server.Give404('/Hello.xml'),
387 '/0mirror/feeds/http/example.com:8000/Hello.xml/latest.xml',
388 '/0mirror/keys/6FCF121BE2390E0B.gpg')
389 driver = Driver(requirements = Requirements('http://example.com:8000/Hello.xml'), config = self.config)
390 self.config.feed_mirror = 'http://example.com:8000/0mirror'
392 refreshed = driver.solve_with_downloads()
393 tasks.wait_for_blocker(refreshed)
394 assert driver.solver.ready
395 finally:
396 sys.stdout = old_out
398 def testReplay(self):
399 old_out = sys.stdout
400 try:
401 sys.stdout = StringIO()
402 getLogger().setLevel(ERROR)
403 iface = self.config.iface_cache.get_interface('http://example.com:8000/Hello.xml')
404 mtime = int(os.stat('Hello-new.xml').st_mtime)
405 self.config.iface_cache.update_feed_from_network(iface.uri, open('Hello-new.xml').read(), mtime + 10000)
407 trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
408 run_server(server.Give404('/Hello.xml'), 'latest.xml', '/0mirror/keys/6FCF121BE2390E0B.gpg', 'Hello.xml')
409 driver = Driver(requirements = Requirements('http://example.com:8000/Hello.xml'), config = self.config)
410 self.config.feed_mirror = 'http://example.com:8000/0mirror'
412 # Update from mirror (should ignore out-of-date timestamp)
413 refreshed = self.config.fetcher.download_and_import_feed(iface.uri, self.config.iface_cache)
414 tasks.wait_for_blocker(refreshed)
416 # Update from upstream (should report an error)
417 refreshed = self.config.fetcher.download_and_import_feed(iface.uri, self.config.iface_cache)
418 try:
419 tasks.wait_for_blocker(refreshed)
420 raise Exception("Should have been rejected!")
421 except model.SafeException as ex:
422 assert "New feed's modification time is before old version" in str(ex)
424 # Must finish with the newest version
425 self.assertEqual(1235911552, self.config.iface_cache._get_signature_date(iface.uri))
426 finally:
427 sys.stdout = old_out
429 def testBackground(self, verbose = False):
430 r = Requirements('http://example.com:8000/Hello.xml')
431 d = Driver(requirements = r, config = self.config)
432 self.import_feed(r.interface_uri, 'Hello.xml')
433 self.config.freshness = 0
434 self.config.network_use = model.network_minimal
435 d.solver.solve(r.interface_uri, arch.get_host_architecture())
436 assert d.solver.ready, d.solver.get_failure_reason()
438 @tasks.async
439 def choose_download(registed_cb, nid, actions):
440 try:
441 assert actions == ['download', 'Download'], actions
442 registed_cb(nid, 'download')
443 except:
444 import traceback
445 traceback.print_exc()
446 yield None
448 global ran_gui
449 ran_gui = False
450 os.environ['DISPLAY'] = 'dummy'
451 old_out = sys.stdout
452 try:
453 sys.stdout = StringIO()
454 run_server('Hello.xml', '6FCF121BE2390E0B.gpg')
455 my_dbus.system_services = {"org.freedesktop.NetworkManager": {"/org/freedesktop/NetworkManager": NetworkManager()}}
456 my_dbus.user_callback = choose_download
458 with trapped_exit(1):
459 from zeroinstall.injector import config
460 key_info = config.DEFAULT_KEY_LOOKUP_SERVER
461 config.DEFAULT_KEY_LOOKUP_SERVER = None
462 try:
463 background.spawn_background_update(d, verbose)
464 finally:
465 config.DEFAULT_KEY_LOOKUP_SERVER = key_info
466 finally:
467 sys.stdout = old_out
468 assert ran_gui
470 def testBackgroundVerbose(self):
471 self.testBackground(verbose = True)
473 def testBackgroundApp(self):
474 my_dbus.system_services = {"org.freedesktop.NetworkManager": {"/org/freedesktop/NetworkManager": NetworkManager()}}
476 trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
478 global ran_gui
480 with output_suppressed():
481 # Select a version of Hello
482 run_server('Hello.xml', '6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
483 r = Requirements('http://example.com:8000/Hello.xml')
484 driver = Driver(requirements = r, config = self.config)
485 tasks.wait_for_blocker(driver.solve_with_downloads())
486 assert driver.solver.ready
487 kill_server_process()
489 # Save it as an app
490 app = self.config.app_mgr.create_app('test-app', r)
491 app.set_selections(driver.solver.selections)
492 timestamp = os.path.join(app.path, 'last-checked')
493 last_check_attempt = os.path.join(app.path, 'last-check-attempt')
494 selections_path = os.path.join(app.path, 'selections.xml')
496 def reset_timestamps():
497 ran_gui = False
498 os.utime(timestamp, (1, 1)) # 1970
499 os.utime(selections_path, (1, 1))
500 if os.path.exists(last_check_attempt):
501 os.unlink(last_check_attempt)
503 # Download the implementation
504 sels = app.get_selections()
505 run_server('HelloWorld.tgz')
506 tasks.wait_for_blocker(app.download_selections(sels))
507 kill_server_process()
509 # Not time for a background update yet
510 self.config.freshness = 100
511 dl = app.download_selections(sels)
512 assert dl == None
513 assert not ran_gui
515 # Trigger a background update - no updates found
516 reset_timestamps()
517 run_server('Hello.xml')
518 with trapped_exit(1):
519 dl = app.download_selections(sels)
520 assert dl == None
521 assert not ran_gui
522 self.assertNotEqual(1, os.stat(timestamp).st_mtime)
523 self.assertEqual(1, os.stat(selections_path).st_mtime)
524 kill_server_process()
526 # Change the selections
527 sels_path = os.path.join(app.path, 'selections.xml')
528 with open(sels_path) as stream:
529 old = stream.read()
530 with open(sels_path, 'w') as stream:
531 stream.write(old.replace('Hello', 'Goodbye'))
533 # Trigger another background update - metadata changes found
534 reset_timestamps()
535 run_server('Hello.xml')
536 with trapped_exit(1):
537 dl = app.download_selections(sels)
538 assert dl == None
539 assert not ran_gui
540 self.assertNotEqual(1, os.stat(timestamp).st_mtime)
541 self.assertNotEqual(1, os.stat(selections_path).st_mtime)
542 kill_server_process()
544 # Trigger another background update - GUI needed now
546 # Delete cached implementation so we need to download it again
547 stored = sels.selections['http://example.com:8000/Hello.xml'].get_path(self.config.stores)
548 assert os.path.basename(stored).startswith('sha1')
549 ro_rmtree(stored)
551 # Replace with a valid local feed so we don't have to download immediately
552 with open(sels_path, 'w') as stream:
553 stream.write(local_hello)
554 sels = app.get_selections()
556 os.environ['DISPLAY'] = 'dummy'
557 reset_timestamps()
558 run_server('Hello.xml')
559 with trapped_exit(1):
560 dl = app.download_selections(sels)
561 assert dl == None
562 assert ran_gui # (so doesn't actually update)
563 kill_server_process()
565 # Now again with no DISPLAY
566 reset_timestamps()
567 del os.environ['DISPLAY']
568 run_server('Hello.xml', 'HelloWorld.tgz')
569 with trapped_exit(1):
570 dl = app.download_selections(sels)
571 assert dl == None
572 assert ran_gui # (so doesn't actually update)
574 self.assertNotEqual(1, os.stat(timestamp).st_mtime)
575 self.assertNotEqual(1, os.stat(selections_path).st_mtime)
576 kill_server_process()
578 sels = app.get_selections()
579 sel, = sels.selections.values()
580 self.assertEqual("sha1=3ce644dc725f1d21cfcf02562c76f375944b266a", sel.id)
582 # Untrust the key
583 trust.trust_db.untrust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'example.com:8000')
585 os.environ['DISPLAY'] = 'dummy'
586 reset_timestamps()
587 run_server('Hello.xml')
588 with trapped_exit(1):
589 #import logging; logging.getLogger().setLevel(logging.INFO)
590 dl = app.download_selections(sels)
591 assert dl == None
592 assert ran_gui
593 kill_server_process()
595 # Update not triggered because of last-check-attempt
596 ran_gui = False
597 os.utime(timestamp, (1, 1)) # 1970
598 os.utime(selections_path, (1, 1))
599 dl = app.download_selections(sels)
600 assert dl == None
601 assert not ran_gui
603 if __name__ == '__main__':
604 try:
605 unittest.main()
606 finally:
607 kill_server_process()