~didrocks/ubuntuone-client/use_result_var

« back to all changes in this revision

Viewing changes to tests/status/test_aggregator.py

  • Committer: Bazaar Package Importer
  • Author(s): Rodney Dawes
  • Date: 2011-03-18 15:32:09 UTC
  • mfrom: (1.1.47 upstream)
  • Revision ID: james.westby@ubuntu.com-20110318153209-eiao2x342xppviid
Tags: 1.5.7-0ubuntu1
* New upstream release.
  - Connection notifications are pointless (LP: #734895)
  - Use a single platform Notification instance (LP: #734985)
* 01_reset-udf-typeerror.patch:
  - Removed patch which is now included upstream.

Show diffs side-by-side

added added

removed removed

Lines of Context:
169
169
    def __init__(self):  # pylint: disable=W0231
170
170
        self.messages_shown = {}
171
171
        self.messages_updated = {}
172
 
 
 
172
        self.callbacks = []
173
173
    # pylint: disable=R0913
174
174
    def show_message(self, sender, callback=None, message_time=None,
175
175
                     message_count=None, icon=None):
184
184
        """Update the count for an existing indicator."""
185
185
        self.messages_updated[sender] = (sender, add_count)
186
186
 
 
187
    def _callback(self, indicator, message_time=None):
 
188
        """Fake callback."""
 
189
        self.callbacks.append((indicator, message_time))
 
190
 
 
191
    def create_callback(self):
 
192
        """Create the callback."""
 
193
        return self._callback
 
194
 
187
195
 
188
196
class FakeStatusAggregator(object):
189
197
    """A fake status aggregator."""
222
230
        """Reset the progress bubble."""
223
231
        self.restart_progress_bubble_called = True
224
232
 
225
 
    def build_notification(self):
 
233
    def get_notification(self):
226
234
        """Create a new toggleable notification object."""
227
 
        return self.notification_switch.build_notification()
 
235
        return self.notification_switch.get_notification()
228
236
 
229
237
 
230
238
class ToggleableNotificationTestCase(TestCase):
234
242
        """Initialize this test instance."""
235
243
        self.patch(aggregator, "Notification", FakeNotification)
236
244
        self.notification_switch = aggregator.NotificationSwitch()
237
 
        self.toggleable = self.notification_switch.build_notification()
 
245
        self.toggleable = self.notification_switch.get_notification()
238
246
 
239
247
    def assertShown(self, notification):
240
248
        """Assert that the notification was shown."""
274
282
        """Initialize this test instance."""
275
283
        self.notification_switch = aggregator.NotificationSwitch()
276
284
 
277
 
    def test_build_notification(self):
 
285
    def test_get_notification(self):
278
286
        """A new notification instance is returned."""
279
 
        notification = self.notification_switch.build_notification()
 
287
        notification = self.notification_switch.get_notification()
280
288
        self.assertEqual(notification.notification_switch,
281
289
                         self.notification_switch)
282
290
 
788
796
        """A new command was unqueued."""
789
797
        self.queued_commands.discard(command)
790
798
 
791
 
    def build_notification(self):
 
799
    def get_notification(self):
792
800
        """Create a new toggleable notification object."""
793
 
        return self.notification_switch.build_notification()
 
801
        return self.notification_switch.get_notification()
794
802
 
795
803
    def download_started(self, command):
796
804
        """A download just started."""
969
977
        self.assertEqual(None, msg[1])
970
978
        # msg did receive a count argument
971
979
        self.assertEqual(1, msg[2])
 
980
        # call callback
 
981
        print msg
 
982
        msg[0]('fake_indicator')
 
983
        self.assertEqual(1, len(self.status_frontend.messaging.callbacks))
972
984
 
973
985
    def test_two_new_udfs_available(self):
974
986
        """A new udf is available for subscription."""
1006
1018
        self.status_frontend.aggregator.connected = True
1007
1019
        self.listener.handle_SYS_CONNECTION_LOST()
1008
1020
        self.assertEqual(
1009
 
            1, len(self.status_frontend.notification.notifications_shown))
1010
 
        self.assertEqual(
1011
 
            (aggregator.UBUNTUONE_TITLE,
1012
 
             aggregator.ConnectionLostStatus.MESSAGE_ONE, None, False),
1013
 
            self.status_frontend.notification.notifications_shown[0])
 
1021
            0, len(self.status_frontend.notification.notifications_shown))
1014
1022
        self.assertFalse(self.status_frontend.aggregator.connected)
1015
1023
 
1016
1024
    def test_server_connection_made(self):
1018
1026
        self.status_frontend.aggregator.connected = False
1019
1027
        self.listener.handle_SYS_CONNECTION_MADE()
1020
1028
        self.assertEqual(
1021
 
            1, len(self.status_frontend.notification.notifications_shown))
1022
 
        self.assertEqual(
1023
 
            (aggregator.UBUNTUONE_TITLE,
1024
 
             aggregator.ConnectionMadeStatus.MESSAGE_ONE, None, False),
1025
 
            self.status_frontend.notification.notifications_shown[0])
 
1029
            0, len(self.status_frontend.notification.notifications_shown))
1026
1030
        self.assertTrue(self.status_frontend.aggregator.connected)
1027
1031
 
1028
1032
    def test_set_show_all_notifications(self):
1509
1513
            status99,
1510
1514
            status99,
1511
1515
            FakeStatus(12),
1512
 
            FakeStatus(1),
1513
 
        ]
 
1516
            FakeStatus(1)]
1514
1517
 
1515
 
        result = [ list(k) for j, k in aggregator.group_statuses(statuses) ]
 
1518
        result = [list(k) for _, k in aggregator.group_statuses(statuses)]
1516
1519
        expected = [
1517
1520
            [statuses[3]],
1518
1521
            [statuses[2]],
1519
 
            [status99, status99],
1520
 
        ]
 
1522
            [status99, status99]]
1521
1523
 
1522
1524
        self.assertEqual(result, expected)
1523
1525