tests/testdownload.py

   1 #!/usr/bin/env python2.3
   2 import sys, tempfile, os, shutil
   3 from StringIO import StringIO
   4 import unittest, signal
   5 from logging import getLogger, DEBUG, INFO
   6 #getLogger().setLevel(DEBUG)
   7
   8 sys.path.insert(0, '..')
   9
  10 from zeroinstall.injector import model, basedir, autopolicy, gpg, iface_cache, download
  11 import data
  12
  13 import server
  14
  15 class Reply:
  16         def __init__(self, reply):
  17                 self.reply = reply
  18
  19         def readline(self):
  20                 return self.reply
  21
  22 class TestDownload(unittest.TestCase):
  23         def setUp(self):
  24                 self.config_home = tempfile.mktemp()
  25                 self.cache_home = tempfile.mktemp()
  26                 os.environ['XDG_CONFIG_HOME'] = self.config_home
  27                 os.environ['XDG_CACHE_HOME'] = self.cache_home
  28                 os.environ['XDG_CACHE_DIRS'] = ''
  29                 reload(basedir)
  30
  31                 os.mkdir(self.config_home, 0700)
  32                 os.mkdir(self.cache_home, 0700)
  33                 if os.environ.has_key('DISPLAY'):
  34                         del os.environ['DISPLAY']
  35                 self.gnupg_home = tempfile.mktemp()
  36                 os.environ['GNUPGHOME'] = self.gnupg_home
  37                 os.mkdir(self.gnupg_home, 0700)
  38                 stream = tempfile.TemporaryFile()
  39                 stream.write(data.thomas_key)
  40                 stream.seek(0)
  41                 gpg.import_key(stream)
  42                 iface_cache.iface_cache.__init__()
  43                 download._downloads = {}
  44                 self.child = None
  45
  46         def tearDown(self):
  47                 if self.child is not None:
  48                         os.kill(self.child, signal.SIGTERM)
  49                         os.waitpid(self.child, 0)
  50                         self.child = None
  51                 shutil.rmtree(self.config_home)
  52                 shutil.rmtree(self.cache_home)
  53                 shutil.rmtree(self.gnupg_home)
  54
  55         def testRejectKey(self):
  56                 old_out = sys.stdout
  57                 try:
  58                         sys.stdout = StringIO()
  59                         self.child = server.handle_requests('Hello', '6FCF121BE2390E0B.gpg')
  60                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello', download_only = False)
  61                         assert policy.need_download()
  62                         sys.stdin = Reply("N\n")
  63                         try:
  64                                 policy.download_and_execute(['Hello'])
  65                                 assert 0
  66                         except model.SafeException, ex:
  67                                 if "Not signed with a trusted key" not in str(ex):
  68                                         raise ex
  69                 finally:
  70                         sys.stdout = old_out
  71
  72         def testRejectKeyXML(self):
  73                 old_out = sys.stdout
  74                 try:
  75                         sys.stdout = StringIO()
  76                         self.child = server.handle_requests('Hello.xml', '6FCF121BE2390E0B.gpg')
  77                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello.xml', download_only = False)
  78                         assert policy.need_download()
  79                         sys.stdin = Reply("N\n")
  80                         try:
  81                                 policy.download_and_execute(['Hello'])
  82                                 assert 0
  83                         except model.SafeException, ex:
  84                                 if "Not signed with a trusted key" not in str(ex):
  85                                         raise ex
  86                 finally:
  87                         sys.stdout = old_out
  88
  89         def testAcceptKey(self):
  90                 old_out = sys.stdout
  91                 try:
  92                         sys.stdout = StringIO()
  93                         self.child = server.handle_requests('Hello', '6FCF121BE2390E0B.gpg', 'HelloWorld.tgz')
  94                         policy = autopolicy.AutoPolicy('http://localhost:8000/Hello', download_only = False)
  95                         assert policy.need_download()
  96                         sys.stdin = Reply("Y\n")
  97                         try:
  98                                 policy.download_and_execute(['Hello'], main = 'Missing')
  99                                 assert 0
 100                         except model.SafeException, ex:
 101                                 if "HelloWorld/Missing" not in str(ex):
 102                                         raise ex
 103                 finally:
 104                         sys.stdout = old_out
 105
 106         def testRecipe(self):
 107                 old_out = sys.stdout
 108                 try:
 109                         sys.stdout = StringIO()
 110                         self.child = server.handle_requests(('HelloWorld.tar.bz2', 'dummy_1-1_all.deb'))
 111                         policy = autopolicy.AutoPolicy(os.path.abspath('Recipe.xml'), download_only = False)
 112                         try:
 113                                 policy.download_and_execute([])
 114                                 assert False
 115                         except model.SafeException, ex:
 116                                 if "HelloWorld/Missing" not in str(ex):
 117                                         raise ex
 118                 finally:
 119                         sys.stdout = old_out
 120
 121         def testAutopackage(self):
 122                 old_out = sys.stdout
 123                 try:
 124                         sys.stdout = StringIO()
 125                         self.child = server.handle_requests('HelloWorld.autopackage')
 126                         policy = autopolicy.AutoPolicy(os.path.abspath('Autopackage.xml'), download_only = False)
 127                         try:
 128                                 policy.download_and_execute([])
 129                                 assert False
 130                         except model.SafeException, ex:
 131                                 if "HelloWorld/Missing" not in str(ex):
 132                                         raise ex
 133                 finally:
 134                         sys.stdout = old_out
 135
 136         def testRecipeFailure(self):
 137                 old_out = sys.stdout
 138                 try:
 139                         sys.stdout = StringIO()
 140                         self.child = server.handle_requests('HelloWorld.tar.bz2')
 141                         policy = autopolicy.AutoPolicy(os.path.abspath('Recipe.xml'), download_only = False)
 142                         try:
 143                                 policy.download_and_execute([])
 144                                 assert False
 145                         except download.DownloadError, ex:
 146                                 if "Connection" not in str(ex):
 147                                         raise ex
 148                 finally:
 149                         sys.stdout = old_out
 150
 151 suite = unittest.makeSuite(TestDownload)
 152 if __name__ == '__main__':
 153         sys.argv.append('-v')
 154         unittest.main()