• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2014 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5
6import json
7import time
8
9from devil.android import forwarder
10from devil.android import ports
11from pylib.base import test_server
12from pylib.constants import host_paths
13
14with host_paths.SysPath(host_paths.BUILD_COMMON_PATH):
15  import chrome_test_server_spawner
16
17
18# The tests should not need more than one test server instance.
19MAX_TEST_SERVER_INSTANCES = 1
20
21
22def _WaitUntil(predicate, max_attempts=5):
23  """Blocks until the provided predicate (function) is true.
24
25  Returns:
26    Whether the provided predicate was satisfied once (before the timeout).
27  """
28  sleep_time_sec = 0.025
29  for _ in range(1, max_attempts):
30    if predicate():
31      return True
32    time.sleep(sleep_time_sec)
33    sleep_time_sec = min(1, sleep_time_sec * 2)  # Don't wait more than 1 sec.
34  return False
35
36
37class PortForwarderAndroid(chrome_test_server_spawner.PortForwarder):
38
39  def __init__(self, device):
40    self.device = device
41
42  def Map(self, port_pairs):
43    forwarder.Forwarder.Map(port_pairs, self.device)
44
45  def GetDevicePortForHostPort(self, host_port):
46    return forwarder.Forwarder.DevicePortForHostPort(host_port)
47
48  def WaitHostPortAvailable(self, port):
49    return _WaitUntil(lambda: ports.IsHostPortAvailable(port))
50
51  def WaitPortNotAvailable(self, port):
52    return _WaitUntil(lambda: not ports.IsHostPortAvailable(port))
53
54  def WaitDevicePortReady(self, port):
55    return _WaitUntil(lambda: ports.IsDevicePortUsed(self.device, port))
56
57  def Unmap(self, device_port):
58    forwarder.Forwarder.UnmapDevicePort(device_port, self.device)
59
60
61class LocalTestServerSpawner(test_server.TestServer):
62
63  def __init__(self, port, device):
64    super().__init__()
65    self._device = device
66    self._spawning_server = chrome_test_server_spawner.SpawningServer(
67        port, PortForwarderAndroid(device), MAX_TEST_SERVER_INSTANCES)
68
69  @property
70  def server_address(self):
71    return self._spawning_server.server.server_address
72
73  @property
74  def port(self):
75    return self.server_address[1]
76
77  #override
78  def SetUp(self):
79    # See net/test/spawned_test_server/remote_test_server.h for description of
80    # the fields in the config file.
81    test_server_config = json.dumps({
82      'spawner_url_base': 'http://localhost:%d' % self.port
83    })
84    self._device.WriteFile(
85        '%s/net-test-server-config' % self._device.GetExternalStoragePath(),
86        test_server_config)
87    forwarder.Forwarder.Map([(self.port, self.port)], self._device)
88    self._spawning_server.Start()
89
90  #override
91  def Reset(self):
92    self._spawning_server.CleanupState()
93
94  #override
95  def TearDown(self):
96    self.Reset()
97    self._spawning_server.Stop()
98    forwarder.Forwarder.UnmapDevicePort(self.port, self._device)
99