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