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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
#!/usr/bin/python
# Copyright (c) 2011 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.
"""Launches and kills ChromeDriver.
For ChromeDriver documentation, refer to:
http://dev.chromium.org/developers/testing/webdriver-for-chrome
"""
import logging
import os
import platform
import signal
import subprocess
import sys
import threading
import urllib2
class ChromeDriverLauncher:
"""Launches and kills the ChromeDriver process."""
def __init__(self, exe_path=None, root_path=None, port=None, url_base=None):
"""Initializes a new launcher.
Args:
exe_path: path to the ChromeDriver executable
root_path: base path from which ChromeDriver webserver will serve files
port: port that ChromeDriver will listen on
url_base: base URL which ChromeDriver webserver will listen from
"""
self._exe_path = exe_path
self._root_path = root_path
self._port = port
self._url_base = url_base
if self._exe_path is None:
self._exe_path = ChromeDriverLauncher.LocateExe()
if self._exe_path is None:
raise RuntimeError('ChromeDriver exe could not be found in its default '
'location. Searched in following directories: ' +
', '.join(self.DefaultExeLocations()))
if self._root_path is not None:
self._root_path = os.path.abspath(self._root_path)
self._process = None
if not os.path.exists(self._exe_path):
raise RuntimeError('ChromeDriver exe not found at: ' + self._exe_path)
os.environ['PATH'] = os.path.dirname(self._exe_path) + os.environ['PATH']
self.Start()
@staticmethod
def DefaultExeLocations():
"""Returns the paths that are used to find the ChromeDriver executable.
Returns:
a list of directories that would be searched for the executable
"""
script_dir = os.path.dirname(__file__)
chrome_src = os.path.abspath(os.path.join(
script_dir, os.pardir, os.pardir, os.pardir))
bin_dirs = {
'linux2': [ os.path.join(chrome_src, 'out', 'Debug'),
os.path.join(chrome_src, 'sconsbuild', 'Debug'),
os.path.join(chrome_src, 'out', 'Release'),
os.path.join(chrome_src, 'sconsbuild', 'Release')],
'linux3': [ os.path.join(chrome_src, 'out', 'Debug'),
os.path.join(chrome_src, 'sconsbuild', 'Debug'),
os.path.join(chrome_src, 'out', 'Release'),
os.path.join(chrome_src, 'sconsbuild', 'Release')],
'darwin': [ os.path.join(chrome_src, 'xcodebuild', 'Debug'),
os.path.join(chrome_src, 'xcodebuild', 'Release')],
'win32': [ os.path.join(chrome_src, 'chrome', 'Debug'),
os.path.join(chrome_src, 'build', 'Debug'),
os.path.join(chrome_src, 'chrome', 'Release'),
os.path.join(chrome_src, 'build', 'Release')],
}
return [os.getcwd()] + bin_dirs.get(sys.platform, [])
@staticmethod
def LocateExe():
"""Attempts to locate the ChromeDriver executable.
This searches the current directory, then checks the appropriate build
locations according to platform.
Returns:
absolute path to the ChromeDriver executable, or None if not found
"""
exe_name = 'chromedriver'
if platform.system() == 'Windows':
exe_name += '.exe'
for dir in ChromeDriverLauncher.DefaultExeLocations():
path = os.path.join(dir, exe_name)
if os.path.exists(path):
return os.path.abspath(path)
return None
def Start(self):
"""Starts a new ChromeDriver process.
Kills a previous one if it is still running.
Raises:
RuntimeError if ChromeDriver does not start
"""
def _WaitForLaunchResult(stdout, started_event, launch_result):
"""Reads from the stdout of ChromeDriver and parses the launch result.
Args:
stdout: handle to ChromeDriver's standard output
started_event: condition variable to notify when the launch result
has been parsed
launch_result: dictionary to add the result of this launch to
"""
status_line = stdout.readline()
started_event.acquire()
try:
launch_result['success'] = status_line.startswith('Started')
launch_result['status_line'] = status_line
if launch_result['success']:
port_line = stdout.readline()
launch_result['port'] = int(port_line.split('=')[1])
started_event.notify()
finally:
started_event.release()
if self._process is not None:
self.Kill()
chromedriver_args = [self._exe_path]
if self._root_path is not None:
chromedriver_args += ['--root=%s' % self._root_path]
if self._port is not None:
chromedriver_args += ['--port=%d' % self._port]
if self._url_base is not None:
chromedriver_args += ['--url-base=%s' % self._url_base]
proc = subprocess.Popen(chromedriver_args,
stdout=subprocess.PIPE)
if proc is None:
raise RuntimeError('ChromeDriver cannot be started')
self._process = proc
# Wait for ChromeDriver to be initialized before returning.
launch_result = {}
started_event = threading.Condition()
started_event.acquire()
spawn_thread = threading.Thread(
target=_WaitForLaunchResult,
args=(proc.stdout, started_event, launch_result))
spawn_thread.start()
started_event.wait(20)
timed_out = 'success' not in launch_result
started_event.release()
if timed_out:
raise RuntimeError('ChromeDriver did not respond')
elif not launch_result['success']:
raise RuntimeError('ChromeDriver failed to launch: ' +
launch_result['status_line'])
self._port = launch_result['port']
logging.info('ChromeDriver running on port %s' % self._port)
def Kill(self):
"""Kills a currently running ChromeDriver process, if it is running."""
def _WaitForShutdown(process, shutdown_event):
"""Waits for the process to quit and then notifies."""
process.wait()
shutdown_event.acquire()
shutdown_event.notify()
shutdown_event.release()
if self._process is None:
return
try:
urllib2.urlopen(self.GetURL() + '/shutdown').close()
except urllib2.URLError:
# Could not shutdown. Kill.
pid = self._process.pid
if platform.system() == 'Windows':
subprocess.call(['taskkill.exe', '/T', '/F', '/PID', str(pid)])
else:
os.kill(pid, signal.SIGTERM)
# Wait for ChromeDriver process to exit before returning.
# Even if we had to kill the process above, we still should call wait
# to cleanup the zombie.
shutdown_event = threading.Condition()
shutdown_event.acquire()
wait_thread = threading.Thread(
target=_WaitForShutdown,
args=(self._process, shutdown_event))
wait_thread.start()
shutdown_event.wait(10)
shutdown_event.release()
self._process = None
def GetURL(self):
url = 'http://localhost:' + str(self._port)
if self._url_base:
url += self._url_base
return url
def GetPort(self):
return self._port
|
