tests/testdownload.py

   1 #!/usr/bin/env python2.4
   2 from basetest import BaseTest
   3 import sys, tempfile, os, shutil
   4 from StringIO import StringIO
   5 import unittest, signal
   6 from logging import getLogger, DEBUG, INFO, WARN, ERROR
   7
   8 sys.path.insert(0, '..')
   9
  10 from zeroinstall.injector import model, autopolicy, gpg, iface_cache, download, reader, trust, handler, background, arch, selections, qdom
  11 from zeroinstall.zerostore import Store; Store._add_with_helper = lambda *unused: False
  12 from zeroinstall.support import basedir, tasks
  13 import data
  14 import my_dbus
  15
  16 import server
  17
  18 ran_gui = False
  19 def raise_gui(*args):
  20         global ran_gui
  21         ran_gui = True
  22 background._detach = lambda: False
  23 background._exec_gui = raise_gui
  24 sys.modules['dbus'] = my_dbus
  25 sys.modules['dbus.glib'] = my_dbus
  26 my_dbus.types = my_dbus
  27 sys.modules['dbus.types'] = my_dbus
  28
  29 class Reply:
  30         def __init__(self, reply):
  31                 self.reply = reply
  32
  33         def readline(self):
  34                 return self.reply
  35
  36 class DummyHandler(handler.Handler):
  37         __slots__ = ['ex', 'tb']
  38
  39         def __init__(self):
  40                 handler.Handler.__init__(self)
  41                 self.ex = None
  42
  43         def wait_for_blocker(self, blocker):
  44                 self.ex = None
  45                 handler.Handler.wait_for_blocker(self, blocker)
  46                 if self.ex:
  47                         raise self.ex, None, self.tb
  48
  49         def report_error(self, ex, tb = None):
  50                 assert self.ex is None, self.ex
  51                 self.ex = ex
  52                 self.tb = tb
  53
  54                 #import traceback
  55                 #traceback.print_exc()
  56
  57 class TestDownload(BaseTest):
  58         def setUp(self):
  59                 BaseTest.setUp(self)
  60
  61                 stream = tempfile.TemporaryFile()
  62                 stream.write(data.thomas_key)
  63                 stream.seek(0)
  64                 gpg.import_key(stream)
  65                 self.child = None
  66
  67                 trust.trust_db.watchers = []
  68
  69         def tearDown(self):
  70                 BaseTest.tearDown(self)
  71                 if self.child is not None:
  72                         os.kill(self.child, signal.SIGTERM)
  73                         os.waitpid(self.child, 0)
  74                         self.child = None
  75
  76         def testRejectKey(self):
  77                 old_out = sys.stdout
  78                 try:
  79                         sys.stdout = StringIO()
  80                         self.child = server.handle_requests('Hello', '6FCF121BE2390E0B.gpg')
  81                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello', download_only = False,
  82                                                        handler = DummyHandler())
  83                         assert policy.need_download()
  84                         sys.stdin = Reply("N\n")
  85                         try:
  86                                 policy.download_and_execute(['Hello'])
  87                                 assert 0
  88                         except model.SafeException, ex:
  89                                 if "Not signed with a trusted key" not in str(ex):
  90                                         raise ex
  91                 finally:
  92                         sys.stdout = old_out
  93
  94         def testRejectKeyXML(self):
  95                 old_out = sys.stdout
  96                 try:
  97                         sys.stdout = StringIO()
  98                         self.child = server.handle_requests('Hello.xml', '6FCF121BE2390E0B.gpg')
  99                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello.xml', download_only = False,
 100                                                        handler = DummyHandler())
 101                         assert policy.need_download()
 102                         sys.stdin = Reply("N\n")
 103                         try:
 104                                 policy.download_and_execute(['Hello'])
 105                                 assert 0
 106                         except model.SafeException, ex:
 107                                 if "Not signed with a trusted key" not in str(ex):
 108                                         raise
 109                 finally:
 110                         sys.stdout = old_out
 111
 112         def testImport(self):
 113                 old_out = sys.stdout
 114                 try:
 115                         from zeroinstall.injector import cli
 116                         import logging
 117
 118                         rootLogger = getLogger()
 119                         rootLogger.disabled = True
 120                         try:
 121                                 try:
 122                                         cli.main(['--import', '-v', 'NO-SUCH-FILE'])
 123                                         assert 0
 124                                 except model.SafeException, ex:
 125                                         assert 'NO-SUCH-FILE' in str(ex)
 126                         finally:
 127                                 rootLogger.disabled = False
 128                                 rootLogger.setLevel(WARN)
 129
 130                         hello = iface_cache.iface_cache.get_interface('http://localhost:8000/Hello')
 131                         self.assertEquals(0, len(hello.implementations))
 132
 133                         sys.stdout = StringIO()
 134                         self.child = server.handle_requests('6FCF121BE2390E0B.gpg')
 135                         sys.stdin = Reply("Y\n")
 136
 137                         assert not trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
 138                         cli.main(['--import', 'Hello'])
 139                         assert trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
 140
 141                         # Check we imported the interface after trusting the key
 142                         reader.update_from_cache(hello)
 143                         self.assertEquals(1, len(hello.implementations))
 144
 145                         # Shouldn't need to prompt the second time
 146                         sys.stdin = None
 147                         cli.main(['--import', 'Hello'])
 148                 finally:
 149                         sys.stdout = old_out
 150
 151         def testSelections(self):
 152                 from zeroinstall.injector.cli import _download_missing_selections
 153                 root = qdom.parse(file("selections.xml"))
 154                 sels = selections.Selections(root)
 155                 class Options: dry_run = False
 156
 157                 old_out = sys.stdout
 158                 try:
 159                         sys.stdout = StringIO()
 160                         self.child = server.handle_requests('Hello.xml', '6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
 161                         sys.stdin = Reply("Y\n")
 162                         _download_missing_selections(Options(), sels)
 163                         path = iface_cache.iface_cache.stores.lookup(sels.selections['http://localhost:8000/Hello.xml'].id)
 164                         assert os.path.exists(os.path.join(path, 'HelloWorld', 'main'))
 165
 166                         assert sels.download_missing(iface_cache.iface_cache, None) is None
 167                 finally:
 168                         sys.stdout = old_out
 169
 170         def testAcceptKey(self):
 171                 old_out = sys.stdout
 172                 try:
 173                         sys.stdout = StringIO()
 174                         self.child = server.handle_requests('Hello', '6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
 175                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello', download_only = False,
 176                                                         handler = DummyHandler())
 177                         assert policy.need_download()
 178                         sys.stdin = Reply("Y\n")
 179                         try:
 180                                 policy.download_and_execute(['Hello'], main = 'Missing')
 181                                 assert 0
 182                         except model.SafeException, ex:
 183                                 if "HelloWorld/Missing" not in str(ex):
 184                                         raise ex
 185                 finally:
 186                         sys.stdout = old_out
 187
 188         def testRecipe(self):
 189                 old_out = sys.stdout
 190                 try:
 191                         sys.stdout = StringIO()
 192                         self.child = server.handle_requests(('HelloWorld.tar.bz2', 'dummy_1-1_all.deb'))
 193                         policy = autopolicy.AutoPolicy(os.path.abspath('Recipe.xml'), download_only = False)
 194                         try:
 195                                 policy.download_and_execute([])
 196                                 assert False
 197                         except model.SafeException, ex:
 198                                 if "HelloWorld/Missing" not in str(ex):
 199                                         raise ex
 200                 finally:
 201                         sys.stdout = old_out
 202
 203         def testSymlink(self):
 204                 old_out = sys.stdout
 205                 try:
 206                         sys.stdout = StringIO()
 207                         self.child = server.handle_requests(('HelloWorld.tar.bz2', 'HelloSym.tgz'))
 208                         policy = autopolicy.AutoPolicy(os.path.abspath('RecipeSymlink.xml'), download_only = False,
 209                                                         handler = DummyHandler())
 210                         try:
 211                                 policy.download_and_execute([])
 212                                 assert False
 213                         except model.SafeException, ex:
 214                                 if 'Attempt to unpack dir over symlink "HelloWorld"' not in str(ex):
 215                                         raise ex
 216                         self.assertEquals(None, basedir.load_first_cache('0install.net', 'implementations', 'main'))
 217                 finally:
 218                         sys.stdout = old_out
 219
 220         def testAutopackage(self):
 221                 old_out = sys.stdout
 222                 try:
 223                         sys.stdout = StringIO()
 224                         self.child = server.handle_requests('HelloWorld.autopackage')
 225                         policy = autopolicy.AutoPolicy(os.path.abspath('Autopackage.xml'), download_only = False)
 226                         try:
 227                                 policy.download_and_execute([])
 228                                 assert False
 229                         except model.SafeException, ex:
 230                                 if "HelloWorld/Missing" not in str(ex):
 231                                         raise
 232                 finally:
 233                         sys.stdout = old_out
 234
 235         def testRecipeFailure(self):
 236                 old_out = sys.stdout
 237                 try:
 238                         sys.stdout = StringIO()
 239                         self.child = server.handle_requests('*')
 240                         policy = autopolicy.AutoPolicy(os.path.abspath('Recipe.xml'), download_only = False,
 241                                                         handler = DummyHandler())
 242                         try:
 243                                 policy.download_and_execute([])
 244                                 assert False
 245                         except download.DownloadError, ex:
 246                                 if "Connection" not in str(ex):
 247                                         raise
 248                 finally:
 249                         sys.stdout = old_out
 250
 251         def testMirrors(self):
 252                 old_out = sys.stdout
 253                 try:
 254                         sys.stdout = StringIO()
 255                         getLogger().setLevel(ERROR)
 256                         trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'localhost:8000')
 257                         self.child = server.handle_requests(server.Give404('/Hello.xml'), 'latest.xml', '/0mirror/keys/6FCF121BE2390E0B.gpg')
 258                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello.xml', download_only = False)
 259                         policy.fetcher.feed_mirror = 'http://localhost:8000/0mirror'
 260
 261                         refreshed = policy.solve_with_downloads()
 262                         policy.handler.wait_for_blocker(refreshed)
 263                         assert policy.ready
 264                 finally:
 265                         sys.stdout = old_out
 266
 267         def testReplay(self):
 268                 old_out = sys.stdout
 269                 try:
 270                         sys.stdout = StringIO()
 271                         getLogger().setLevel(ERROR)
 272                         iface = iface_cache.iface_cache.get_interface('http://localhost:8000/Hello.xml')
 273                         mtime = int(os.stat('Hello-new.xml').st_mtime)
 274                         iface_cache.iface_cache.update_interface_from_network(iface, file('Hello-new.xml').read(), mtime + 10000)
 275
 276                         trust.trust_db.trust_key('DE937DD411906ACF7C263B396FCF121BE2390E0B', 'localhost:8000')
 277                         self.child = server.handle_requests(server.Give404('/Hello.xml'), 'latest.xml', '/0mirror/keys/6FCF121BE2390E0B.gpg', 'Hello.xml')
 278                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello.xml', download_only = False)
 279                         policy.fetcher.feed_mirror = 'http://localhost:8000/0mirror'
 280
 281                         # Update from mirror (should ignore out-of-date timestamp)
 282                         refreshed = policy.fetcher.download_and_import_feed(iface.uri, iface_cache.iface_cache)
 283                         policy.handler.wait_for_blocker(refreshed)
 284
 285                         # Update from upstream (should report an error)
 286                         refreshed = policy.fetcher.download_and_import_feed(iface.uri, iface_cache.iface_cache)
 287                         try:
 288                                 policy.handler.wait_for_blocker(refreshed)
 289                                 raise Exception("Should have been rejected!")
 290                         except model.SafeException, ex:
 291                                 assert "New interface's modification time is before old version" in str(ex)
 292
 293                         # Must finish with the newest version
 294                         self.assertEquals(1209206132, iface_cache.iface_cache._get_signature_date(iface.uri))
 295                 finally:
 296                         sys.stdout = old_out
 297
 298         def testBackground(self, verbose = False):
 299                 p = autopolicy.AutoPolicy('http://localhost:8000/Hello.xml')
 300                 reader.update(iface_cache.iface_cache.get_interface(p.root), 'Hello.xml')
 301                 p.freshness = 0
 302                 p.network_use = model.network_minimal
 303                 p.solver.solve(p.root, arch.get_host_architecture())
 304                 assert p.ready
 305
 306                 @tasks.async
 307                 def choose_download(registed_cb, nid, actions):
 308                         try:
 309                                 assert actions == ['download', 'Download'], actions
 310                                 registed_cb(nid, 'download')
 311                         except:
 312                                 import traceback
 313                                 traceback.print_exc()
 314                         yield None
 315
 316                 global ran_gui
 317                 ran_gui = False
 318                 old_out = sys.stdout
 319                 try:
 320                         sys.stdout = StringIO()
 321                         self.child = server.handle_requests('Hello.xml', '6FCF121BE2390E0B.gpg')
 322                         my_dbus.user_callback = choose_download
 323                         pid = os.getpid()
 324                         old_exit = os._exit
 325                         def my_exit(code):
 326                                 # The background handler runs in the same process
 327                                 # as the tests, so don't let it abort.
 328                                 if os.getpid() == pid:
 329                                         raise SystemExit(code)
 330                                 # But, child download processes are OK
 331                                 old_exit(code)
 332                         try:
 333                                 try:
 334                                         os._exit = my_exit
 335                                         background.spawn_background_update(p, verbose)
 336                                         assert False
 337                                 except SystemExit, ex:
 338                                         self.assertEquals(1, ex.code)
 339                         finally:
 340                                 os._exit = old_exit
 341                 finally:
 342                         sys.stdout = old_out
 343                 assert ran_gui
 344
 345         def testBackgroundVerbose(self):
 346                 self.testBackground(verbose = True)
 347
 348 suite = unittest.makeSuite(TestDownload)
 349 if __name__ == '__main__':
 350         sys.argv.append('-v')
 351         unittest.main()