tests/testpackagekit.py

   1 #!/usr/bin/env python
   2 from basetest import BaseTest
   3 import sys
   4 import unittest
   5
   6 sys.path.insert(0, '..')
   7 from zeroinstall.injector import packagekit, handler, model, fetch
   8 from zeroinstall.support import tasks
   9
  10 import dbus
  11
  12 class PackageKit05:
  13         x = 0
  14
  15         def GetTid(self):
  16                 self.x += 1
  17                 return "/tid/%d" % self.x
  18
  19         class Tid:
  20                 def __init__(self):
  21                         self.signals = {}
  22
  23                 def connect_to_signal(self, signal, cb):
  24                         self.signals[signal] = cb
  25
  26                 def get_dbus_method(self, method):
  27                         if hasattr(self, method):
  28                                 return getattr(self, method)
  29                         raise dbus.exceptions.DBusException('org.freedesktop.DBus.Error.UnknownMethod')
  30
  31                 def SetLocale(self, locale):
  32                         pass
  33
  34         class Tid1(Tid):
  35                 def Resolve(self, query, package_names):
  36                         @tasks.async
  37                         def later():
  38                                 yield
  39                                 result = "success"
  40                                 for package_name in package_names:
  41                                         if package_name == 'gimp':
  42                                                 info = {}
  43                                                 self.signals['Package'](info, "gimp;2.6.8-2ubuntu1.1;amd64;Ubuntu", "summary")
  44                                         else:
  45                                                 self.signals['Error']("package-not-found", "Package name %s could not be resolved" % package_name)
  46                                                 result = "failed"
  47                                 yield
  48                                 self.signals['Finished'](result, 100)
  49                         later()
  50
  51         class Tid2(Tid):
  52                 def GetDetails(self, package_ids):
  53                         @tasks.async
  54                         def later():
  55                                 yield
  56                                 for package_id in package_ids:
  57                                         assert package_id == "gimp;2.6.8-2ubuntu1.1;amd64;Ubuntu"
  58                                         self.signals['Details'](package_id, "GPL", "Graphics", "detail", "http://foo", 100)
  59
  60                                 yield
  61                                 self.signals['Finished']("success", 100)
  62                         later()
  63
  64         class Install(Tid):
  65                 def GetProgress(self):
  66                         # %task, %sub-task, time-task, time-sub-task
  67                         return (50, 101, 351, 0)
  68
  69                 def InstallPackages(self, only_trusted, package_ids = None):
  70                         if package_ids is None:
  71                                 # newer 2-arg form
  72                                 raise dbus.exceptions.DBusException('org.freedesktop.DBus.Error.UnknownMethod')
  73
  74                         assert only_trusted == False
  75                         @tasks.async
  76                         def later():
  77                                 yield
  78                                 for package_id in package_ids:
  79                                         assert package_id == "gimp;2.6.8-2ubuntu1.1;amd64;Ubuntu"
  80                                         self.signals['StatusChanged']("setup")
  81
  82                                         # Unknown % for task and subtask
  83                                         # 0s time used so far
  84                                         #self.signals['ProgressChanged'](101, 101, 0, 0)
  85
  86                                         yield
  87
  88                                         #self.signals['ProgressChanged'](50, 101, 351, 0)
  89
  90                                         #self.signals['AllowCancel'](False)
  91                                         self.signals['Package']("installing", "gimp;2.6.8-2ubuntu1.1;amd64;Ubuntu", "Graphics package")
  92
  93                                         yield
  94
  95                                         #self.signals['ProgressChanged'](100, 101, 1351, 0)
  96                                         self.signals['Package']("finished", "gimp;2.6.8-2ubuntu1.1;amd64;Ubuntu", "Graphics package")
  97
  98                                 yield
  99                                 self.signals['Finished']("success", 100)
 100                         later()
 101
 102 class TestPackageKit(BaseTest):
 103         def setUp(self):
 104                 BaseTest.setUp(self)
 105
 106         def tearDown(self):
 107                 sys.meta_path = []
 108                 BaseTest.tearDown(self)
 109
 110         def find_module(self, fullname, path=None):
 111                 if fullname.startswith('dbus'):
 112                         raise ImportError("No (dummy) D-BUS")
 113                 else:
 114                         return None
 115
 116         def testNoDBUS(self):
 117                 import dbus
 118                 old_dbus = dbus
 119                 try:
 120                         sys.meta_path = [self]
 121                         del sys.modules['dbus']
 122
 123                         reload(packagekit)
 124
 125                         pk = packagekit.PackageKit()
 126                         assert pk.available == False
 127
 128                         factory = Exception("not called")
 129                         pk.get_candidates('gimp', factory, 'package:null')
 130                 finally:
 131                         self.meta_path = []
 132                         sys.modules['dbus'] = old_dbus
 133
 134         def testNoPackageKit(self):
 135                 reload(packagekit)
 136                 pk = packagekit.PackageKit()
 137                 assert not pk.available
 138
 139                 factory = Exception("not called")
 140                 pk.get_candidates('gimp', factory, 'package:null')
 141
 142         def testPackageKit05(self):
 143                 #import logging; logging.getLogger().setLevel(logging.DEBUG)
 144
 145                 dbus.system_services['org.freedesktop.PackageKit'] = {
 146                         '/org/freedesktop/PackageKit': PackageKit05(),
 147                         '/tid/1': PackageKit05.Tid1(),
 148                         '/tid/2': PackageKit05.Tid2(),
 149                         '/tid/3': PackageKit05.Install(),
 150                 }
 151                 reload(packagekit)
 152                 pk = packagekit.PackageKit()
 153                 assert pk.available
 154
 155                 factory = Exception("not called")
 156                 pk.get_candidates('gimp', factory, 'package:test')
 157
 158                 blocker = pk.fetch_candidates(["gimp"])
 159                 tasks.wait_for_blocker(blocker)
 160                 tasks.check(blocker)
 161
 162                 impls = {}
 163                 def factory(impl_id, only_if_missing, installed):
 164                         assert impl_id.startswith('package:')
 165                         assert only_if_missing is True
 166                         assert installed is False
 167
 168                         feed = None
 169
 170                         impl = model.DistributionImplementation(feed, impl_id, self)
 171                         impl.installed = installed
 172                         impls[impl_id] = impl
 173                         return impl
 174
 175                 pk.get_candidates('gimp', factory, 'package:test')
 176                 self.assertEquals(["package:test:gimp:2.6.8-2:x86_64"], impls.keys())
 177
 178                 impl, = impls.values()
 179                 fetcher = fetch.Fetcher(config = self.config)
 180                 self.config.handler.allow_downloads = True
 181                 b = fetcher.download_impl(impl, impl.download_sources[0], stores = None)
 182                 tasks.wait_for_blocker(b)
 183                 tasks.check(b)
 184
 185 if __name__ == '__main__':
 186         unittest.main()