[Sync] Add tests for migration triggered by notifications

net/tools/testserver/xmppserver_test.py

 #!/usr/bin/python2.4
 # Copyright (c) 2010 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Tests exercising the various classes in xmppserver.py."""
 
 import unittest
 
 import base64
@@ skipping 146 matching lines @@
     self.DoHandshake('resource_prefix', 'resource',
                      'foo', 'bar.com', 'baz.com', 'quux.com')
     self.DoHandshake('resource_prefix', 'resource',
                      'foo', 'bar.com', 'baz.com', '')
     self.DoHandshake('resource_prefix', 'resource',
                      'foo', 'bar.com', '', '')
     self.DoHandshake('resource_prefix', 'resource',
                      'foo', '', '', '')
 
 
-class XmppConnectionTest(unittest.TestCase):
+class FakeSocket(object):
+  """A fake socket object used for testing.
+  """
 
-  def setUp(self):
-    self.connections = set()
-    self.data = []
+  def __init__(self):
+    self._sent_data = []
+
+  def GetSentData(self):
+    return self._sent_data
 
   # socket-like methods.
   def fileno(self):
     return 0
 
   def setblocking(self, int):
     pass
 
   def getpeername(self):
     return ('', 0)
 
   def send(self, data):
-    self.data.append(data)
+    self._sent_data.append(data)
     pass
 
   def close(self):
     pass
 
+
+class XmppConnectionTest(unittest.TestCase):
+
+  def setUp(self):
+    self.connections = set()
+    self.fake_socket = FakeSocket()
+
   # XmppConnection delegate methods.
   def OnXmppHandshakeDone(self, xmpp_connection):
     self.connections.add(xmpp_connection)
 
   def OnXmppConnectionClosed(self, xmpp_connection):
     self.connections.discard(xmpp_connection)
 
   def ForwardNotification(self, unused_xmpp_connection, notification_stanza):
     for connection in self.connections:
       connection.ForwardNotification(notification_stanza)
 
   def testBasic(self):
     socket_map = {}
     xmpp_connection = xmppserver.XmppConnection(
-      self, socket_map, self, ('', 0))
+      self.fake_socket, socket_map, self, ('', 0))
     self.assertEqual(len(socket_map), 1)
     self.assertEqual(len(self.connections), 0)
     xmpp_connection.HandshakeDone(xmppserver.Jid('foo', 'bar'))
     self.assertEqual(len(socket_map), 1)
     self.assertEqual(len(self.connections), 1)
 
+    sent_data = self.fake_socket.GetSentData()
+
     # Test subscription request.
-    self.assertEqual(len(self.data), 0)
+    self.assertEqual(len(sent_data), 0)
     xmpp_connection.collect_incoming_data(
       '<iq><subscribe xmlns="google:push"></subscribe></iq>')
-    self.assertEqual(len(self.data), 1)
+    self.assertEqual(len(sent_data), 1)
 
     # Test acks.
     xmpp_connection.collect_incoming_data('<iq type="result"/>')
-    self.assertEqual(len(self.data), 1)
+    self.assertEqual(len(sent_data), 1)
 
     # Test notification.
     xmpp_connection.collect_incoming_data(
       '<message><push xmlns="google:push"/></message>')
-    self.assertEqual(len(self.data), 2)
+    self.assertEqual(len(sent_data), 2)
 
     # Test unexpected stanza.
     def SendUnexpectedStanza():
       xmpp_connection.collect_incoming_data('<foo/>')
     self.assertRaises(xmppserver.UnexpectedXml, SendUnexpectedStanza)
 
     # Test unexpected notifier command.
     def SendUnexpectedNotifierCommand():
       xmpp_connection.collect_incoming_data(
         '<iq><foo xmlns="google:notifier"/></iq>')
     self.assertRaises(xmppserver.UnexpectedXml,
                       SendUnexpectedNotifierCommand)
 
     # Test close
     xmpp_connection.close()
     self.assertEqual(len(socket_map), 0)
     self.assertEqual(len(self.connections), 0)
 
+
+class FakeXmppServer(xmppserver.XmppServer):
+  """A fake XMPP server object used for testing.
+  """
+
+  def __init__(self):
+    self._socket_map = {}
+    self._fake_sockets = set()
+    self._next_jid_suffix = 1
+    xmppserver.XmppServer.__init__(self, self._socket_map, ('', 0))
+
+  def GetSocketMap(self):
+    return self._socket_map
+
+  def GetFakeSockets(self):
+    return self._fake_sockets
+
+  def AddHandshakeCompletedConnection(self):
+    """Creates a new XMPP connection and completes its handshake.
+    """
+    xmpp_connection = self.handle_accept()
+    jid = xmppserver.Jid('user%s' % self._next_jid_suffix, 'domain.com')
+    self._next_jid_suffix += 1
+    xmpp_connection.HandshakeDone(jid)
+
+  # XmppServer overrides.
+  def accept(self):
+    fake_socket = FakeSocket()
+    self._fake_sockets.add(fake_socket)
+    return (fake_socket, ('', 0))
+
+  def close(self):
+    self._fake_sockets.clear()
+    xmppserver.XmppServer.close(self)
+
+
 class XmppServerTest(unittest.TestCase):
 
-  # socket-like methods.
-  def fileno(self):
-    return 0
-
-  def setblocking(self, int):
-    pass
-
-  def getpeername(self):
-    return ('', 0)
-
-  def close(self):
-    pass
-
   def testBasic(self):
-    class FakeXmppServer(xmppserver.XmppServer):
-      def accept(self2):
-        return (self, ('', 0))
-
-    socket_map = {}
-    self.assertEqual(len(socket_map), 0)
-    xmpp_server = FakeXmppServer(socket_map, ('', 0))
+    xmpp_server = FakeXmppServer()
+    socket_map = xmpp_server.GetSocketMap()
     self.assertEqual(len(socket_map), 1)
-    xmpp_server.handle_accept()
+    xmpp_server.AddHandshakeCompletedConnection()
     self.assertEqual(len(socket_map), 2)
     xmpp_server.close()
     self.assertEqual(len(socket_map), 0)
 
+  def testMakeNotification(self):
+    xmpp_server = FakeXmppServer()
+    notification = xmpp_server.MakeNotification('channel', 'data')
+    expected_xml = (
+      '<message>'
+      '  <push channel="channel" xmlns="google:push">'
+      '    <data>%s</data>'
+      '  </push>'
+      '</message>' % base64.b64encode('data'))
+    self.assertEqual(notification.toxml(), expected_xml)
+
+  def testSendNotification(self):
+    xmpp_server = FakeXmppServer()
+
+    # Add a few connections.
+    for _ in xrange(0, 7):
+      xmpp_server.AddHandshakeCompletedConnection()
+
+    fake_sockets = xmpp_server.GetFakeSockets()
+    self.assertEqual(len(fake_sockets), 7)
+
+    def AssertSentDataLength(expectedLength):

[Nicolas Zea, 2011/08/26 21:42:56]

Pull this out into it's own method since it's used below as well?

[akalin, 2011/08/26 23:42:09]

Done.

+      for fake_socket in fake_sockets:
+        self.assertEqual(len(fake_socket.GetSentData()), expectedLength)
+
+    AssertSentDataLength(0)
+    xmpp_server.SendNotification('channel', 'data')
+    AssertSentDataLength(1)
+
+  def testEnableDisableNotifications(self):
+    xmpp_server = FakeXmppServer()
+
+    # Add a few connections.
+    for _ in xrange(0, 5):
+      xmpp_server.AddHandshakeCompletedConnection()
+
+    fake_sockets = xmpp_server.GetFakeSockets()
+    self.assertEqual(len(fake_sockets), 5)
+
+    def AssertSentDataLength(expectedLength):
+      for fake_socket in fake_sockets:
+        self.assertEqual(len(fake_socket.GetSentData()), expectedLength)
+
+    AssertSentDataLength(0)
+    xmpp_server.SendNotification('channel', 'data')
+    AssertSentDataLength(1)
+
+    xmpp_server.EnableNotifications()
+    xmpp_server.SendNotification('channel', 'data')
+    AssertSentDataLength(2)
+
+    xmpp_server.DisableNotifications()
+    xmpp_server.SendNotification('channel', 'data')
+    AssertSentDataLength(2)
+
+    xmpp_server.DisableNotifications()
+    xmpp_server.SendNotification('channel', 'data')
+    AssertSentDataLength(2)
+
+    xmpp_server.EnableNotifications()
+    xmpp_server.SendNotification('channel', 'data')
+    AssertSentDataLength(3)
+
 
 if __name__ == '__main__':
   unittest.main()