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
|
from __future__ import with_statement
import rpyc
import unittest
class MyService(rpyc.Service):
def exposed_set_version(self):
rpyc.version.__version__ = '1.0.0'
def exposed_remote_assert(self, val):
assert val
class TestRemoteException(unittest.TestCase):
def setUp(self):
self.server = rpyc.utils.server.OneShotServer(MyService, port=0)
self.server.logger.quiet = False
self.server._start_in_thread()
self.original_version_string = rpyc.version.__version__
self.conn = rpyc.connect("localhost", port=self.server.port)
def tearDown(self):
rpyc.version.__version__ = self.original_version_string
self.conn.close()
def test_remote_exception(self):
# Since the server/client share the same namespace, the version will change for both.
# Even so, this should suffice for unit testing
warn_msg = 'WARNING: Remote is on RPyC 1.0.0 and local is on RPyC 1.0.0.'
try:
self.conn.root.remote_assert(False)
except Exception as exc:
exc_rpyc_version = exc._remote_version
exc_remote_tb = exc._remote_tb
else:
exc_rpyc_version = None
exc_remote_tb = ''
self.assertEqual(self.original_version_string, exc_rpyc_version)
self.assertFalse(warn_msg in exc_remote_tb)
try:
self.conn.root.set_version()
self.conn.root.remote_assert(False)
except Exception as exc:
exc_rpyc_version = exc._remote_version
exc_remote_tb = exc._remote_tb
else:
exc_rpyc_version = None
exc_remote_tb = ''
self.assertEqual('1.0.0', exc_rpyc_version)
self.assertTrue(warn_msg in exc_remote_tb)
class TestExclusionsRemoteException(unittest.TestCase):
def setUp(self):
config = {'include_local_traceback': False, 'include_local_version': False}
self.server = rpyc.utils.server.OneShotServer(MyService, port=0, protocol_config=config)
self.server.logger.quiet = False
self.server._start_in_thread()
self.conn = rpyc.connect("localhost", port=self.server.port)
def tearDown(self):
self.conn.close()
def test_remote_exception(self):
try:
self.conn.root.remote_assert(False)
except Exception as exc:
exc_rpyc_version = exc._remote_version
exc_remote_tb = exc._remote_tb
else:
exc_rpyc_version = None
exc_remote_tb = ''
self.assertEqual("<traceback denied>", exc_remote_tb)
self.assertEqual("<version denied>", exc_rpyc_version)
if __name__ == "__main__":
unittest.main()
|
