src/tools/telemetry/telemetry/core/platform/network_controller_backend_unittest.py

   1 # Copyright 2014 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 import tempfile
   6 import unittest
   7
   8 from telemetry.core import wpr_modes
   9 from telemetry.core.platform import network_controller_backend
  10
  11
  12 class FakePlatformBackend(object):
  13   pass
  14
  15
  16 class FakeBrowserBackend(object):
  17   pass
  18
  19
  20 class FakeReplayServer(object):
  21   def __init__(self, browser_backend, **replay_args):
  22     self.browser_backend = browser_backend
  23     self.replay_args = replay_args
  24     self.is_closed = False
  25
  26   def Close(self):
  27     assert not self.is_closed
  28     self.is_closed = True
  29
  30
  31 class TestNetworkControllerBackend(
  32     network_controller_backend.NetworkControllerBackend):
  33   """NetworkControllerBackend with a fake ReplayServer."""
  34
  35   def __init__(self, platform_backend):
  36     super(TestNetworkControllerBackend, self).__init__(platform_backend)
  37     self.fake_replay_server = None
  38
  39   def _ReplayServer(self, browser_backend, replay_args):
  40     self.fake_replay_server = FakeReplayServer(browser_backend, **replay_args)
  41     return self.fake_replay_server
  42
  43
  44 class NetworkControllerBackendTest(unittest.TestCase):
  45
  46   def testSameArgsReusesServer(self):
  47     b = TestNetworkControllerBackend(FakePlatformBackend())
  48     with tempfile.NamedTemporaryFile() as temp_file:
  49       archive_file = temp_file.name
  50       # Create Replay server.
  51       b.SetReplayArgs(archive_file, wpr_modes.WPR_REPLAY, '3g', ['--some-arg'])
  52       browser_backend = FakeBrowserBackend()
  53       b.UpdateReplay(browser_backend)
  54       self.assertIs(browser_backend, b.fake_replay_server.browser_backend)
  55       expected_replay_args = dict(
  56           archive_path=archive_file,
  57           wpr_mode=wpr_modes.WPR_REPLAY,
  58           netsim='3g',
  59           extra_wpr_args=['--some-arg'],
  60           make_javascript_deterministic=False)
  61       self.assertEqual(expected_replay_args, b.fake_replay_server.replay_args)
  62
  63       # Reuse Replay server.
  64       fake_replay_server = b.fake_replay_server
  65       b.SetReplayArgs(archive_file, wpr_modes.WPR_REPLAY, '3g', ['--some-arg'])
  66       b.UpdateReplay(browser_backend)
  67
  68     self.assertIs(fake_replay_server, b.fake_replay_server)
  69     b.StopReplay()
  70     self.assertTrue(b.fake_replay_server.is_closed)
  71
  72   def testDifferentArgsUsesDifferentServer(self):
  73     b = TestNetworkControllerBackend(FakePlatformBackend())
  74     with tempfile.NamedTemporaryFile() as temp_file:
  75       archive_file = temp_file.name
  76       # Create Replay server.
  77       b.SetReplayArgs(archive_file, wpr_modes.WPR_REPLAY, '3g', ['--some-arg'])
  78       browser_backend = FakeBrowserBackend()
  79       b.UpdateReplay(browser_backend)
  80       self.assertIs(browser_backend, b.fake_replay_server.browser_backend)
  81       expected_replay_args = dict(
  82           archive_path=archive_file,
  83           wpr_mode=wpr_modes.WPR_REPLAY,
  84           netsim='3g',
  85           extra_wpr_args=['--some-arg'],
  86           make_javascript_deterministic=False)
  87       self.assertEqual(expected_replay_args, b.fake_replay_server.replay_args)
  88       fake_replay_server = b.fake_replay_server
  89
  90       # Create different Replay server.
  91       # Set netsim to None instead of '3g'.
  92       b.SetReplayArgs(archive_file, wpr_modes.WPR_REPLAY, None, ['--some-arg'])
  93       b.UpdateReplay(browser_backend)
  94
  95       self.assertIsNot(fake_replay_server, b.fake_replay_server)
  96       self.assertTrue(fake_replay_server.is_closed)
  97       self.assertFalse(b.fake_replay_server.is_closed)
  98       different_replay_args = dict(
  99           archive_path=archive_file,
 100           wpr_mode=wpr_modes.WPR_REPLAY,
 101           netsim=None,  # first call used '3g'
 102           extra_wpr_args=['--some-arg'],
 103           make_javascript_deterministic=False)
 104
 105     self.assertEqual(different_replay_args, b.fake_replay_server.replay_args)
 106     b.StopReplay()
 107     self.assertTrue(b.fake_replay_server.is_closed)
 108
 109   def testUpdateReplayWithoutArgsIsOkay(self):
 110     b = TestNetworkControllerBackend(FakePlatformBackend())
 111     b.UpdateReplay(FakeBrowserBackend())  # does not raise
 112
 113   def testBadArchivePathRaises(self):
 114     b = TestNetworkControllerBackend(FakePlatformBackend())
 115     b.SetReplayArgs('/tmp/nonexistant', wpr_modes.WPR_REPLAY, '3g', [])
 116     with self.assertRaises(network_controller_backend.ArchiveDoesNotExistError):
 117       b.UpdateReplay(FakeBrowserBackend())
 118
 119   def testBadArchivePathOnRecordIsOkay(self):
 120     """No ArchiveDoesNotExistError for record mode."""
 121     b = TestNetworkControllerBackend(FakePlatformBackend())
 122     b.SetReplayArgs('/tmp/nonexistant', wpr_modes.WPR_RECORD, '3g', [])
 123     b.UpdateReplay(FakeBrowserBackend())  # does not raise
 124
 125   def testModeOffDoesNotCreateReplayServer(self):
 126     b = TestNetworkControllerBackend(FakePlatformBackend())
 127     b.SetReplayArgs('/tmp/nonexistant', wpr_modes.WPR_OFF, '3g', [])
 128     b.UpdateReplay(FakeBrowserBackend())
 129     self.assertIsNone(b.fake_replay_server)