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
|
import os
import rpyc
import unittest
class ClassicMode(unittest.TestCase):
def setUp(self):
self.conn = rpyc.classic.connect_thread()
def tearDown(self):
self.conn.close()
self.conn = None
def test_piped_server(self):
# this causes the following lines to be printed to stderr on Windows:
#
# close failed in file object destructor:
# IOError: [Errno 9] Bad file descriptor
# close failed in file object destructor:
# IOError: [Errno 9] Bad file descriptor
#
# this is because the pipe objects that hold the child process' stdin
# and stdout were disowned by Win32PipeStream (it forcefully takes
# ownership of the file handles). so when the low-level pipe objects
# are gc'ed, they cry that their fd is already closed. this is all
# considered harmless, but there's no way to disable that message
# to stderr
server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "bin", "rpyc_classic.py")
conn = rpyc.classic.connect_subproc(server_file)
conn.modules.sys.path.append("xxx")
self.assertEqual(conn.modules.sys.path.pop(-1), "xxx")
conn.close()
self.assertEqual(conn.proc.wait(), 0)
def test_buffiter(self):
bi = rpyc.buffiter(self.conn.builtin.range(10000))
self.assertEqual(list(bi), list(range(10000)))
def test_classic(self):
self.conn.execute("x = 5")
self.assertEqual(self.conn.namespace["x"], 5)
self.assertEqual(self.conn.eval("1+x"), 6)
def test_mock_connection(self):
from rpyc.utils.classic import MockClassicConnection
import sys
import xml.dom.minidom
conn = MockClassicConnection()
self.assertTrue(conn.modules.sys is sys)
self.assertTrue(conn.modules["xml.dom.minidom"].Element is xml.dom.minidom.Element)
self.assertTrue(conn.builtin.open is open)
self.assertEqual(conn.eval("2+3"), 5)
def test_modules(self):
self.assertIn('tests.test_magic', self.conn.modules)
self.assertNotIn('test_badmagic', self.conn.modules)
self.assertIsNone(self.conn.builtins.locals()['self']._last_traceback)
if __name__ == "__main__":
unittest.main()
|
