Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
when attach and strict both true, SessionBrokenError will raise if app not running
Raises:
requests.HTTPError, SessionBrokenError
"""
if pkg_name is None:
return self._default_session
if not attach:
request_data = {"flags": "-S"}
if launch_timeout:
request_data["timeout"] = str(launch_timeout)
resp = self._reqsess.post(self.path2url("/session/" + pkg_name),
data=request_data)
if resp.status_code == 410: # Gone
raise SessionBrokenError(pkg_name, resp.text)
resp.raise_for_status()
jsondata = resp.json()
if not jsondata["success"]:
raise SessionBrokenError("app launch failed",
jsondata["error"], jsondata["output"])
time.sleep(2.5) # wait launch finished, maybe no need
pid = self._pidof_app(pkg_name)
if not pid:
if strict:
raise SessionBrokenError(pkg_name)
return self.session(pkg_name,
attach=False,
launch_timeout=launch_timeout)
return Session(self, pkg_name, pid)
"""
if pkg_name is None:
return self._default_session
if not attach:
request_data = {"flags": "-S"}
if launch_timeout:
request_data["timeout"] = str(launch_timeout)
resp = self._reqsess.post(self.path2url("/session/" + pkg_name),
data=request_data)
if resp.status_code == 410: # Gone
raise SessionBrokenError(pkg_name, resp.text)
resp.raise_for_status()
jsondata = resp.json()
if not jsondata["success"]:
raise SessionBrokenError("app launch failed",
jsondata["error"], jsondata["output"])
time.sleep(2.5) # wait launch finished, maybe no need
pid = self._pidof_app(pkg_name)
if not pid:
if strict:
raise SessionBrokenError(pkg_name)
return self.session(pkg_name,
attach=False,
launch_timeout=launch_timeout)
return Session(self, pkg_name, pid)
}
data = json.dumps(data).encode('utf-8')
res = self._reqsess.post(
jsonrpc_url, # +"?m="+method, #?method is for debug
headers={"Content-Type": "application/json"},
timeout=http_timeout,
data=data)
if DEBUG:
print("Shell$ curl -X POST -d '{}' {}".format(data, jsonrpc_url))
print("Output> " + res.text)
if res.status_code == 502:
raise GatewayError(
res, "gateway error, time used %.1fs" %
(time.time() - request_start))
if res.status_code == 410: # http status gone: session broken
raise SessionBrokenError("app quit or crash", jsonrpc_url,
res.text)
if res.status_code != 200:
raise UiaError(jsonrpc_url, data, res.status_code, res.text,
"HTTP Return code is not 200", res.text)
jsondata = res.json()
error = jsondata.get('error')
if not error:
return jsondata.get('result')
# error happends
err = JsonRpcError(error, method)
def is_exception(err, exception_name):
return err.exception_name == exception_name or exception_name in err.message
request_data["timeout"] = str(launch_timeout)
resp = self._reqsess.post(self.path2url("/session/" + pkg_name),
data=request_data)
if resp.status_code == 410: # Gone
raise SessionBrokenError(pkg_name, resp.text)
resp.raise_for_status()
jsondata = resp.json()
if not jsondata["success"]:
raise SessionBrokenError("app launch failed",
jsondata["error"], jsondata["output"])
time.sleep(2.5) # wait launch finished, maybe no need
pid = self._pidof_app(pkg_name)
if not pid:
if strict:
raise SessionBrokenError(pkg_name)
return self.session(pkg_name,
attach=False,
launch_timeout=launch_timeout)
return Session(self, pkg_name, pid)
def inner(self, *args, **kwargs):
if not self.running():
raise SessionBrokenError(self._pkg_name)
return fn(self, *args, **kwargs)