Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "bin", "rpyc_classic.py")
self.proc = subprocess.Popen([sys.executable, server_file, "--mode=oneshot", "--host=localhost", "-p0"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
line = self.proc.stdout.readline().strip()
if not line:
print(self.proc.stderr.read())
self.fail("server failed to start")
self.assertEqual(line, b("rpyc-oneshot"), "server failed to start")
host, port = self.proc.stdout.readline().strip().split(b("\t"))
self.conn = rpyc.classic.connect(host, int(port))
def setUp(self):
self.conn = rpyc.classic.connect_thread()
def test_connection(self):
with rpyc.classic.connect("localhost", port=18878) as c:
c.execute("x = 5")
self.assertEqual(c.namespace["x"], 5)
self.assertEqual(c.eval("1+x"), 6)
def setUp(self):
self.conn = rpyc.classic.connect_thread()
def test_multiple_connections(self):
def get_ident(gevent):
return gevent.monkey.get_original('threading', 'get_ident')()
c1 = rpyc.classic.connect("localhost", port=18878)
c2 = rpyc.classic.connect("localhost", port=18878)
c3 = rpyc.classic.connect("localhost", port=18878)
with c1, c2, c3:
id0 = get_ident(gevent)
id1 = get_ident(c1.modules.gevent)
id2 = get_ident(c2.modules.gevent)
id3 = get_ident(c3.modules.gevent)
# all server greenlets and clients running in same OS thread ;)
self.assertEqual(id0, id1)
self.assertEqual(id1, id2)
self.assertEqual(id1, id3)
def remote(host="localhost", port=DEFAULT_SERVER_PORT):
srv = rpyc.classic.connect(host, port=port) #server
cl = rpyc.classic.connect(host, port=port) #client
return (cl, srv)
def read_stats(self):
while True:
try:
self.stats = rpyc.classic.obtain(self._conn.root.get_stats())
except Exception:
try:
self._conn.close()
except Exception:
pass
return
sleep(1)
class Number(object):
def __init__(self,number):
self.number=number
def f(number):
print( number.number)
@contextmanager
def ASYNC(func):
wrapper = rpyc.async(func)
yield wrapper
if __name__ == "__main__":
conn = rpyc.classic.connect("localhost")
conn.modules.sys.path.append(os.path.dirname(__file__))
mod = conn.modules["issue14"]
n = Number(999)
#f = rpyc.async(mod.f)(n)
#print( f )
#print( f.value )
f2 = rpyc.async(mod.f)
res = f2(n)
print res.value
with ASYNC(mod.f) as f2:
res = f2(n)
print res.value
def connect_ida(port):
link = rpyc.classic.connect("localhost", port)
l.debug("Connected!")
idc = link.root.getmodule("idc")
idaapi = link.root.getmodule("idaapi")
idautils = link.root.getmodule("idautils")
return link, idc, idaapi, idautils
def remote(host="localhost", port=DEFAULT_SERVER_PORT):
srv = rpyc.classic.connect(host, port=port) #server
cl = rpyc.classic.connect(host, port=port) #client
return (cl, srv)