Proxy BBT fetch requests on Android using adb reverse

b/165629644

cobalt/black_box_tests/black_box_cobalt_runner.py

@@ -43,14 +43,17 @@ class BlackBoxCobaltRunner(cobalt_runner.CobaltRunner):
   class AssertException(Exception):
     """Raised when assert condition fails."""
 
-  def __init__(self,
-               launcher_params,
-               url,
-               log_file=None,
-               target_params=None,
-               success_message=None,
-               poll_until_wait_seconds=POLL_UNTIL_WAIT_SECONDS,
-               **kwargs):
+  def __init__(
+      self,
+      launcher_params,
+      url,
+      log_file=None,
+      target_params=None,
+      success_message=None,
+      poll_until_wait_seconds=POLL_UNTIL_WAIT_SECONDS,
+      web_server_port=None,
+      **kwargs,
+  ):
     # For black box tests, don't log inline script warnings, we intend to
     # explicitly control timings for suspends and resumes, so we are not
     # concerned about a "suspend at the wrong time".
@@ -65,7 +68,9 @@ def __init__(self,
         target_params,
         success_message,
         poll_until_wait_seconds=poll_until_wait_seconds,
-        **kwargs)
+        web_server_port=web_server_port,
+        **kwargs,
+    )
 
     self.poll_until_wait_seconds = poll_until_wait_seconds
 
@@ -82,10 +87,12 @@ def PollUntilFoundOrTestsFailedWithReconnects(self, css_selector):
         if self.FindElements(css_selector):
           break
         is_failed = self.JSTestsFailed()
-      except (cobalt_runner.CobaltRunner.AssertException,
-              selenium_exceptions.NoSuchElementException,
-              selenium_exceptions.NoSuchWindowException,
-              selenium_exceptions.WebDriverException) as e:
+      except (
+          cobalt_runner.CobaltRunner.AssertException,
+          selenium_exceptions.NoSuchElementException,
+          selenium_exceptions.NoSuchWindowException,
+          selenium_exceptions.WebDriverException,
+      ) as e:
         # If the page is reloaded, the webdriver client status becomes
         # stale and should be reconnected.
         logging.warning(e)
@@ -103,17 +110,18 @@ def JSTestsSucceeded(self):
     # JavaScript test logic completion.
     self.PollUntilFound('[' + _TEST_STATUS_ELEMENT_NAME + ']')
     body_element = self.UniqueFind('body')
-    return body_element.get_attribute(
-        _TEST_STATUS_ELEMENT_NAME) == _JS_TEST_SUCCESS_MESSAGE
+    return (body_element.get_attribute(_TEST_STATUS_ELEMENT_NAME) ==
+            _JS_TEST_SUCCESS_MESSAGE)
 
   def JSTestsFailed(self):
     """Check failed test assertion in HTML page."""
 
     # Call onTestEnd() in black_box_js_test_utils.js to unblock the waiting for
     # JavaScript test logic completion.
     body_element = self.UniqueFind('body')
-    return body_element and body_element.get_attribute(
-        _TEST_STATUS_ELEMENT_NAME) == _JS_TEST_FAILURE_MESSAGE
+    return (body_element and
+            body_element.get_attribute(_TEST_STATUS_ELEMENT_NAME)
+            == _JS_TEST_FAILURE_MESSAGE)
 
   def WaitForJSTestsSetup(self):
     """Poll setup status until JavaScript gives green light."""

cobalt/black_box_tests/black_box_tests.py

@@ -34,7 +34,6 @@
 from starboard.tools import log_level
 
 _DISABLED_BLACKBOXTEST_CONFIGS = [
-    'android-arm/devel',
     'android-arm64/devel',
     'android-x86/devel',
     'evergreen-arm/devel',
@@ -146,7 +145,11 @@ def tearDownClass(cls):
     super(BlackBoxTestCase, cls).tearDownClass()
     logging.info('Done %s', cls.__name__)
 
-  def CreateCobaltRunner(self, url=None, target_params=None, **kwargs):
+  def CreateCobaltRunner(self,
+                         url=None,
+                         target_params=None,
+                         web_server_port=None,
+                         **kwargs):
     all_target_params = list(target_params) if target_params else []
     if _launcher_params.target_params is not None:
       all_target_params += _launcher_params.target_params
@@ -155,7 +158,9 @@ def CreateCobaltRunner(self, url=None, target_params=None, **kwargs):
         launcher_params=_launcher_params,
         url=url,
         target_params=all_target_params,
-        **kwargs)
+        web_server_port=web_server_port,
+        **kwargs,
+    )
 
   def GetBindingAddress(self):
     return _server_binding_address
@@ -175,7 +180,8 @@ def LoadTests(launcher_params, test_set):
       out_directory=launcher_params.out_directory,
       loader_platform=launcher_params.loader_platform,
       loader_config=launcher_params.loader_config,
-      loader_out_directory=launcher_params.loader_out_directory)
+      loader_out_directory=launcher_params.loader_out_directory,
+  )
 
   test_targets = []
 
@@ -212,7 +218,8 @@ def LoadEvergreenEndToEndTests(launcher_params):
       loader_out_directory=launcher_params.loader_out_directory,
       # The more lightweight elf_loader_sandbox can't be used since it has no
       # knowledge of updates or installations.
-      loader_target='loader_app')
+      loader_target='loader_app',
+  )
 
   test_targets = _TESTS_EVERGREEN_END_TO_END
 
@@ -227,10 +234,9 @@ class BlackBoxTests(object):
   """Helper class to run all black box tests and return results."""
 
   def __init__(self, args):
-
     self.args = args
 
-    #TODO(b/137905502): These globals should be refactored
+    # TODO(b/137905502): These globals should be refactored
     # Setup global variables used by test cases.
     global _launcher_params
     _launcher_params = command_line.CreateLauncherParams()
@@ -273,9 +279,12 @@ def __init__(self, args):
     # Test domains used in web platform tests to be resolved to the server
     # binding address.
     hosts = [
-        'web-platform.test', 'www.web-platform.test', 'www1.web-platform.test',
-        'www2.web-platform.test', 'xn--n8j6ds53lwwkrqhv28a.web-platform.test',
-        'xn--lve-6lad.web-platform.test'
+        'web-platform.test',
+        'www.web-platform.test',
+        'www1.web-platform.test',
+        'www2.web-platform.test',
+        'xn--n8j6ds53lwwkrqhv28a.web-platform.test',
+        'xn--lve-6lad.web-platform.test',
     ]
     self.host_resolve_map = {host: _server_binding_address for host in hosts}
 
@@ -292,7 +301,9 @@ def Run(self):
       run_cobalt_tests = False
       logging.warning(
           'Cobalt blackbox tests disabled for platform:%s config:%s',
-          _launcher_params.platform, _launcher_params.config)
+          _launcher_params.platform,
+          _launcher_params.config,
+      )
 
     if launch_config in _EVERGREEN_COMPATIBLE_CONFIGS:
       run_evergreen_tests = self.args.test_set in ['all', 'evergreen']
@@ -304,22 +315,22 @@ def LoadAndRunTests():
       if self.args.test_name:
         suite = unittest.TestLoader().loadTestsFromName(_TEST_DIR_PATH +
                                                         self.args.test_name)
-        return_code = not unittest.TextTestRunner(
-            verbosity=2, stream=sys.stdout).run(suite).wasSuccessful()
+        return_code = (not unittest.TextTestRunner(
+            verbosity=2, stream=sys.stdout).run(suite).wasSuccessful())
         return return_code
       else:
         cobalt_tests_return_code = 0
         if run_cobalt_tests:
           suite = LoadTests(_launcher_params, self.args.test_set)
           # Using verbosity=2 to log individual test function names and results.
-          cobalt_tests_return_code = not unittest.TextTestRunner(
-              verbosity=2, stream=sys.stdout).run(suite).wasSuccessful()
+          cobalt_tests_return_code = (not unittest.TextTestRunner(
+              verbosity=2, stream=sys.stdout).run(suite).wasSuccessful())
 
         evergreen_tests_return_code = 0
         if run_evergreen_tests:
           suite = LoadEvergreenEndToEndTests(_launcher_params)
-          evergreen_tests_return_code = not unittest.TextTestRunner(
-              verbosity=2, stream=sys.stdout).run(suite).wasSuccessful()
+          evergreen_tests_return_code = (not unittest.TextTestRunner(
+              verbosity=2, stream=sys.stdout).run(suite).wasSuccessful())
 
         return cobalt_tests_return_code or evergreen_tests_return_code
 
@@ -328,7 +339,8 @@ def LoadAndRunTests():
       with ProxyServer(
           port=self.proxy_port,
           host_resolve_map=self.host_resolve_map,
-          client_ips=self.args.device_ips):
+          client_ips=self.args.device_ips,
+      ):
         return LoadAndRunTests()
     else:
       return LoadAndRunTests()
@@ -357,8 +369,10 @@ def GetUnusedPort(self, addresses):
             break
         if unused:
           return port
-      logging.error('Can not find unused port on addresses within %s attempts.',
-                    _PORT_SELECTION_RETRY_LIMIT)
+      logging.error(
+          'Can not find unused port on addresses within %s attempts.',
+          _PORT_SELECTION_RETRY_LIMIT,
+      )
       return -1
     finally:
       for sock in socks:
@@ -394,45 +408,52 @@ def main():
   parser.add_argument(
       '--server_binding_address',
       default='127.0.0.1',
-      help='Binding address used to create the test server.')
+      help='Binding address used to create the test server.',
+  )
   parser.add_argument(
       '--proxy_address',
       default=None,
       help=('Address to the proxy server that all black box'
             'tests are run through. If not specified, the'
-            'server binding address is used.'))
+            'server binding address is used.'),
+  )
   parser.add_argument(
       '--proxy_port',
       default=None,
       help=('Port used to create the proxy server that all'
             'black box tests are run through. If not'
-            'specified, a random free port is used.'))
+            'specified, a random free port is used.'),
+  )
   parser.add_argument(
       '--test_name',
       default=None,
-      help=('Name of test to be run. If not specified, all '
-            'tests are run.'))
+      help='Name of test to be run. If not specified, all tests are run.',
+  )
   parser.add_argument(
       '--wpt_http_port',
       default=None,
       help=('Port used to create the web platform test http'
             'server. If not specified, a random free port is'
-            'used.'))
+            'used.'),
+  )
   parser.add_argument(
       '--device_id',
       default=None,
       help=('ID of test device to connect. If specified, it will be passed '
-            'as --dev_servers_listen_ip param on the test device.'))
+            'as --dev_servers_listen_ip param on the test device.'),
+  )
   parser.add_argument(
       '--device_ips',
       default=None,
       nargs='*',
       help=('IPs of test devices that will be allowed to connect. If not '
-            'specified, all IPs will be allowed to connect.'))
+            'specified, all IPs will be allowed to connect.'),
+  )
   parser.add_argument(
       '--test_set',
       choices=['all', 'wpt', 'blackbox', 'evergreen'],
-      default='all')
+      default='all',
+  )
   args, _ = parser.parse_known_args()
 
   log_level.InitializeLogging(args)

cobalt/black_box_tests/proxy_server.py

@@ -33,15 +33,22 @@
 class ProxyServer(object):
   """HTTP Proxy Server."""
 
-  def __init__(self,
-               hostname='0.0.0.0',
-               port='8000',
-               host_resolve_map=None,
-               client_ips=None):
+  def __init__(
+      self,
+      hostname='0.0.0.0',
+      port='8000',
+      host_resolve_map=None,
+      client_ips=None,
+  ):
     self.command = [
         'python3',
         os.path.join(SRC_DIR, 'third_party', 'proxy_py', 'proxy.py'),
-        '--hostname', hostname, '--port', port, '--log-level', 'WARNING'
+        '--hostname',
+        hostname,
+        '--port',
+        port,
+        '--log-level',
+        'WARNING',
     ]
     self.host_resolver_path = None
 

cobalt/black_box_tests/tests/allow_eval.py

@@ -25,9 +25,9 @@ class AllowEvalTest(black_box_tests.BlackBoxTestCase):
   """Ensure that client page can use eval() when CSP is missing."""
 
   def test_allow_eval(self):
-
     with ThreadedWebServer(binding_address=self.GetBindingAddress()) as server:
       url = server.GetURL(file_name='testdata/allow_eval.html')
 
-      with self.CreateCobaltRunner(url=url) as runner:
+      with self.CreateCobaltRunner(
+          url=url, web_server_port=server.GetPort()) as runner:
         self.assertTrue(runner.JSTestsSucceeded())

cobalt/black_box_tests/tests/cancel_sync_loads_when_suspended.py

@@ -38,14 +38,14 @@
 
 import logging
 import os
-from six.moves import SimpleHTTPServer
-from six.moves.urllib.parse import urlparse
 import threading
 import traceback
 
 from cobalt.black_box_tests import black_box_tests
 from cobalt.black_box_tests.threaded_web_server import MakeRequestHandlerClass
 from cobalt.black_box_tests.threaded_web_server import ThreadedWebServer
+from six.moves import SimpleHTTPServer
+from six.moves.urllib.parse import urlparse
 
 _CANCEL_SYNC_LOADS_WHEN_SUSPENDED_HTML = 'cancel_sync_loads_when_suspended.html'
 _CANCEL_SYNC_LOADS_WHEN_SUSPENDED_JS = 'cancel_sync_loads_when_suspended.js'
@@ -90,17 +90,18 @@ def _load_page(self, webdriver, url):
       traceback.print_exc()
 
   def test_cancel_sync_loads_when_suspended(self):
-
     # Step 2. Start Cobalt, and point it to the socket created in Step 1.
     try:
       with ThreadedWebServer(JavascriptRequestDetector,
                              self.GetBindingAddress()) as server:
-        with self.CreateCobaltRunner(url='about:blank') as runner:
+        with self.CreateCobaltRunner(
+            url='about:blank', web_server_port=server.GetPort()) as runner:
           target_url = server.GetURL(file_name='../testdata/' +
                                      _CANCEL_SYNC_LOADS_WHEN_SUSPENDED_HTML)
           cobalt_launcher_thread = threading.Thread(
               target=CancelSyncLoadsWhenSuspended._load_page,
-              args=(self, runner.webdriver, target_url))
+              args=(self, runner.webdriver, target_url),
+          )
           cobalt_launcher_thread.start()
 
           # Step 3. Wait HTTP request for html resource.

cobalt/black_box_tests/tests/compression_test.py

@@ -125,10 +125,12 @@ def execute_test(test_runner, encoding_type):
   path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
   with ThreadedWebServer(
       binding_address=test_runner.GetBindingAddress(),
-      handler=make_request_handler_class(path, encoding_type)) as server:
+      handler=make_request_handler_class(path, encoding_type),
+  ) as server:
     url = server.GetURL(file_name='testdata/compression.html')
 
-    with test_runner.CreateCobaltRunner(url=url) as runner:
+    with test_runner.CreateCobaltRunner(
+        url=url, web_server_port=server.GetPort()) as runner:
       test_runner.assertTrue(runner.JSTestsSucceeded())
 
 

cobalt/black_box_tests/tests/conceal_visibility.py

@@ -25,11 +25,11 @@ class ConcealVisibilityTest(black_box_tests.BlackBoxTestCase):
   """Verify correct visibility changes during and after conceal event."""
 
   def test_conceal_visibility(self):
-
     with ThreadedWebServer(binding_address=self.GetBindingAddress()) as server:
       url = server.GetURL(file_name='testdata/conceal_visibility.html')
 
-      with self.CreateCobaltRunner(url=url) as runner:
+      with self.CreateCobaltRunner(
+          url=url, web_server_port=server.GetPort()) as runner:
         runner.WaitForJSTestsSetup()
         runner.SendConceal()
         runner.SendFocus()