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