-
Notifications
You must be signed in to change notification settings - Fork 22
Expand file tree
/
Copy pathlocal.py
More file actions
145 lines (121 loc) · 4.56 KB
/
local.py
File metadata and controls
145 lines (121 loc) · 4.56 KB
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
import subprocess, os, time, json, logging, re
import psutil
from browserstack.local_binary import LocalBinary
from browserstack.bserrors import BrowserStackLocalError
logger = logging.getLogger(__name__)
try:
from importlib.metadata import version as package_version, PackageNotFoundError
except:
import pkg_resources
class Local:
def __init__(self, key=None, binary_path=None, **kwargs):
self.key = os.environ['BROWSERSTACK_ACCESS_KEY'] if 'BROWSERSTACK_ACCESS_KEY' in os.environ else key
self.options = kwargs
self.local_logfile_path = os.path.join(os.getcwd(), 'local.log')
LocalBinary.set_version(self.get_package_version())
def __xstr(self, key, value):
if key is None:
return ['']
if str(value).lower() == "true":
return ['-' + key]
elif str(value).lower() == "false":
return ['']
else:
return ['-' + key, str(value)]
def get_package_version(self):
name = "browserstack-local"
version = 'None'
use_fallback = False
try:
temp = package_version
except NameError: # Only catch if package_version is not defined(and not other errors)
use_fallback = True
if use_fallback:
try:
version = pkg_resources.get_distribution(name).version
except pkg_resources.DistributionNotFound:
version = 'None'
else:
try:
version = package_version(name)
except PackageNotFoundError:
version = 'None'
return version
def _generate_cmd(self):
cmd = [self.binary_path, '-d', 'start', '-logFile', self.local_logfile_path, "-k", self.key, '--source', 'python:' + self.get_package_version()]
for o in self.options.keys():
if self.options.get(o) is not None:
cmd = cmd + self.__xstr(o, self.options.get(o))
return cmd
def _generate_stop_cmd(self):
cmd = self._generate_cmd()
cmd[2] = 'stop'
return cmd
def start(self, **kwargs):
for k, v in kwargs.items():
self.options[k] = v
if 'key' in self.options:
self.key = self.options['key']
del self.options['key']
if 'binarypath' in self.options:
candidate = os.path.realpath(self.options['binarypath'])
if not os.path.isfile(candidate):
raise BrowserStackLocalError('binarypath does not point to a file')
try:
version_output = subprocess.check_output([candidate, '--version'], timeout=10).decode('utf-8')
except (subprocess.SubprocessError, OSError) as e:
raise BrowserStackLocalError('binarypath failed verification: {}'.format(e))
if not re.match(LocalBinary.VERSION_REGEX, version_output):
raise BrowserStackLocalError('binarypath failed verification')
self.binary_path = candidate
del self.options['binarypath']
else:
l = LocalBinary(self.key)
try:
self.binary_path = l.get_binary()
except Exception as e:
l = LocalBinary(self.key, e)
self.binary_path = l.get_binary()
if 'logfile' in self.options:
self.local_logfile_path = self.options['logfile']
del self.options['logfile']
if "onlyCommand" in kwargs and kwargs["onlyCommand"]:
return
if 'source' in self.options:
del self.options['source']
logfile_dir = os.path.dirname(self.local_logfile_path)
if logfile_dir:
os.makedirs(logfile_dir, exist_ok=True)
try:
with open(self.local_logfile_path, 'w') as f:
f.write('')
except OSError as e:
raise BrowserStackLocalError('Unable to open logfile: {}'.format(e))
self.proc = subprocess.Popen(self._generate_cmd(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = self.proc.communicate()
try:
if out:
output_string = out.decode()
else:
output_string = err.decode()
data = json.loads(output_string)
if data['state'] != "connected":
raise BrowserStackLocalError(data["message"]["message"])
else:
self.pid = data['pid']
except ValueError:
logger.error("BinaryOutputParseError: Raw String = '{}'".format(output_string) )
raise BrowserStackLocalError('Error parsing JSON output from daemon. Raw String = "{}"'.format(output_string))
def isRunning(self):
return hasattr(self, 'pid') and psutil.pid_exists(self.pid)
def stop(self):
try:
proc = subprocess.Popen(self._generate_stop_cmd(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = proc.communicate()
except Exception as e:
return
def __enter__(self):
self.start(**self.options)
return self
def __exit__(self, *args):
self.stop()