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..684f887ed2b90febb876fd21854749c3d4ea4752 100755
--- a/chrome/test/chromedriver/test/run_py_tests.py
+++ b/chrome/test/chromedriver/test/run_py_tests.py
@@ -883,6 +883,182 @@ 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'))
+
+    time.sleep(10)
+
+    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()
+
+    time.sleep(10)
+
+    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())
+
+    time.sleep(10)
+
+    # 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)
+
+    time.sleep(10)
+
+    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
@@ -1283,6 +1459,8 @@ class ChromeDriverPageLoadTimeoutTest(ChromeDriverBaseTestWithWebServer):
       self.assertNotEqual(-1, e.message.find('timeout'))
       # Verify that the browser actually made that request.
       self.assertTrue(request_received_event.wait(1))
+    except:
+      print 'something else happenend'

[samuong, 2016/06/28 23:23:40]

we forgot to delete this while we were debugging, please remove it before
landing

[roisinmcl, 2016/06/29 17:55:13]

Done.

     finally:
       send_response_event.set()
       pass