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
80
81
82
83
84
85
|
"""
Tests for the built-in test echo server.
Pyro - Python Remote Objects. Copyright by Irmen de Jong (irmen@razorvine.net).
"""
import time
import unittest
from threading import Thread, Event
import Pyro4.test.echoserver as echoserver
import Pyro4.util
from Pyro4.configuration import config
class EchoServerThread(Thread):
def __init__(self):
super(EchoServerThread, self).__init__()
self.setDaemon(True)
self.started = Event()
self.echodaemon = self.echoserver = self.uri = None
def run(self):
self.echodaemon, self.echoserver, self.uri = echoserver.main(args=["-q"], returnWithoutLooping=True)
self.started.set()
self.echodaemon.requestLoop(loopCondition=lambda: not self.echoserver._must_shutdown)
class TestEchoserver(unittest.TestCase):
def setUp(self):
self.echoserverthread = EchoServerThread()
self.echoserverthread.start()
self.echoserverthread.started.wait()
self.uri = self.echoserverthread.uri
def tearDown(self):
self.echoserverthread.echodaemon.shutdown()
time.sleep(0.02)
self.echoserverthread.join()
config.SERVERTYPE = "thread"
def testExposed(self):
e = Pyro4.test.echoserver.EchoServer()
self.assertTrue(hasattr(e, "_pyroExposed"))
def testEcho(self):
with Pyro4.core.Proxy(self.uri) as echo:
try:
self.assertEqual("hello", echo.echo("hello"))
self.assertEqual(None, echo.echo(None))
self.assertEqual([1, 2, 3], echo.echo([1, 2, 3]))
finally:
echo.shutdown()
def testError(self):
with Pyro4.core.Proxy(self.uri) as echo:
try:
echo.error()
self.fail("expected exception")
except Exception as x:
tb = "".join(Pyro4.util.getPyroTraceback())
self.assertIn("Remote traceback", tb)
self.assertIn("ValueError", tb)
self.assertEqual("expected error from echoserver error() method", str(x))
try:
echo.error_with_text()
self.fail("expected exception")
except Exception as x:
tb = "".join(Pyro4.util.getPyroTraceback())
self.assertIn("Remote traceback", tb)
self.assertIn("ValueError", tb)
self.assertEqual("the message of the error", str(x))
def testGenerator(self):
with Pyro4.core.Proxy(self.uri) as echo:
remotegenerator = echo.generator()
self.assertIsInstance(remotegenerator, Pyro4.core._StreamResultIterator)
next(remotegenerator)
next(remotegenerator)
next(remotegenerator)
with self.assertRaises(StopIteration):
next(remotegenerator)
if __name__ == "__main__":
unittest.main()
|
