File: server.py

package info (click to toggle)
chromium 138.0.7204.183-1
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 6,071,908 kB
  • sloc: cpp: 34,937,088; ansic: 7,176,967; javascript: 4,110,704; python: 1,419,953; asm: 946,768; xml: 739,971; pascal: 187,324; sh: 89,623; perl: 88,663; objc: 79,944; sql: 50,304; cs: 41,786; fortran: 24,137; makefile: 21,806; php: 13,980; tcl: 13,166; yacc: 8,925; ruby: 7,485; awk: 3,720; lisp: 3,096; lex: 1,327; ada: 727; jsp: 228; sed: 36
file content (170 lines) | stat: -rw-r--r-- 6,088 bytes parent folder | download | duplicates (6)
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
# Copyright 2013 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import atexit
import os
import socket
import subprocess
import psutil
import threading
import time
import urllib

class Server(object):
  """A running ChromeDriver server."""

  def __init__(self, exe_path, log_path=None, verbose=True,
               replayable=False, devtools_replay_path=None,
               bidi_mapper_path=None, remote_chromedriver_port=None,
               additional_args=None):
    """Starts the ChromeDriver server and waits for it to be ready.

    Args:
      exe_path: path to the ChromeDriver executable
      log_path: path to the log file
      verbose: make the logged data verbose
      replayable: don't truncate strings in log to make the session replayable
      devtools_replay_path: replay devtools events from the log at this path
      additional_args: list of additional arguments to pass to ChromeDriver
    Raises:
      RuntimeError: if ChromeDriver fails to start
    """
    if not os.path.exists(exe_path):
      raise RuntimeError('ChromeDriver exe not found at: ' + exe_path)

    port = self._FindOpenPort()
    chromedriver_args = [exe_path, '--port=%d' % port]
    if log_path:
      chromedriver_args.extend(['--log-path=%s' % log_path])
      chromedriver_args.extend(['--append-log'])
      chromedriver_args.extend(['--readable-timestamp'])
      if verbose:
        chromedriver_args.extend(['--verbose',
                                  '--vmodule=*/chrome/test/chromedriver/*=3'])
      if replayable:
        chromedriver_args.extend(['--replayable'])

    if devtools_replay_path:
      chromedriver_args.extend(['--devtools-replay=%s' % devtools_replay_path])

    if bidi_mapper_path:
      chromedriver_args.extend(['--bidi-mapper-path=%s' % bidi_mapper_path])

    if additional_args:
      for arg in additional_args:
        if not arg.startswith('--'):
          arg = '--' + arg
        chromedriver_args.extend([arg])

    self._process = subprocess.Popen(chromedriver_args)
    self._pid = self._process.pid
    self._host = '127.0.0.1'
    self._port = port
    if remote_chromedriver_port is not None:
      self._port = remote_chromedriver_port
    self._url = 'http://%s:%d' % (self._host, self._port)
    if self._process is None:
      raise RuntimeError('ChromeDriver server cannot be started')

    max_time = time.time() + 40
    while not self.IsRunning():
      if time.time() > max_time:
        self._process.poll()
        if self._process.returncode is None:
          print('ChromeDriver process still running, but not responding')
        else:
          print('ChromeDriver process exited with return code %d'
                % self._process.returncode)
        self._process.terminate()
        raise RuntimeError('ChromeDriver server did not start')
      time.sleep(0.1)

    atexit.register(self.Kill)

  def _FindOpenPort(self):
    for port in range(9500, 10000):
      try:
        socket.create_connection(('127.0.0.1', port), 0.2).close()
      except socket.error:
        return port
    raise RuntimeError('Cannot find open port to launch ChromeDriver')

  def GetUrl(self):
    return self._url

  def GetPid(self):
    return self._pid

  def GetHost(self):
    return self._host

  def GetPort(self):
    return self._port

  def IsRunning(self):
    """Returns whether the server is up and running."""
    try:
      urllib.request.urlopen(self.GetUrl() + '/status')
      return True
    except urllib.error.URLError:
      return False

  def Kill(self):
    """Kills the ChromeDriver server, if it is running."""
    if self._process is None:
      return

    chromedriver_proc = psutil.Process(self._pid)
    # Child processes must be queried before the call to shutdown.
    # If queried later their parent ChromeDriver process might already be gone.
    # In this case the 'children' function will discover nothing or it might
    # also discover children of a different process that obtained the same PID
    # from the OS.
    processes = chromedriver_proc.children(recursive=True)

    try:
      urllib.request.urlopen(self.GetUrl() + '/shutdown', timeout=10).close()
    except:
      self._process.terminate()

    # By this point of execution the ChromeDriver process might already be gone.
    # The system might also assign the same PID to a different process.
    # Still we can safely terminate or kill it because psutil.Process.terminate
    # and psutil.Process.kill check preemptively if the PID has been reused.
    # S/A: https://psutil.readthedocs.io/en/latest/#psutil.Process.terminate
    # S/A: https://psutil.readthedocs.io/en/latest/#psutil.Process.kill
    # As for psutil.wait_procs - there is no such guarantee in the
    # documentation. Imagine that the process has gone and its PID has been
    # reused. If psutil.Process.wait does not check the PID for reuse and some
    # process obtained the same PID then psuti.wait_procs will always return
    # this process as alive. The code below will try to terminate / kill it,
    # this time with a check that will throw psutil.NoSuchProcess exception.
    # This exception is anticipated and the throwing process will be excluded
    # from the following attempts of cleaning the resources.
    alive = [chromedriver_proc] + processes
    _, alive = psutil.wait_procs(alive, timeout=5)
    if len(alive):
      print('Terminating %d processes' % len(alive))
      non_existing = []
      for proc in alive:
        try:
          proc.terminate()
        except psutil.NoSuchProcess:
          # The process might be gone by this point
          non_existing.append(proc)
          pass

      for proc in non_existing:
        alive.remove(proc)

    _, alive = psutil.wait_procs(alive, timeout=5)
    if len(alive):
      print('Killing %d processes' % len(alive))
      for proc in alive:
        try:
          proc.kill()
        except psutil.NoSuchProcess:
          # The process might be gone by this point
          pass
    self._process = None