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