-
Notifications
You must be signed in to change notification settings - Fork 22
Expand file tree
/
Copy pathlocal.py
More file actions
128 lines (105 loc) · 3.76 KB
/
local.py
File metadata and controls
128 lines (105 loc) · 3.76 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
import subprocess, os, time, json,logging
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:
self.binary_path = self.options['binarypath']
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']
self.proc = subprocess.Popen(self._generate_cmd(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = self.proc.communicate()
os.system('echo "" > "'+ self.local_logfile_path +'"')
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()