tests/basetest.py

   1 #!/usr/bin/env python
   2 import sys, tempfile, os, shutil, StringIO
   3 import unittest
   4 import logging
   5 import warnings
   6 from xml.dom import minidom
   7 warnings.filterwarnings("ignore", message = 'The CObject type')
   8
   9 # Catch silly mistakes...
  10 os.environ['HOME'] = '/home/idontexist'
  11 os.environ['LANGUAGE'] = 'C'
  12
  13 sys.path.insert(0, '..')
  14 from zeroinstall.injector import qdom
  15 from zeroinstall.injector import iface_cache, download, distro, model, handler, policy, reader
  16 from zeroinstall.zerostore import NotStored, Store, Stores; Store._add_with_helper = lambda *unused: False
  17 from zeroinstall import support
  18 from zeroinstall.support import basedir, tasks
  19
  20 dpkgdir = os.path.join(os.path.dirname(__file__), 'dpkg')
  21
  22 empty_feed = qdom.parse(StringIO.StringIO("""<interface xmlns='http://zero-install.sourceforge.net/2004/injector/interface'>
  23 <name>Empty</name>
  24 <summary>just for testing</summary>
  25 </interface>"""))
  26
  27 import my_dbus
  28 sys.modules['dbus'] = my_dbus
  29 sys.modules['dbus.glib'] = my_dbus
  30 my_dbus.types = my_dbus
  31 sys.modules['dbus.types'] = my_dbus
  32 sys.modules['dbus.mainloop'] = my_dbus
  33 sys.modules['dbus.mainloop.glib'] = my_dbus
  34
  35 mydir = os.path.dirname(__file__)
  36
  37 # Catch us trying to run the GUI and return a dummy string instead
  38 old_execvp = os.execvp
  39 def test_execvp(prog, args):
  40         if prog.endswith('/0launch-gui'):
  41                 prog = os.path.join(mydir, 'test-gui')
  42         return old_execvp(prog, args)
  43
  44 os.execvp = test_execvp
  45
  46 test_locale = (None, None)
  47 assert model.locale
  48 class TestLocale:
  49         LC_ALL = 0      # Note: LC_MESSAGES not present on Windows
  50         def getlocale(self, x):
  51                 return test_locale
  52 model.locale = TestLocale()
  53
  54 class DummyPackageKit:
  55         available = False
  56
  57         def get_candidates(self, package, factory, prefix):
  58                 pass
  59
  60 class DummyHandler(handler.Handler):
  61         __slots__ = ['ex', 'tb', 'allow_downloads']
  62
  63         def __init__(self):
  64                 handler.Handler.__init__(self)
  65                 self.ex = None
  66                 self.allow_downloads = False
  67
  68         def get_download(self, url, force = False, hint = None, factory = None):
  69                 if self.allow_downloads:
  70                         return handler.Handler.get_download(self, url, force, hint, factory)
  71                 raise model.SafeException("DummyHandler: " + url)
  72
  73         def wait_for_blocker(self, blocker):
  74                 self.ex = None
  75                 handler.Handler.wait_for_blocker(self, blocker)
  76                 if self.ex:
  77                         raise self.ex, None, self.tb
  78
  79         def report_error(self, ex, tb = None):
  80                 assert self.ex is None, self.ex
  81                 self.ex = ex
  82                 self.tb = tb
  83
  84                 #import traceback
  85                 #traceback.print_exc()
  86
  87 class DummyKeyInfo:
  88         def __init__(self, fpr):
  89                 self.fpr = fpr
  90                 self.info = [minidom.parseString('<item vote="bad"/>')]
  91                 self.blocker = None
  92
  93 class TestFetcher:
  94         def __init__(self, config):
  95                 self.allowed_downloads = set()
  96                 self.allowed_feed_downloads = {}
  97                 self.config = config
  98
  99         def allow_download(self, digest):
 100                 assert isinstance(self.config.stores, TestStores)
 101                 self.allowed_downloads.add(digest)
 102
 103         def allow_feed_download(self, url, feed):
 104                 self.allowed_feed_downloads[url] = feed
 105
 106         def download_impls(self, impls, stores):
 107                 @tasks.async
 108                 def fake_download():
 109                         yield
 110                         for impl in impls:
 111                                 assert impl.id in self.allowed_downloads, impl
 112                                 self.allowed_downloads.remove(impl.id)
 113                                 self.config.stores.add_fake(impl.id)
 114                 return fake_download()
 115
 116         def download_and_import_feed(self, feed_url, iface_cache, force = False):
 117                 @tasks.async
 118                 def fake_download():
 119                         yield
 120                         assert feed_url in self.allowed_feed_downloads, feed_url
 121                         self.config.iface_cache._feeds[feed_url] = self.allowed_feed_downloads[feed_url]
 122                         del self.allowed_feed_downloads[feed_url]
 123                 return fake_download()
 124
 125         def fetch_key_info(self, fingerprint):
 126                 return DummyKeyInfo(fingerprint)
 127
 128 class TestStores:
 129         def __init__(self):
 130                 self.fake_impls = set()
 131
 132         def add_fake(self, digest):
 133                 self.fake_impls.add(digest)
 134
 135         def lookup_maybe(self, digests):
 136                 for d in digests:
 137                         if d in self.fake_impls:
 138                                 return '/fake_store/' + d
 139                 return None
 140
 141         def lookup_any(self, digests):
 142                 path = self.lookup_maybe(digests)
 143                 if path:
 144                         return path
 145                 raise NotStored()
 146
 147 class TestConfig:
 148         freshness = 0
 149         help_with_testing = False
 150         network_use = model.network_full
 151
 152         def __init__(self):
 153                 self.iface_cache = iface_cache.IfaceCache()
 154                 self.handler = DummyHandler()
 155                 self.stores = Stores()
 156                 self.fetcher = TestFetcher(self)
 157
 158 class BaseTest(unittest.TestCase):
 159         def setUp(self):
 160                 warnings.resetwarnings()
 161
 162                 self.config_home = tempfile.mktemp()
 163                 self.cache_home = tempfile.mktemp()
 164                 self.cache_system = tempfile.mktemp()
 165                 self.gnupg_home = tempfile.mktemp()
 166                 os.environ['GNUPGHOME'] = self.gnupg_home
 167                 os.environ['XDG_CONFIG_HOME'] = self.config_home
 168                 os.environ['XDG_CONFIG_DIRS'] = ''
 169                 os.environ['XDG_CACHE_HOME'] = self.cache_home
 170                 os.environ['XDG_CACHE_DIRS'] = self.cache_system
 171                 reload(basedir)
 172                 assert basedir.xdg_config_home == self.config_home
 173
 174                 os.mkdir(self.config_home, 0700)
 175                 os.mkdir(self.cache_home, 0700)
 176                 os.mkdir(self.cache_system, 0500)
 177                 os.mkdir(self.gnupg_home, 0700)
 178
 179                 if os.environ.has_key('DISPLAY'):
 180                         del os.environ['DISPLAY']
 181
 182                 self.config = TestConfig()
 183                 policy._config = self.config    # XXX
 184                 iface_cache.iface_cache = self.config.iface_cache
 185
 186                 logging.getLogger().setLevel(logging.WARN)
 187
 188                 download._downloads = {}
 189
 190                 self.old_path = os.environ['PATH']
 191                 os.environ['PATH'] = dpkgdir + ':' + self.old_path
 192
 193                 distro._host_distribution = distro.DebianDistribution(dpkgdir + '/status',
 194                                                                       dpkgdir + '/pkgcache.bin')
 195                 distro._host_distribution._packagekit = DummyPackageKit()
 196
 197                 my_dbus.system_services = {}
 198
 199         def tearDown(self):
 200                 assert self.config.handler.ex is None, self.config.handler.ex
 201
 202                 shutil.rmtree(self.config_home)
 203                 support.ro_rmtree(self.cache_home)
 204                 shutil.rmtree(self.cache_system)
 205                 shutil.rmtree(self.gnupg_home)
 206
 207                 os.environ['PATH'] = self.old_path
 208
 209         def import_feed(self, url, path):
 210                 iface_cache = self.config.iface_cache
 211                 iface_cache.get_interface(url)
 212                 iface_cache._feeds[url] = reader.load_feed(path)