tests/basetest.py

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