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