| #!/usr/bin/env python |
| |
| # Copyright (c) 2009 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import os |
| import unittest |
| |
| import dbus |
| import dbus.mainloop.glib |
| import dbus.service |
| import gobject |
| import mox |
| |
| |
| OFDP = 'org.freedesktop.DBus.Properties' |
| OFM = 'org.freedesktop.ModemManager' |
| OFMM = OFM + '.Modem' |
| |
| |
| class ModemManager(dbus.service.Object): |
| """Glue between DBUS ModemManager interface and a mock ModemManager.""" |
| |
| def __init__(self, mocker, *args, **kwargs): |
| dbus.service.Object.__init__(self, *args, **kwargs) |
| self.mock = mocker.CreateMock(ModemManager) |
| |
| @dbus.service.signal(OFM, signature='o') |
| def DeviceAdded(self, o): |
| pass # signal |
| |
| @dbus.service.signal(OFM, signature='o') |
| def DeviceRemoved(self, o): |
| pass # signal |
| |
| @dbus.service.method(OFM, in_signature='', out_signature='ao') |
| def EnumerateDevices(self): |
| return self.mock.EnumerateDevices() |
| |
| |
| class Modem(dbus.service.Object): |
| """Glue between DBUS Modem interface and a mock Modem.""" |
| |
| def __init__(self, mocker, *args, **kwargs): |
| dbus.service.Object.__init__(self, *args, **kwargs) |
| self.mock = mocker.CreateMock(Modem) |
| |
| @dbus.service.method(OFDP, in_signature='s') |
| def GetAll(self, interface): |
| return self.mock.GetAll(interface) |
| |
| @dbus.service.method(OFMM, in_signature='b') |
| def Enable(self, enable): |
| self.mock.Enable(enable) |
| |
| |
| class FlimflamTest(unittest.TestCase): |
| """Test suite for FlimFlam.""" |
| def setUp(self): |
| self._timed_out = 0 |
| |
| self._mocker = mox.Mox() |
| |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| self._mainloop = gobject.MainLoop() |
| |
| self._bus = dbus.SystemBus() |
| |
| # We don't have a good system for mocking signal receipt yet; |
| # pymox doesn't do "expect these calls and maybe some others" |
| # self._bus.add_signal_receiver(PrintSignal) |
| |
| self._name = dbus.service.BusName(OFM, self._bus) |
| |
| def QuitLoop(self, *_): |
| self._mainloop.quit() |
| |
| def TimedOut(self, *_): |
| print 'Timed out' |
| self._timed_out = 1 |
| self._mainloop.quit() |
| |
| def testEnable(self): |
| manager = ModemManager(self._mocker, |
| self._bus, |
| '/org/freedesktop/ModemManager') |
| modem = Modem(self._mocker, self._bus, '%s/%s/%d' % ( |
| manager.__dbus_object_path__, 'Modems', os.getpid())) |
| manager.mock.EnumerateDevices().AndReturn([modem.__dbus_object_path__]) |
| |
| modem.mock.GetAll(OFMM).AndReturn(dbus.Dictionary({ |
| 'Device': '/dev/ttyUSB0', |
| 'MasterDevice': 'really/long/path', |
| 'Driver': 'driver', |
| 'Type': dbus.UInt32(0), |
| 'IpMethod': dbus.UInt32(0), |
| }, signature='sv')) |
| modem.mock.Enable(True).WithSideEffects(self.QuitLoop) |
| |
| self._mocker.ReplayAll() |
| |
| def StartTest(): |
| manager.DeviceAdded(modem.__dbus_object_path__) |
| |
| gobject.timeout_add(1, StartTest) |
| gobject.timeout_add(2000, self.TimedOut) |
| self._mainloop.run() |
| |
| self._mocker.VerifyAll() |
| |
| # This is belt-and-suspenders; we should have failed mock |
| # verification already |
| self.assert_(not self._timed_out) |
| |
| |
| def main(): |
| unittest.main() |
| |
| |
| if __name__ == '__main__': |
| main() |