[chromedriver] Apply page load timeout to slow cross-process navigations

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 7d031b8115e7a43f149411f8e82ea60a372fe954..e2af94884caad00bc9abeff63f64287778fca138 100755
--- a/chrome/test/chromedriver/test/run_py_tests.py
+++ b/chrome/test/chromedriver/test/run_py_tests.py
@@ -254,13 +254,28 @@ class ChromeDriverBaseTest(unittest.TestCase):
     return False
 
 
-class ChromeDriverTest(ChromeDriverBaseTest):
+class ChromeDriverBaseTestWithWebServer(ChromeDriverBaseTest):
+
+  @staticmethod
+  def GlobalSetUp():
+    ChromeDriverBaseTestWithWebServer._http_server = webserver.WebServer(
+        chrome_paths.GetTestData())
+
+  @staticmethod
+  def GlobalTearDown():
+    ChromeDriverBaseTestWithWebServer._http_server.Shutdown()
+
+  @staticmethod
+  def GetHttpUrlForFile(file_path):
+    return ChromeDriverBaseTestWithWebServer._http_server.GetUrl() + file_path
+
+
+class ChromeDriverTest(ChromeDriverBaseTestWithWebServer):
   """End to end tests for ChromeDriver."""
 
   @staticmethod
   def GlobalSetUp():
-    ChromeDriverTest._http_server = webserver.WebServer(
-        chrome_paths.GetTestData())
+    ChromeDriverBaseTestWithWebServer.GlobalSetUp()
     ChromeDriverTest._sync_server = webserver.SyncWebServer()
     if _ANDROID_PACKAGE_KEY:
       ChromeDriverTest._device = device_utils.DeviceUtils.HealthyDevices()[0]
@@ -274,11 +289,7 @@ class ChromeDriverTest(ChromeDriverBaseTest):
   def GlobalTearDown():
     if _ANDROID_PACKAGE_KEY:
       forwarder.Forwarder.UnmapAllDevicePorts(ChromeDriverTest._device)
-    ChromeDriverTest._http_server.Shutdown()
-
-  @staticmethod
-  def GetHttpUrlForFile(file_path):
-    return ChromeDriverTest._http_server.GetUrl() + file_path
+    ChromeDriverBaseTestWithWebServer.GlobalTearDown()
 
   def setUp(self):
     self._driver = self.CreateDriver()
@@ -1213,6 +1224,48 @@ class ChromeDriverTest(ChromeDriverBaseTest):
     self.assertEquals('http://invalid./', self._driver.GetCurrentUrl())
 
 
+class ChromeDriverPageLoadTimeoutTest(ChromeDriverBaseTestWithWebServer):
+
+  def _CheckPageLoadTimeout(self, driver, host=None):
+    initial_url = self.GetHttpUrlForFile('/chromedriver/empty.html')
+    driver.Load(initial_url)
+
+    request_received_event = threading.Event()
+    send_response_event = threading.Event()
+    def hang(request):
+      request_received_event.set()
+      # Don't hang infinitely, 10 seconds are enough.
+      send_response_event.wait(10)
+      return {}, 'Hi!'
+
+    try:
+      self._http_server.SetCallbackForPath('/hang', hang)
+      # NB: Wuth a too small timeout chromedriver might not send the

[samuong, 2016/04/13 03:59:30]

"With" instead of "Wuth"?

[Alexander Semashko, 2016/04/13 15:51:35]

Done.

+      # Navigate command at all.
+      driver.SetTimeout('page load', 500) # 500 ms
+      driver.Load(self._http_server.GetUrl(host) + '/hang')
+    except chromedriver.ChromeDriverException as e:
+      self.assertNotEqual(-1, e.message.find('timeout'))
+      # Verify that the browser actually made that request.
+      self.assertTrue(request_received_event.wait(1))
+    finally:
+      send_response_event.set()
+      pass
+
+    self.assertEquals(initial_url, driver.GetCurrentUrl())
+
+  def testPageLoadTimeout(self):
+    self._CheckPageLoadTimeout(self.CreateDriver())
+
+  def testPageLoadTimeoutCrossDomain(self):
+    driver = self.CreateDriver(
+        chrome_switches=['host-resolver-rules=MAP * 127.0.0.1'])
+    # Cross-domain navigation is likely to be a cross-process one. In this case
+    # DevToolsAgentHost behaves quite differently and does not send command
+    # responses if the navigation hangs, so this case deserves a dedicated test.
+    self._CheckPageLoadTimeout(driver, 'foo.bar')
+
+
 class ChromeDriverAndroidTest(ChromeDriverBaseTest):
   """End to end tests for Android-specific tests."""