Add a method to override the network conditions of the ChromeDriver session.

chrome/test/chromedriver/test/run_py_tests.py

Index: chrome/test/chromedriver/test/run_py_tests.py
diff --git a/chrome/test/chromedriver/test/run_py_tests.py b/chrome/test/chromedriver/test/run_py_tests.py
index 8c0acbdd2744c9cbe0d37f893c49390cdbfb18cc..7dfc0f663e52dd3519805199614e9f7c89dcf4cb 100755
--- a/chrome/test/chromedriver/test/run_py_tests.py
+++ b/chrome/test/chromedriver/test/run_py_tests.py
@@ -883,6 +883,174 @@ class ChromeDriverTest(ChromeDriverBaseTestWithWebServer):
     self.assertRaises(chromedriver.UnknownError,
                       self._driver.GetNetworkConditions)
 
+  def testEmulateNetworkConnection(self):
+    # Network conditions must be set before it can be retrieved.
+    self.assertRaises(chromedriver.UnknownError,
+                      self._driver.GetNetworkConditions)
+
+    # Test 4G connection.
+    connection_type = 0x8
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 20)
+    self.assertEquals(network['upload_throughput'], 4096 * 1024)
+    self.assertEquals(network['upload_throughput'], 4096 * 1024)
+    self.assertEquals(network['offline'], False)
+
+    # Test 3G connection.
+    connection_type = 0x10
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 100)
+    self.assertEquals(network['upload_throughput'], 750 * 1024)
+    self.assertEquals(network['upload_throughput'], 750 * 1024)
+    self.assertEquals(network['offline'], False)
+
+    # Test 2G connection.
+    connection_type = 0x20
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 300)
+    self.assertEquals(network['upload_throughput'], 250 * 1024)
+    self.assertEquals(network['upload_throughput'], 250 * 1024)
+    self.assertEquals(network['offline'], False)
+
+    # Connection with 4G, 3G, and 2G bits on.
+    # Tests that 4G takes precedence.
+    connection_type = 0x38
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 20)
+    self.assertEquals(network['upload_throughput'], 4096 * 1024)
+    self.assertEquals(network['upload_throughput'], 4096 * 1024)
+    self.assertEquals(network['offline'], False)
+
+    # Network Conditions again cannot be retrieved after they've been deleted.
+    self._driver.DeleteNetworkConditions()
+    self.assertRaises(chromedriver.UnknownError,
+                      self._driver.GetNetworkConditions)
+
+  def testWifiEmulation(self):
+    connection_type = 0x2
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 2)
+    self.assertEquals(network['upload_throughput'], 30720 * 1024)
+    self.assertEquals(network['download_throughput'], 30720 * 1024)
+    self.assertEquals(network['offline'], False)
+
+  def testAirplaneModeEmulation(self):
+    connection_type = 0x1
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 0)
+    self.assertEquals(network['upload_throughput'], 0)
+    self.assertEquals(network['download_throughput'], 0)
+    self.assertEquals(network['offline'], True)
+
+  def testWifiAndAirplaneModeEmulation(self):
+    # Connection with both Wifi and Airplane Mode on.
+    # Tests that Wifi takes precedence over Airplane Mode.
+    connection_type = 0x3
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 2)
+    self.assertEquals(network['upload_throughput'], 30720 * 1024)
+    self.assertEquals(network['download_throughput'], 30720 * 1024)
+    self.assertEquals(network['offline'], False)
+
+  def testNetworkConnectionAcrossTabs(self):
+    # Set network to online
+    connection_type = 0x10
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 100)
+    self.assertEquals(network['upload_throughput'], 750 * 1024)
+    self.assertEquals(network['offline'], False)
+
+    # Open a window with two divs counting successful + unsuccessful
+    # attempts to complete XML task
+    self._driver.Load(
+        self.GetHttpUrlForFile('/chromedriver/xmlrequest_test.html'))
+
+    def respondWithString(request):
+      return {}, """
+        <html>
+        <body>%s</body>
+        </html>""" % "hello world!"
+
+    self._http_server.SetCallbackForPath(
+      '/helloworld', respondWithString)
+
+    self.assertEquals(
+        1, self._driver.ExecuteScript('window.name = "oldWindow"; return 1;'))
+    window1_handle = self._driver.GetCurrentWindowHandle()
+    old_handles = self._driver.GetWindowHandles()
+    self._driver.FindElement('id', 'requestButton').Click()
+
+    self._driver.FindElement('id', 'link').Click()
+    new_window_handle = self.WaitForNewWindow(self._driver, old_handles)
+    self.assertNotEqual(None, new_window_handle)
+    self._driver.SwitchToWindow(new_window_handle)
+    self.assertEquals(new_window_handle, self._driver.GetCurrentWindowHandle())
+
+    # Set network to offline to determine whether the XML task continues to
+    # run in the background, indicating that the conditions are only applied
+    # to the current WebView
+    connection_type = 0x1
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 0)
+    self.assertEquals(network['offline'], True)
+
+    self._driver.SwitchToWindow(window1_handle)
+    connection_type = 0x1
+    self._driver.SetNetworkConnection(connection_type)
+    self.assertEquals(network['latency'], 0)
+    self.assertEquals(network['offline'], True)
+
+  def testNetworkConditionsDifferentWebViews(self):
+
+    self.assertRaises(chromedriver.UnknownError,
+                      self._driver.GetNetworkConditions)
+
+    self._driver.Load(self.GetHttpUrlForFile('/chromedriver/page_test.html'))
+    self.assertEquals(
+        1, self._driver.ExecuteScript('window.name = "oldWindow"; return 1;'))
+    window1_handle = self._driver.GetCurrentWindowHandle()
+    old_handles = self._driver.GetWindowHandles()
+
+    # Test connection is offline.
+    connection_type = 0x1;
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 0)
+    self.assertEquals(network['offline'], True)
+
+    # Navigate to another window.
+    self._driver.FindElement('id', 'link').Click()
+    new_window_handle = self.WaitForNewWindow(self._driver, old_handles)
+    self.assertNotEqual(None, new_window_handle)
+    self._driver.SwitchToWindow(new_window_handle)
+    self.assertEquals(new_window_handle, self._driver.GetCurrentWindowHandle())
+    self.assertRaises(
+        chromedriver.NoSuchElement, self._driver.FindElement, 'id', 'link')
+
+    # Set connection to 3G in second window.
+    connection_type = 0x10;
+    self._driver.SetNetworkConnection(connection_type)
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 100)
+    self.assertEquals(network['upload_throughput'], 750 * 1024)
+    self.assertEquals(network['offline'], False)
+
+    self._driver.SwitchToWindow('oldWindow')
+    self.assertEquals(window1_handle, self._driver.GetCurrentWindowHandle())
+
+    # Test whether first window has old or new network conditions.
+    network = self._driver.GetNetworkConditions()
+    self.assertEquals(network['latency'], 100)
+
   def testEmulateNetworkConditionsName(self):
     # DSL: 2Mbps throughput, 5ms RTT
     #latency = 5
@@ -1553,6 +1721,12 @@ class MobileEmulationCapabilityTest(ChromeDriverBaseTest):
         <body>%s</body>
         </html>""" % request.GetHeader('User-Agent')
 
+    #def respondWithHello(request):
+    #  return {}, """
+    #    <html>
+    #    <body>%s</body>
+    #    </html>""" % hello
+
     def respondWithUserAgentStringUseDeviceWidth(request):
       return {}, """
         <html>
@@ -1568,6 +1742,8 @@ class MobileEmulationCapabilityTest(ChromeDriverBaseTest):
         '/userAgent', respondWithUserAgentString)
     MobileEmulationCapabilityTest._http_server.SetCallbackForPath(
         '/userAgentUseDeviceWidth', respondWithUserAgentStringUseDeviceWidth)
+    #testNetworkConnectionAcrossTabs._http_server.SetCallbackForPath(
+    #   '/hello', respondWithHello)
 
   @staticmethod
   def GlobalTearDown():