tests/testdownload.py

   1 #!/usr/bin/env python2.3
   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
   7 #getLogger().setLevel(DEBUG)
   8
   9 sys.path.insert(0, '..')
  10
  11 from zeroinstall.injector import model, basedir, autopolicy, gpg, iface_cache, download, reader, trust
  12 from zeroinstall.zerostore import Store; Store._add_with_helper = lambda *unused: False
  13 import data
  14
  15 import server
  16
  17 class Reply:
  18         def __init__(self, reply):
  19                 self.reply = reply
  20
  21         def readline(self):
  22                 return self.reply
  23
  24 class TestDownload(BaseTest):
  25         def setUp(self):
  26                 BaseTest.setUp(self)
  27
  28                 stream = tempfile.TemporaryFile()
  29                 stream.write(data.thomas_key)
  30                 stream.seek(0)
  31                 gpg.import_key(stream)
  32                 self.child = None
  33
  34                 trust.trust_db.watchers = []
  35
  36         def tearDown(self):
  37                 BaseTest.tearDown(self)
  38                 if self.child is not None:
  39                         os.kill(self.child, signal.SIGTERM)
  40                         os.waitpid(self.child, 0)
  41                         self.child = None
  42
  43         def testRejectKey(self):
  44                 old_out = sys.stdout
  45                 try:
  46                         sys.stdout = StringIO()
  47                         self.child = server.handle_requests('Hello', '6FCF121BE2390E0B.gpg')
  48                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello', download_only = False)
  49                         assert policy.need_download()
  50                         sys.stdin = Reply("N\n")
  51                         try:
  52                                 policy.download_and_execute(['Hello'])
  53                                 assert 0
  54                         except model.SafeException, ex:
  55                                 if "Not signed with a trusted key" not in str(ex):
  56                                         raise ex
  57                 finally:
  58                         sys.stdout = old_out
  59
  60         def testRejectKeyXML(self):
  61                 old_out = sys.stdout
  62                 try:
  63                         sys.stdout = StringIO()
  64                         self.child = server.handle_requests('Hello.xml', '6FCF121BE2390E0B.gpg')
  65                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello.xml', download_only = False)
  66                         assert policy.need_download()
  67                         sys.stdin = Reply("N\n")
  68                         try:
  69                                 policy.download_and_execute(['Hello'])
  70                                 assert 0
  71                         except model.SafeException, ex:
  72                                 if "Not signed with a trusted key" not in str(ex):
  73                                         raise
  74                 finally:
  75                         sys.stdout = old_out
  76
  77         def testImport(self):
  78                 old_out = sys.stdout
  79                 try:
  80                         from zeroinstall.injector import cli
  81                         import logging
  82
  83                         rootLogger = getLogger()
  84                         rootLogger.disabled = True
  85                         try:
  86                                 try:
  87                                         cli.main(['--import', '-v', 'NO-SUCH-FILE'])
  88                                         assert 0
  89                                 except model.SafeException, ex:
  90                                         assert 'NO-SUCH-FILE' in str(ex)
  91                         finally:
  92                                 rootLogger.disabled = False
  93                                 rootLogger.setLevel(WARN)
  94
  95                         hello = iface_cache.iface_cache.get_interface('http://localhost:8000/Hello')
  96                         self.assertEquals(0, len(hello.implementations))
  97
  98                         sys.stdout = StringIO()
  99                         self.child = server.handle_requests('6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
 100                         sys.stdin = Reply("Y\n")
 101
 102                         assert not trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
 103                         cli.main(['--import', 'Hello'])
 104                         assert trust.trust_db.is_trusted('DE937DD411906ACF7C263B396FCF121BE2390E0B')
 105
 106                         # Check we imported the interface after trusting the key
 107                         reader.update_from_cache(hello)
 108                         self.assertEquals(1, len(hello.implementations))
 109
 110                         # Shouldn't need to prompt the second time
 111                         sys.stdin = None
 112                         cli.main(['--import', 'Hello'])
 113                 finally:
 114                         sys.stdout = old_out
 115
 116         def testAcceptKey(self):
 117                 old_out = sys.stdout
 118                 try:
 119                         sys.stdout = StringIO()
 120                         self.child = server.handle_requests('Hello', '6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
 121                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello', download_only = False)
 122                         assert policy.need_download()
 123                         sys.stdin = Reply("Y\n")
 124                         try:
 125                                 policy.download_and_execute(['Hello'], main = 'Missing')
 126                                 assert 0
 127                         except model.SafeException, ex:
 128                                 if "HelloWorld/Missing" not in str(ex):
 129                                         raise ex
 130                 finally:
 131                         sys.stdout = old_out
 132
 133         def testRecipe(self):
 134                 old_out = sys.stdout
 135                 try:
 136                         sys.stdout = StringIO()
 137                         self.child = server.handle_requests(('HelloWorld.tar.bz2', 'dummy_1-1_all.deb'))
 138                         policy = autopolicy.AutoPolicy(os.path.abspath('Recipe.xml'), download_only = False)
 139                         try:
 140                                 policy.download_and_execute([])
 141                                 assert False
 142                         except model.SafeException, ex:
 143                                 if "HelloWorld/Missing" not in str(ex):
 144                                         raise ex
 145                 finally:
 146                         sys.stdout = old_out
 147
 148         def testSymlink(self):
 149                 old_out = sys.stdout
 150                 try:
 151                         sys.stdout = StringIO()
 152                         self.child = server.handle_requests(('HelloWorld.tar.bz2', 'HelloSym.tgz'))
 153                         policy = autopolicy.AutoPolicy(os.path.abspath('RecipeSymlink.xml'), download_only = False)
 154                         try:
 155                                 policy.download_and_execute([])
 156                                 assert False
 157                         except model.SafeException, ex:
 158                                 if 'Attempt to unpack dir over symlink "HelloWorld"' not in str(ex):
 159                                         raise ex
 160                         self.assertEquals(None, basedir.load_first_cache('0install.net', 'implementations', 'main'))
 161                 finally:
 162                         sys.stdout = old_out
 163
 164         def testAutopackage(self):
 165                 old_out = sys.stdout
 166                 try:
 167                         sys.stdout = StringIO()
 168                         self.child = server.handle_requests('HelloWorld.autopackage')
 169                         policy = autopolicy.AutoPolicy(os.path.abspath('Autopackage.xml'), download_only = False)
 170                         try:
 171                                 policy.download_and_execute([])
 172                                 assert False
 173                         except model.SafeException, ex:
 174                                 if "HelloWorld/Missing" not in str(ex):
 175                                         raise ex
 176                 finally:
 177                         sys.stdout = old_out
 178
 179         def testRecipeFailure(self):
 180                 old_out = sys.stdout
 181                 try:
 182                         sys.stdout = StringIO()
 183                         self.child = server.handle_requests('*')
 184                         policy = autopolicy.AutoPolicy(os.path.abspath('Recipe.xml'), download_only = False)
 185                         try:
 186                                 policy.download_and_execute([])
 187                                 assert False
 188                         except download.DownloadError, ex:
 189                                 if "Connection" not in str(ex):
 190                                         raise ex
 191                 finally:
 192                         sys.stdout = old_out
 193
 194 suite = unittest.makeSuite(TestDownload)
 195 if __name__ == '__main__':
 196         sys.argv.append('-v')
 197         unittest.main()