Reland of "[telemetry] Wire TsProxy through telemetry's network stack."

telemetry/telemetry/internal/platform/network_controller_backend_unittest.py

Index: telemetry/telemetry/internal/platform/network_controller_backend_unittest.py
diff --git a/telemetry/telemetry/internal/platform/network_controller_backend_unittest.py b/telemetry/telemetry/internal/platform/network_controller_backend_unittest.py
new file mode 100644
index 0000000000000000000000000000000000000000..3375ad02e5c25ea7cb7a7dc2a2d6532bf0b34aac
--- /dev/null
+++ b/telemetry/telemetry/internal/platform/network_controller_backend_unittest.py
@@ -0,0 +1,390 @@
+# Copyright 2014 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.
+
+import os
+import mock
+import unittest
+
+from telemetry.internal import forwarders
+from telemetry.internal.platform import network_controller_backend
+from telemetry.util import wpr_modes
+
+
+DEFAULT_PORTS = forwarders.PortSet(http=1111, https=2222, dns=3333)
+FORWARDER_HOST_IP = '123.321.123.321'
+EXPECTED_WPR_CA_CERT_PATH = os.path.join('[tempdir]', 'testca.pem')
+
+
+class FakePlatformBackend(object):
+  def __init__(self):
+    self.forwarder_factory = FakeForwarderFactory()
+    self.supports_test_ca = True
+    self.is_test_ca_installed = False
+    self.faulty_cert_installer = False
+    self.wpr_port_pairs = None
+    # Normally test using all default ports.
+    self.SetWprPortPairs(http=(0, 0), https=(0, 0), dns=(0, 0))
+
+  def SetWprPortPairs(self, http, https, dns):
+    self.wpr_port_pairs = forwarders.PortPairs(
+        forwarders.PortPair(*http),
+        forwarders.PortPair(*https),
+        forwarders.PortPair(*dns) if dns is not None else None)
+
+  def GetWprPortPairs(self):
+    return self.wpr_port_pairs
+
+  def InstallTestCa(self, ca_cert_path):
+    del ca_cert_path  # Unused argument.
+    self.is_test_ca_installed = True
+    # Exception is raised after setting the "installed" value to confirm that
+    # cleaup code is being called in case of errors.
+    if self.faulty_cert_installer:
+      raise Exception('Cert install failed!')
+
+  def RemoveTestCa(self):
+    self.is_test_ca_installed = False
+
+
+class FakeForwarderFactory(object):
+  def __init__(self):
+    self.host_ip = FORWARDER_HOST_IP
+
+  def Create(self, port_pairs):
+    return forwarders.Forwarder(
+        forwarders.PortPairs(*[
+            forwarders.PortPair(p.local_port, p.remote_port or p.local_port)
+            if p else None for p in port_pairs]))
+
+
+class FakeReplayServer(object):
+  DEFAULT_PORTS = NotImplemented  # Will be assigned during test setUp.
+
+  def __init__(self, archive_path, host_ip, http_port, https_port, dns_port,
+               replay_args):
+    self.archive_path = archive_path
+    self.host_ip = host_ip
+    self.ports = forwarders.PortSet(
+        http_port or self.DEFAULT_PORTS.http,
+        https_port or self.DEFAULT_PORTS.https,
+        dns_port or self.DEFAULT_PORTS.dns if dns_port is not None else None)
+    self.replay_args = replay_args
+    self.is_running = False
+
+  def StartServer(self):
+    assert not self.is_running
+    self.is_running = True
+    return self.ports
+
+  def StopServer(self):
+    assert self.is_running
+    self.is_running = False
+
+
+class TestNetworkControllerBackend(
+    network_controller_backend.NetworkControllerBackend):
+  """Expose some private properties for testing purposes."""
+
+  @property
+  def wpr_ca_cert_path(self):
+    return self._wpr_ca_cert_path
+
+  @property
+  def replay_server(self):
+    return self._wpr_server
+
+  @property
+  def forwarder(self):
+    return self._forwarder
+
+  @property
+  def platform_backend(self):
+    return self._platform_backend
+
+
+# pylint: disable=no-member
+class NetworkControllerBackendTest(unittest.TestCase):
+  def Patch(self, *args, **kwargs):
+    """Patch an object for the duration of a test, and return its mock."""
+    patcher = mock.patch(*args, **kwargs)
+    mock_object = patcher.start()
+    self.addCleanup(patcher.stop)
+    return mock_object
+
+  def PatchImportedModule(self, name):
+    """Shorthand to patch a module imported by network_controller_backend."""
+    return self.Patch(
+        'telemetry.internal.platform.network_controller_backend.%s' % name)
+
+  def setUp(self):
+    # Always use our FakeReplayServer.
+    FakeReplayServer.DEFAULT_PORTS = DEFAULT_PORTS  # Use global defaults.
+    self.Patch(
+        'telemetry.internal.util.wpr_server.ReplayServer', FakeReplayServer)
+
+    # Pretend that only some predefined set of files exist.
+    def fake_path_exists(filename):
+      return filename in ['some-archive.wpr', 'another-archive.wpr']
+
+    self.Patch('os.path.exists', side_effect=fake_path_exists)
+
+    # Mock some imported modules.
+    mock_certutils = self.PatchImportedModule('certutils')
+    mock_certutils.openssl_import_error = None
+    mock_certutils.generate_dummy_ca_cert.return_value = ('-', '-')
+
+    mock_platformsettings = self.PatchImportedModule('platformsettings')
+    mock_platformsettings.HasSniSupport.return_value = True
+
+    mock_tempfile = self.PatchImportedModule('tempfile')
+    mock_tempfile.mkdtemp.return_value = '[tempdir]'
+
+    self.PatchImportedModule('shutil')
+
+    self.network_controller_backend = TestNetworkControllerBackend(
+        FakePlatformBackend())
+
+  def testOpenCloseController(self):
+    b = self.network_controller_backend
+    self.assertFalse(b.is_open)
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg']) # Also installs test CA.
+    self.assertTrue(b.is_open)
+    self.assertTrue(b.is_test_ca_installed)
+    self.assertTrue(b.platform_backend.is_test_ca_installed)
+    b.Close() # Also removes test CA.
+    self.assertFalse(b.is_open)
+    self.assertFalse(b.is_test_ca_installed)
+    self.assertFalse(b.platform_backend.is_test_ca_installed)
+    b.Close()  # It's fine to close a closed controller.
+    self.assertFalse(b.is_open)
+
+  def testOpeningOpenControllerRaises(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    with self.assertRaises(AssertionError):
+      b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+
+  def testInstallTestCaFailure(self):
+    b = self.network_controller_backend
+    b.platform_backend.faulty_cert_installer = True
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg']) # Try to install test CA.
+
+    # Test CA is not installed, but the controller is otherwise open and safe
+    # to use.
+    self.assertTrue(b.is_open)
+    self.assertFalse(b.is_test_ca_installed)
+    self.assertFalse(b.platform_backend.is_test_ca_installed)
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+
+    b.Close() # No test CA to remove.
+    self.assertFalse(b.is_open)
+    self.assertFalse(b.is_test_ca_installed)
+    self.assertFalse(b.platform_backend.is_test_ca_installed)
+
+  def testStartStopReplay(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    self.assertFalse(b.is_replay_active)
+
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertTrue(b.replay_server.is_running)
+    self.assertIsNotNone(b.forwarder.port_pairs)
+
+    old_replay_server = b.replay_server
+    old_forwarder = b.forwarder
+    b.StopReplay()
+    self.assertFalse(b.is_replay_active)
+    self.assertFalse(old_replay_server.is_running)
+    self.assertIsNone(old_forwarder.port_pairs)
+    self.assertTrue(b.is_open)  # Controller is still open.
+
+    b.Close()
+    self.assertFalse(b.is_open)
+
+  def testClosingControllerAlsoStopsReplay(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertTrue(b.replay_server.is_running)
+    self.assertIsNotNone(b.forwarder.port_pairs)
+
+    old_replay_server = b.replay_server
+    old_forwarder = b.forwarder
+    b.Close()
+    self.assertFalse(b.is_replay_active)
+    self.assertFalse(old_replay_server.is_running)
+    self.assertIsNone(old_forwarder.port_pairs)
+    self.assertFalse(b.is_open)
+
+  def testReplayOnClosedControllerRaises(self):
+    b = self.network_controller_backend
+    self.assertFalse(b.is_open)
+    with self.assertRaises(AssertionError):
+      b.StartReplay('some-archive.wpr')
+
+  def testReplayWithSameArgsReuseServer(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertTrue(b.replay_server.is_running)
+
+    old_replay_server = b.replay_server
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertIs(b.replay_server, old_replay_server)
+    self.assertTrue(b.replay_server.is_running)
+
+  def testReplayWithDifferentArgsUseDifferentServer(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertTrue(b.replay_server.is_running)
+
+    old_replay_server = b.replay_server
+    b.StartReplay('another-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertIsNot(b.replay_server, old_replay_server)
+    self.assertTrue(b.replay_server.is_running)
+    self.assertFalse(old_replay_server.is_running)
+
+  def testReplayWithoutArchivePathDoesNotStopReplay(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+
+    b.StartReplay('some-archive.wpr')
+    self.assertTrue(b.is_replay_active)
+    self.assertTrue(b.replay_server.is_running)
+    old_replay_server = b.replay_server
+
+    b.StartReplay(None)
+    self.assertTrue(b.is_replay_active)
+    self.assertIs(b.replay_server, old_replay_server)
+    self.assertTrue(b.replay_server.is_running)
+    self.assertEqual(b.replay_server.archive_path, 'some-archive.wpr')
+
+  def testModeOffDoesNotCreateReplayServer(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_OFF, ['--some-arg'])
+    b.StartReplay('may-or-may-not-exist.wpr')
+    self.assertFalse(b.is_replay_active)
+    self.assertIsNone(b.replay_server)
+    self.assertIsNone(b.forwarder)
+
+  def testBadArchivePathRaises(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    with self.assertRaises(network_controller_backend.ArchiveDoesNotExistError):
+      b.StartReplay('does-not-exist.wpr')
+
+  def testBadArchivePathOnRecordIsOkay(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_RECORD, ['--some-arg'])
+    b.StartReplay('does-not-exist-yet.wpr')  # Does not raise.
+    self.assertTrue(b.is_replay_active)
+
+  def testReplayServerSettings(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_RECORD, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+
+    # Externally visible properties
+    self.assertTrue(b.is_replay_active)
+    self.assertEqual(b.host_ip, FORWARDER_HOST_IP)
+    self.assertEqual(b.wpr_device_ports, DEFAULT_PORTS)
+
+    # Private replay server settings.
+    self.assertTrue(b.replay_server.is_running)
+    self.assertEqual(b.replay_server.archive_path, 'some-archive.wpr')
+    self.assertEqual(b.replay_server.host_ip, FORWARDER_HOST_IP)
+    self.assertEqual(b.replay_server.replay_args, [
+        '--some-arg', '--record', '--inject_scripts=',
+        '--should_generate_certs',
+        '--https_root_ca_cert_path=%s' % EXPECTED_WPR_CA_CERT_PATH])
+
+  def testReplayServerOffSettings(self):
+    b = self.network_controller_backend
+    b.platform_backend.wpr_ca_cert_path = 'CERT_FILE'
+    b.Open(wpr_modes.WPR_OFF, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+
+    self.assertFalse(b.is_replay_active)
+    self.assertEqual(b.host_ip, FORWARDER_HOST_IP)
+    self.assertEqual(b.wpr_device_ports, None)
+    self.assertIsNone(b.replay_server)
+
+  def testUseDefaultPorts(self):
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+    self.assertEqual(b.replay_server.ports, DEFAULT_PORTS)
+    self.assertEqual(b.wpr_device_ports, DEFAULT_PORTS)
+
+    # Invariant
+    self.assertEqual(b.forwarder.port_pairs.local_ports, b.replay_server.ports)
+    self.assertEqual(b.forwarder.port_pairs.remote_ports, b.wpr_device_ports)
+
+  def testUseDefaultLocalPorts(self):
+    b = self.network_controller_backend
+    b.platform_backend.SetWprPortPairs(
+        http=(0, 8888), https=(0, 4444), dns=(0, 2222))
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+    self.assertEqual(b.replay_server.ports, DEFAULT_PORTS)
+    self.assertEqual(b.wpr_device_ports, forwarders.PortSet(8888, 4444, 2222))
+
+    # Invariant
+    self.assertEqual(b.forwarder.port_pairs.local_ports, b.replay_server.ports)
+    self.assertEqual(b.forwarder.port_pairs.remote_ports, b.wpr_device_ports)
+
+  def testUseSpecificPorts(self):
+    b = self.network_controller_backend
+    b.platform_backend.SetWprPortPairs(
+        http=(88, 8888), https=(44, 4444), dns=None)
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+    self.assertEqual(b.replay_server.ports, forwarders.PortSet(88, 44, None))
+    self.assertEqual(b.wpr_device_ports, forwarders.PortSet(8888, 4444, None))
+
+    # Invariant
+    self.assertEqual(b.forwarder.port_pairs.local_ports, b.replay_server.ports)
+    self.assertEqual(b.forwarder.port_pairs.remote_ports, b.wpr_device_ports)
+
+  def testRestartReplayShouldReusePorts(self):
+    FakeReplayServer.DEFAULT_PORTS = forwarders.PortSet(123, 456, 789)
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+    self.assertEqual(b.wpr_device_ports, forwarders.PortSet(123, 456, 789))
+
+    # If replay restarts, the factory may use a different set of default ports.
+    FakeReplayServer.DEFAULT_PORTS = forwarders.PortSet(987, 654, 321)
+    b.StartReplay('another-archive.wpr')
+
+    # However same ports must be used, because apps/browsers may already be
+    # configured to use the old set of ports.
+    self.assertEqual(b.wpr_device_ports, forwarders.PortSet(123, 456, 789))
+
+  def testNewControllerSessionMayUseDifferentPorts(self):
+    FakeReplayServer.DEFAULT_PORTS = forwarders.PortSet(123, 456, 789)
+    b = self.network_controller_backend
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+    self.assertEqual(b.wpr_device_ports, forwarders.PortSet(123, 456, 789))
+    b.Close()
+
+    # If replay restarts, the factory may use a different set of default ports.
+    FakeReplayServer.DEFAULT_PORTS = forwarders.PortSet(987, 654, 321)
+    b.Open(wpr_modes.WPR_REPLAY, ['--some-arg'])
+    b.StartReplay('some-archive.wpr')
+
+    # This time the network controller session was closed between replay's,
+    # so it's fine to use a different set of ports.
+    self.assertEqual(b.wpr_device_ports, forwarders.PortSet(987, 654, 321))