# 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))