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

chrome/test/chromedriver/test/run_py_tests.py

 #!/usr/bin/env python
 # Copyright 2013 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.
 
 """End to end tests for ChromeDriver."""
 
 import base64
 import json
 import math
@@ skipping 865 matching lines @@
     self.assertEquals(latency, network['latency']);
     self.assertEquals(throughput, network['download_throughput']);
     self.assertEquals(throughput, network['upload_throughput']);
     self.assertEquals(False, network['offline']);
 
     # Network Conditions again cannot be retrieved after they've been deleted.
     self._driver.DeleteNetworkConditions()
     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
     #throughput = 2048 * 1024
     self._driver.SetNetworkConditionsName('DSL')
 
     network = self._driver.GetNetworkConditions()
     self.assertEquals(5, network['latency']);
     self.assertEquals(2048*1024, network['download_throughput']);
     self.assertEquals(2048*1024, network['upload_throughput']);
@@ skipping 650 matching lines @@
   """
 
   @staticmethod
   def GlobalSetUp():
     def respondWithUserAgentString(request):
       return {}, """
         <html>
         <body>%s</body>
         </html>""" % request.GetHeader('User-Agent')
 
+    #def respondWithHello(request):
+    #  return {}, """
+    #    <html>
+    #    <body>%s</body>
+    #    </html>""" % hello
+
     def respondWithUserAgentStringUseDeviceWidth(request):
       return {}, """
         <html>
         <head>
         <meta name="viewport" content="width=device-width,minimum-scale=1.0">
         </head>
         <body>%s</body>
         </html>""" % request.GetHeader('User-Agent')
 
     MobileEmulationCapabilityTest._http_server = webserver.WebServer(
         chrome_paths.GetTestData())
     MobileEmulationCapabilityTest._http_server.SetCallbackForPath(
         '/userAgent', respondWithUserAgentString)
     MobileEmulationCapabilityTest._http_server.SetCallbackForPath(
         '/userAgentUseDeviceWidth', respondWithUserAgentStringUseDeviceWidth)
+    #testNetworkConnectionAcrossTabs._http_server.SetCallbackForPath(
+    #   '/hello', respondWithHello)
 
   @staticmethod
   def GlobalTearDown():
     MobileEmulationCapabilityTest._http_server.Shutdown()
 
   def testDeviceMetricsWithStandardWidth(self):
     driver = self.CreateDriver(
         mobile_emulation = {
             'deviceMetrics': {'width': 360, 'height': 640, 'pixelRatio': 3},
             'userAgent': 'Mozilla/5.0 (Linux; Android 4.2.1; en-us; Nexus 5 Bui'
@@ skipping 351 matching lines @@
 
   all_tests_suite = unittest.defaultTestLoader.loadTestsFromModule(
       sys.modules[__name__])
   tests = unittest_util.FilterTestSuite(all_tests_suite, options.filter)
   ChromeDriverTest.GlobalSetUp()
   MobileEmulationCapabilityTest.GlobalSetUp()
   result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(tests)
   ChromeDriverTest.GlobalTearDown()
   MobileEmulationCapabilityTest.GlobalTearDown()
   sys.exit(len(result.failures) + len(result.errors))