tests/basetest.py

   1 #!/usr/bin/env python
   2 import sys, tempfile, os, shutil, imp
   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
  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):
 101                 self.allowed_feed_downloads[url] = feed
 102
 103         def download_impls(self, impls, stores):
 104                 @tasks.async
 105                 def fake_download():
 106                         yield
 107                         for impl in impls:
 108                                 assert impl.id in self.allowed_downloads, impl
 109                                 self.allowed_downloads.remove(impl.id)
 110                                 self.config.stores.add_fake(impl.id)
 111                 return fake_download()
 112
 113         def download_and_import_feed(self, feed_url, iface_cache, force = False):
 114                 @tasks.async
 115                 def fake_download():
 116                         yield
 117                         assert feed_url in self.allowed_feed_downloads, feed_url
 118                         self.config.iface_cache._feeds[feed_url] = self.allowed_feed_downloads[feed_url]
 119                         del self.allowed_feed_downloads[feed_url]
 120                 return fake_download()
 121
 122         def fetch_key_info(self, fingerprint):
 123                 return DummyKeyInfo(fingerprint)
 124
 125 class TestStores:
 126         def __init__(self):
 127                 self.fake_impls = set()
 128
 129         def add_fake(self, digest):
 130                 self.fake_impls.add(digest)
 131
 132         def lookup_maybe(self, digests):
 133                 for d in digests:
 134                         if d in self.fake_impls:
 135                                 return '/fake_store/' + d
 136                 return None
 137
 138         def lookup_any(self, digests):
 139                 path = self.lookup_maybe(digests)
 140                 if path:
 141                         return path
 142                 raise NotStored()
 143
 144 class TestConfig:
 145         freshness = 0
 146         help_with_testing = False
 147         network_use = model.network_full
 148         key_info_server = None
 149         auto_approve_keys = False
 150
 151         def __init__(self):
 152                 self.iface_cache = iface_cache.IfaceCache()
 153                 self.handler = DummyHandler()
 154                 self.stores = Stores()
 155                 self.fetcher = TestFetcher(self)
 156                 self.trust_db = trust.trust_db
 157                 self.trust_mgr = trust.TrustMgr(self)
 158
 159 class BaseTest(unittest.TestCase):
 160         def setUp(self):
 161                 warnings.resetwarnings()
 162
 163                 self.config_home = tempfile.mktemp()
 164                 self.cache_home = tempfile.mktemp()
 165                 self.cache_system = tempfile.mktemp()
 166                 self.gnupg_home = tempfile.mktemp()
 167                 os.environ['GNUPGHOME'] = self.gnupg_home
 168                 os.environ['XDG_CONFIG_HOME'] = self.config_home
 169                 os.environ['XDG_CONFIG_DIRS'] = ''
 170                 os.environ['XDG_CACHE_HOME'] = self.cache_home
 171                 os.environ['XDG_CACHE_DIRS'] = self.cache_system
 172                 imp.reload(basedir)
 173                 assert basedir.xdg_config_home == self.config_home
 174
 175                 os.mkdir(self.config_home, 0o700)
 176                 os.mkdir(self.cache_home, 0o700)
 177                 os.mkdir(self.cache_system, 0o500)
 178                 os.mkdir(self.gnupg_home, 0o700)
 179
 180                 if 'DISPLAY' in os.environ:
 181                         del os.environ['DISPLAY']
 182
 183                 self.config = TestConfig()
 184                 policy._config = self.config    # XXX
 185                 iface_cache.iface_cache = self.config.iface_cache
 186
 187                 logging.getLogger().setLevel(logging.WARN)
 188
 189                 download._downloads = {}
 190
 191                 self.old_path = os.environ['PATH']
 192                 os.environ['PATH'] = dpkgdir + ':' + self.old_path
 193
 194                 distro._host_distribution = distro.DebianDistribution(dpkgdir + '/status')
 195                 distro._host_distribution._packagekit = DummyPackageKit()
 196
 197                 my_dbus.system_services = {}
 198
 199         def tearDown(self):
 200                 if self.config.handler.ex:
 201                         support.raise_with_traceback(self.config.handler.ex, self.config.handler.tb)
 202
 203                 shutil.rmtree(self.config_home)
 204                 support.ro_rmtree(self.cache_home)
 205                 shutil.rmtree(self.cache_system)
 206                 shutil.rmtree(self.gnupg_home)
 207
 208                 os.environ['PATH'] = self.old_path
 209
 210         def import_feed(self, url, path):
 211                 iface_cache = self.config.iface_cache
 212                 iface_cache.get_interface(url)
 213                 feed = iface_cache._feeds[url] = reader.load_feed(path)
 214                 return feed