Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
source = inspect.getsource(source)
elif isinstance(source, types.FunctionType):
call_name = source.__name__
file_name = inspect.getsourcefile(source)
source = _source_of_function(source)
else:
source = textwrap.dedent(str(source))
if not call_name and kwargs:
raise TypeError("can't pass kwargs to non-function remote_exec")
channel = self.newchannel()
self._send(
Message.CHANNEL_EXEC,
channel.id,
gateway_base.dumps_internal((source, file_name, call_name, kwargs)),
)
return channel
def test_no_tracing_by_default(self):
assert gateway_base.trace == gateway_base.notrace, \
"trace does not to default to empty tracing"
def bootstrap_socket(io, id):
# XXX: switch to spec
from execnet.gateway_socket import SocketIO
sendexec(
io,
inspect.getsource(gateway_base),
"import socket",
inspect.getsource(SocketIO),
"try: execmodel",
"except NameError:",
" execmodel = get_execmodel('thread')",
"io = SocketIO(clientsock, execmodel)",
"io.write('1'.encode('ascii'))",
"serve(io, id='%s-worker')" % id,
)
s = io.read(1)
assert s == "1".encode("ascii")
def run_me(channel=None):
raise ValueError('me')
if __name__ == '__channelexec__':
run_me()
"""
)
)
monkeypatch.syspath_prepend(tmpdir)
import remotetest
ch = gw.remote_exec(remotetest)
try:
ch.receive()
except execnet.gateway_base.RemoteError as e:
assert 'remotetest.py", line 3, in run_me' in str(e)
assert "ValueError: me" in str(e)
finally:
ch.close()
ch = gw.remote_exec(remotetest.run_me)
try:
ch.receive()
except execnet.gateway_base.RemoteError as e:
assert 'remotetest.py", line 3, in run_me' in str(e)
assert "ValueError: me" in str(e)
finally:
ch.close()
def test_opcodes():
data = vars(gateway_base.opcode)
computed = {k: v for k, v in data.items() if "__" not in k}
assert computed == {
"BUILDTUPLE": "@".encode("ascii"),
"BYTES": "A".encode("ascii"),
"CHANNEL": "B".encode("ascii"),
"FALSE": "C".encode("ascii"),
"FLOAT": "D".encode("ascii"),
"FROZENSET": "E".encode("ascii"),
"INT": "F".encode("ascii"),
"LONG": "G".encode("ascii"),
"LONGINT": "H".encode("ascii"),
"LONGLONG": "I".encode("ascii"),
"NEWDICT": "J".encode("ascii"),
"NEWLIST": "K".encode("ascii"),
"NONE": "L".encode("ascii"),
"PY2STRING": "M".encode("ascii"),
def test_no_tracing_by_default(self):
assert (
gateway_base.trace == gateway_base.notrace
), "trace does not to default to empty tracing"
def test_remoteerror_readable_traceback(self, gw):
e = py.test.raises(gateway_base.RemoteError,
'gw.remote_exec("x y").waitclose()')
assert "gateway_base" in e.value.formatted
def test_serializer_api_version_error(monkeypatch):
bchr = gateway_base.bchr
monkeypatch.setattr(gateway_base, "DUMPFORMAT_VERSION", bchr(1))
dumped = execnet.dumps(42)
monkeypatch.setattr(gateway_base, "DUMPFORMAT_VERSION", bchr(2))
pytest.raises(execnet.DataFormatError, lambda: execnet.loads(dumped))
if isinstance(source, types.ModuleType):
linecache.updatecache(inspect.getsourcefile(source))
source = inspect.getsource(source)
elif isinstance(source, types.FunctionType):
call_name = source.__name__
source = _source_of_function(source)
else:
source = textwrap.dedent(str(source))
if call_name is None and kwargs:
raise TypeError("can't pass kwargs to non-function remote_exec")
channel = self.newchannel()
self._send(Message.CHANNEL_EXEC,
channel.id,
gateway_base.dumps_internal((source, call_name, kwargs)))
return channel