• 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  def __init__(self, device, tool):
39    self.device = device
40    self.tool = tool
41
42  def Map(self, port_pairs):
43    forwarder.Forwarder.Map(port_pairs, self.device, self.tool)
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, tool):
64    super().__init__()
65    self._device = device
66    self._spawning_server = chrome_test_server_spawner.SpawningServer(
67        port, PortForwarderAndroid(device, tool), MAX_TEST_SERVER_INSTANCES)
68    self._tool = tool
69
70  @property
71  def server_address(self):
72    return self._spawning_server.server.server_address
73
74  @property
75  def port(self):
76    return self.server_address[1]
77
78  #override
79  def SetUp(self):
80    # See net/test/spawned_test_server/remote_test_server.h for description of
81    # the fields in the config file.
82    test_server_config = json.dumps({
83      'spawner_url_base': 'http://localhost:%d' % self.port
84    })
85    self._device.WriteFile(
86        '%s/net-test-server-config' % self._device.GetExternalStoragePath(),
87        test_server_config)
88    forwarder.Forwarder.Map(
89        [(self.port, self.port)], self._device, self._tool)
90    self._spawning_server.Start()
91
92  #override
93  def Reset(self):
94    self._spawning_server.CleanupState()
95
96  #override
97  def TearDown(self):
98    self.Reset()
99    self._spawning_server.Stop()
100    forwarder.Forwarder.UnmapDevicePort(self.port, self._device)
101