""" Tests for the daemon. Pyro - Python Remote Objects. Copyright by Irmen de Jong (irmen@razorvine.net). """ from __future__ import print_function import sys import time import socket import uuid import unittest import Pyro4.core import Pyro4.naming import Pyro4.constants import Pyro4.socketutil import Pyro4.message import Pyro4.util from Pyro4.errors import DaemonError, PyroError, SerializeError from Pyro4.configuration import config from Pyro4.core import current_context from testsupport import * class MyObj(object): def __init__(self, arg): self.arg = arg def __eq__(self, other): return self.arg == other.arg __hash__ = object.__hash__ class CustomDaemonInterface(Pyro4.core.DaemonObject): def __init__(self, daemon): super(CustomDaemonInterface, self).__init__(daemon) def custom_daemon_method(self): return 42 class DaemonTests(unittest.TestCase): # We create a daemon, but notice that we are not actually running the requestloop. # 'on-line' tests are all taking place in another test, to keep this one simple. def setUp(self): config.POLLTIMEOUT = 0.1 def sendHandshakeMessage(self, conn, correlation_id=None): ser = Pyro4.util.get_serializer_by_id(Pyro4.util.MarshalSerializer.serializer_id) data, _ = ser.serializeData({"handshake": "hello", "object": Pyro4.constants.DAEMON_NAME}, False) annotations = {"CORR": correlation_id.bytes} if correlation_id else None msg = Pyro4.message.Message(Pyro4.message.MSG_CONNECT, data, Pyro4.util.MarshalSerializer.serializer_id, 0, 99, annotations=annotations) msg.send(conn) def testSerializerConfig(self): self.assertIsInstance(config.SERIALIZERS_ACCEPTED, set) self.assertIsInstance(config.SERIALIZER, basestring) self.assertGreater(len(config.SERIALIZERS_ACCEPTED), 1) def testSerializerAccepted(self): self.assertIn("marshal", config.SERIALIZERS_ACCEPTED) self.assertNotIn("pickle", config.SERIALIZERS_ACCEPTED) self.assertNotIn("cloudpickle", config.SERIALIZERS_ACCEPTED) self.assertNotIn("dill", config.SERIALIZERS_ACCEPTED) with Pyro4.core.Daemon(port=0) as d: msg = Pyro4.message.Message(Pyro4.message.MSG_INVOKE, b"", Pyro4.util.MarshalSerializer.serializer_id, 0, 0, hmac_key=d._pyroHmacKey) cm = ConnectionMock(msg) d.handleRequest(cm) # marshal serializer should be accepted msg = Pyro4.message.Message(Pyro4.message.MSG_INVOKE, b"", Pyro4.util.PickleSerializer.serializer_id, 0, 0, hmac_key=d._pyroHmacKey) cm = ConnectionMock(msg) try: d.handleRequest(cm) self.fail("should crash") except Pyro4.errors.ProtocolError as x: self.assertIn("serializer that is not accepted", str(x)) pass msg = Pyro4.message.Message(Pyro4.message.MSG_INVOKE, b"", Pyro4.util.CloudpickleSerializer.serializer_id, 0, 0, hmac_key=d._pyroHmacKey) cm = ConnectionMock(msg) try: d.handleRequest(cm) self.fail("should crash") except Pyro4.errors.ProtocolError as x: self.assertTrue("no serializer available for id" in str(x) or "serializer that is not accepted" in str(x)) msg = Pyro4.message.Message(Pyro4.message.MSG_INVOKE, b"", Pyro4.util.DillSerializer.serializer_id, 0, 0, hmac_key=d._pyroHmacKey) cm = ConnectionMock(msg) try: d.handleRequest(cm) self.fail("should crash") except Pyro4.errors.ProtocolError as x: self.assertTrue("no serializer available for id" in str(x) or "serializer that is not accepted" in str(x)) def testDaemon(self): with Pyro4.core.Daemon(port=0) as d: hostname, port = d.locationStr.split(":") port = int(port) self.assertIn(Pyro4.constants.DAEMON_NAME, d.objectsById) self.assertEqual("PYRO:" + Pyro4.constants.DAEMON_NAME + "@" + d.locationStr, str(d.uriFor(Pyro4.constants.DAEMON_NAME))) # check the string representations expected = "" % (id(d), d.locationStr, Pyro4.socketutil.family_str(d.sock)) self.assertEqual(expected, str(d)) self.assertEqual(expected, unicode(d)) self.assertEqual(expected, repr(d)) sockname = d.sock.getsockname() self.assertEqual(port, sockname[1]) daemonobj = d.objectsById[Pyro4.constants.DAEMON_NAME] daemonobj.ping() daemonobj.registered() try: daemonobj.shutdown() self.fail("should not succeed to call unexposed method") except AttributeError: pass def testDaemonCustomInterface(self): with Pyro4.core.Daemon(port=0, interface=CustomDaemonInterface) as d: obj = d.objectsById[Pyro4.constants.DAEMON_NAME] self.assertEqual(42, obj.custom_daemon_method()) def testDaemonConnectedSocket(self): try: Pyro4.config.SERVERTYPE = "thread" with Pyro4.core.Daemon() as d: self.assertTrue("Thread" in d.transportServer.__class__.__name__) s1, s2 = socket.socketpair() with Pyro4.core.Daemon(connected_socket=s1) as d: self.assertTrue(d.locationStr=="./u:<>" or d.locationStr.startswith("127.0.")) self.assertFalse("Thread" in d.transportServer.__class__.__name__) self.assertTrue("Existing" in d.transportServer.__class__.__name__) Pyro4.config.SERVERTYPE = "multiplex" with Pyro4.core.Daemon() as d: self.assertTrue("Multiplex" in d.transportServer.__class__.__name__) s1, s2 = socket.socketpair() with Pyro4.core.Daemon(connected_socket=s1) as d: self.assertTrue(d.locationStr=="./u:<>" or d.locationStr.startswith("127.0.")) self.assertFalse("Multiplex" in d.transportServer.__class__.__name__) self.assertTrue("Existing" in d.transportServer.__class__.__name__) finally: Pyro4.config.SERVERTYPE = "thread" @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "unix domain sockets required") def testDaemonUnixSocket(self): SOCKNAME = "test_unixsocket" with Pyro4.core.Daemon(unixsocket=SOCKNAME) as d: locationstr = "./u:" + SOCKNAME self.assertEqual(locationstr, d.locationStr) self.assertEqual("PYRO:" + Pyro4.constants.DAEMON_NAME + "@" + locationstr, str(d.uriFor(Pyro4.constants.DAEMON_NAME))) # check the string representations expected = "" % (id(d), locationstr) self.assertEqual(expected, str(d)) self.assertEqual(SOCKNAME, d.sock.getsockname()) self.assertEqual(socket.AF_UNIX, d.sock.family) @unittest.skipUnless(hasattr(socket, "AF_UNIX") and sys.platform.startswith("linux"), "unix domain sockets required and linux os") def testDaemonUnixSocketAbstractNS(self): SOCKNAME = "\0test_unixsocket" # mind the \0 at the start, for a Linux abstract namespace socket with Pyro4.core.Daemon(unixsocket=SOCKNAME) as d: locationstr = "./u:" + SOCKNAME self.assertEqual(locationstr, d.locationStr) self.assertEqual("PYRO:" + Pyro4.constants.DAEMON_NAME + "@" + locationstr, str(d.uriFor(Pyro4.constants.DAEMON_NAME))) # check the string representations expected = "" % (id(d), locationstr) self.assertEqual(expected, str(d)) sn_bytes = tobytes(SOCKNAME) self.assertEqual(sn_bytes, d.sock.getsockname()) self.assertEqual(socket.AF_UNIX, d.sock.family) def testServertypeThread(self): old_servertype = config.SERVERTYPE config.SERVERTYPE = "thread" with Pyro4.core.Daemon(port=0) as d: self.assertIn(d.sock, d.sockets, "daemon's socketlist should contain the server socket") self.assertTrue(len(d.sockets) == 1, "daemon without connections should have just 1 socket") config.SERVERTYPE = old_servertype def testServertypeMultiplex(self): old_servertype = config.SERVERTYPE config.SERVERTYPE = "multiplex" with Pyro4.core.Daemon(port=0) as d: self.assertIn(d.sock, d.sockets, "daemon's socketlist should contain the server socket") self.assertTrue(len(d.sockets) == 1, "daemon without connections should have just 1 socket") config.SERVERTYPE = old_servertype def testServertypeFoobar(self): old_servertype = config.SERVERTYPE config.SERVERTYPE = "foobar" self.assertRaises(PyroError, Pyro4.core.Daemon) config.SERVERTYPE = old_servertype def testRegisterTwice(self): with Pyro4.core.Daemon(port=0) as d: o1 = MyObj("object1") d.register(o1) with self.assertRaises(DaemonError) as x: d.register(o1) self.assertEqual("object or class already has a Pyro id", str(x.exception)) d.unregister(o1) d.register(o1, "samename") o2 = MyObj("object2") with self.assertRaises(DaemonError) as x: d.register(o2, "samename") self.assertEqual("an object or class is already registered with that id", str(x.exception)) self.assertTrue(hasattr(o1, "_pyroId")) self.assertTrue(hasattr(o1, "_pyroDaemon")) d.unregister(o1) self.assertFalse(hasattr(o1, "_pyroId")) self.assertFalse(hasattr(o1, "_pyroDaemon")) o1._pyroId = "FOOBAR" with self.assertRaises(DaemonError) as x: d.register(o1) self.assertEqual("object or class already has a Pyro id", str(x.exception)) o1._pyroId = "" d.register(o1) # with empty-string _pyroId register should work def testRegisterTwiceForced(self): with Pyro4.core.Daemon(port=0) as d: o1 = MyObj("object1") d.register(o1, "name1") d.register(o1, "name2", force=True) d.register(o1, "name1", force=True) self.assertIs(d.objectsById["name1"], d.objectsById["name2"]) d.unregister(o1) o1._pyroId = "FOOBAR_ID" d.register(o1, "newname", force=True) self.assertEqual("newname", o1._pyroId) self.assertIn("newname", d.objectsById) def testRegisterEtc(self): with Pyro4.core.Daemon(port=0) as d: self.assertEqual(1, len(d.objectsById)) o1 = MyObj("object1") o2 = MyObj("object2") d.register(o1) self.assertRaises(DaemonError, d.register, o2, Pyro4.constants.DAEMON_NAME) # cannot use daemon name d.register(o2, "obj2a") self.assertEqual(3, len(d.objectsById)) self.assertEqual(o1, d.objectsById[o1._pyroId]) self.assertEqual(o2, d.objectsById["obj2a"]) self.assertEqual("obj2a", o2._pyroId) self.assertEqual(d, o2._pyroDaemon) # test unregister d.unregister("unexisting_thingie") self.assertRaises(ValueError, d.unregister, None) d.unregister("obj2a") d.unregister(o1._pyroId) self.assertEqual(1, len(d.objectsById)) self.assertNotIn(o1._pyroId, d.objectsById) self.assertNotIn(o2._pyroId, d.objectsById) # test unregister objects del o2._pyroId d.register(o2) objectid = o2._pyroId self.assertIn(objectid, d.objectsById) self.assertEqual(2, len(d.objectsById)) d.unregister(o2) # no more _pyro attributes must remain after unregistering for attr in vars(o2): self.assertFalse(attr.startswith("_pyro")) self.assertEqual(1, len(d.objectsById)) self.assertNotIn(objectid, d.objectsById) self.assertRaises(DaemonError, d.unregister, [1, 2, 3]) # test unregister daemon name d.unregister(Pyro4.constants.DAEMON_NAME) self.assertIn(Pyro4.constants.DAEMON_NAME, d.objectsById) # weird args w = MyObj("weird") self.assertRaises(AttributeError, d.register, None) self.assertRaises(AttributeError, d.register, 4444) self.assertRaises(TypeError, d.register, w, 666) # uri return value from register uri = d.register(MyObj("xyz")) self.assertIsInstance(uri, Pyro4.core.URI) uri = d.register(MyObj("xyz"), "test.register") self.assertEqual("test.register", uri.object) def testRegisterClass(self): with Pyro4.core.Daemon(port=0) as d: self.assertEqual(1, len(d.objectsById)) d.register(MyObj) with self.assertRaises(DaemonError): d.register(MyObj) self.assertEqual(2, len(d.objectsById)) d.uriFor(MyObj) # unregister: d.unregister(MyObj) self.assertEqual(1, len(d.objectsById)) def testRegisterUnicode(self): with Pyro4.core.Daemon(port=0) as d: myobj1 = MyObj("hello1") myobj2 = MyObj("hello2") myobj3 = MyObj("hello3") uri1 = d.register(myobj1, "str_name") uri2 = d.register(myobj2, unicode("unicode_name")) uri3 = d.register(myobj3, "unicode_" + unichr(0x20ac)) self.assertEqual(4, len(d.objectsById)) uri = d.uriFor(myobj1) self.assertEqual(uri1, uri) _ = Pyro4.core.Proxy(uri) uri = d.uriFor(myobj2) self.assertEqual(uri2, uri) _ = Pyro4.core.Proxy(uri) uri = d.uriFor(myobj3) self.assertEqual(uri3, uri) _ = Pyro4.core.Proxy(uri) uri = d.uriFor("str_name") self.assertEqual(uri1, uri) _ = Pyro4.core.Proxy(uri) uri = d.uriFor(unicode("unicode_name")) self.assertEqual(uri2, uri) _ = Pyro4.core.Proxy(uri) uri = d.uriFor("unicode_" + unichr(0x20ac)) self.assertEqual(uri3, uri) _ = Pyro4.core.Proxy(uri) def testDaemonObject(self): with Pyro4.core.Daemon(port=0) as d: daemon = Pyro4.core.DaemonObject(d) obj1 = MyObj("object1") obj2 = MyObj("object2") obj3 = MyObj("object2") d.register(obj1, "obj1") d.register(obj2, "obj2") d.register(obj3) daemon.ping() registered = daemon.registered() self.assertTrue(type(registered) is list) self.assertEqual(4, len(registered)) self.assertIn("obj1", registered) self.assertIn("obj2", registered) self.assertIn(obj3._pyroId, registered) try: daemon.shutdown() self.fail("should not succeed to call unexposed method") except AttributeError: pass def testUriFor(self): d = Pyro4.core.Daemon(port=0) try: o1 = MyObj("object1") o2 = MyObj("object2") self.assertRaises(DaemonError, d.uriFor, o1) self.assertRaises(DaemonError, d.uriFor, o2) d.register(o1, None) d.register(o2, "object_two") o3 = MyObj("object3") self.assertRaises(DaemonError, d.uriFor, o3) # can't get an uri for an unregistered object (note: unregistered name is ok) u1 = d.uriFor(o1) u2 = d.uriFor(o2._pyroId) u3 = d.uriFor("unexisting_thingie") # unregistered name is no problem, it's just an uri we're requesting u4 = d.uriFor(o2) self.assertEqual(Pyro4.core.URI, type(u1)) self.assertEqual("PYRO", u1.protocol) self.assertEqual("PYRO", u2.protocol) self.assertEqual("PYRO", u3.protocol) self.assertEqual("PYRO", u4.protocol) self.assertEqual("object_two", u4.object) self.assertEqual(Pyro4.core.URI("PYRO:unexisting_thingie@" + d.locationStr), u3) finally: d.close() def testDaemonWithStmt(self): d = Pyro4.core.Daemon() self.assertIsNotNone(d.transportServer) d.close() # closes the transportserver and sets it to None self.assertIsNone(d.transportServer) with Pyro4.core.Daemon() as d: self.assertIsNotNone(d.transportServer) pass self.assertIsNone(d.transportServer) try: with Pyro4.core.Daemon() as d: print(1 // 0) # cause an error self.fail("expected error") except ZeroDivisionError: pass self.assertIsNone(d.transportServer) d = Pyro4.core.Daemon() with d: pass try: with d: pass self.fail("expected error") except PyroError: # you cannot re-use a daemon object in multiple with statements pass d.close() def testRequestloopCondition(self): with Pyro4.core.Daemon(port=0) as d: condition = lambda: False start = time.time() d.requestLoop(loopCondition=condition) # this should return almost immediately duration = time.time() - start self.assertLess(duration, 0.4) def testSimpleHandshake(self): conn = ConnectionMock() with Pyro4.core.Daemon(port=0) as d: self.sendHandshakeMessage(conn) success = d._handshake(conn) self.assertTrue(success) msg = Pyro4.message.Message.recv(conn, hmac_key=d._pyroHmacKey) self.assertEqual(Pyro4.message.MSG_CONNECTOK, msg.type) self.assertEqual(99, msg.seq) def testHandshakeDenied(self): class HandshakeFailDaemon(Pyro4.core.Daemon): def validateHandshake(self, conn, data): raise ValueError("handshake fail validation error") conn = ConnectionMock() with HandshakeFailDaemon(port=0) as d: self.sendHandshakeMessage(conn) success = d._handshake(conn) self.assertFalse(success) msg = Pyro4.message.Message.recv(conn, hmac_key=d._pyroHmacKey) self.assertEqual(Pyro4.message.MSG_CONNECTFAIL, msg.type) self.assertEqual(99, msg.seq) self.assertTrue(b"handshake fail validation error" in msg.data) with Pyro4.core.Daemon(port=0) as d: self.sendHandshakeMessage(conn) success = d._handshake(conn, denied_reason="no way, handshake denied") self.assertFalse(success) msg = Pyro4.message.Message.recv(conn, hmac_key=d._pyroHmacKey) self.assertEqual(Pyro4.message.MSG_CONNECTFAIL, msg.type) self.assertEqual(99, msg.seq) self.assertTrue(b"no way, handshake denied" in msg.data) def testCustomHandshake(self): conn = ConnectionMock() class CustomHandshakeDaemon(Pyro4.core.Daemon): def validateHandshake(self, conn, data): return ["sure", "have", "fun"] def annotations(self): return {"XYZZ": b"custom annotation set by daemon"} with CustomHandshakeDaemon(port=0) as d: corr_id = uuid.uuid4() self.sendHandshakeMessage(conn, correlation_id=corr_id) self.assertNotEqual(corr_id, current_context.correlation_id) success = d._handshake(conn) self.assertTrue(success) self.assertEqual(corr_id, current_context.correlation_id) msg = Pyro4.message.Message.recv(conn, hmac_key=d._pyroHmacKey) self.assertEqual(Pyro4.message.MSG_CONNECTOK, msg.type) self.assertEqual(99, msg.seq) self.assertEqual(2, len(msg.annotations)) self.assertEqual(corr_id.bytes, msg.annotations["CORR"]) self.assertEqual(b"custom annotation set by daemon", msg.annotations["XYZZ"]) ser = Pyro4.util.get_serializer_by_id(msg.serializer_id) data = ser.deserializeData(msg.data, msg.flags & Pyro4.message.FLAGS_COMPRESSED) self.assertEqual(["sure", "have", "fun"], data) def testNAT(self): with Pyro4.core.Daemon() as d: self.assertIsNone(d.natLocationStr) with Pyro4.core.Daemon(nathost="nathosttest", natport=12345) as d: self.assertEqual("nathosttest:12345", d.natLocationStr) self.assertNotEqual(d.locationStr, d.natLocationStr) uri = d.register(MyObj(1)) self.assertEqual("nathosttest:12345", uri.location) uri = d.uriFor("object") self.assertEqual("nathosttest:12345", uri.location) uri = d.uriFor("object", nat=False) self.assertNotEqual("nathosttest:12345", uri.location) d = Pyro4.core.Daemon(nathost="bla") self.assertTrue(d.natLocationStr.startswith("bla:")) try: _ = Pyro4.core.Daemon(natport=5555) self.fail("expected error") except ValueError: pass try: _ = Pyro4.core.Daemon(nathost="bla", natport=5555, unixsocket="testsock") self.fail("expected error") except ValueError: pass def testNATzeroPort(self): servertype = config.SERVERTYPE try: config.SERVERTYPE = "multiplex" with Pyro4.core.Daemon(nathost="nathosttest", natport=99999) as d: host, port = d.locationStr.split(":") self.assertNotEqual(99999, port) self.assertEqual("nathosttest:99999", d.natLocationStr) with Pyro4.core.Daemon(nathost="nathosttest", natport=0) as d: host, port = d.locationStr.split(":") self.assertEqual("nathosttest:%s" % port, d.natLocationStr) config.SERVERTYPE = "thread" with Pyro4.core.Daemon(nathost="nathosttest", natport=99999) as d: host, port = d.locationStr.split(":") self.assertNotEqual(99999, port) self.assertEqual("nathosttest:99999", d.natLocationStr) with Pyro4.core.Daemon(nathost="nathosttest", natport=0) as d: host, port = d.locationStr.split(":") self.assertEqual("nathosttest:%s" % port, d.natLocationStr) finally: config.SERVERTYPE = servertype def testNATconfig(self): try: config.NATHOST = None config.NATPORT = 0 with Pyro4.core.Daemon() as d: self.assertIsNone(d.natLocationStr) config.NATHOST = "nathosttest" config.NATPORT = 12345 with Pyro4.core.Daemon() as d: self.assertEqual("nathosttest:12345", d.natLocationStr) finally: config.NATHOST = None config.NATPORT = 0 def testBehaviorDefaults(self): class TestClass: pass with Pyro4.core.Daemon() as d: d.register(TestClass) instance_mode, instance_creator = TestClass._pyroInstancing self.assertEqual("session", instance_mode) self.assertIsNone(instance_creator) def testInstanceCreationSingle(self): def creator(clazz): return clazz("testname") @Pyro4.core.behavior(instance_mode="single", instance_creator=creator) class TestClass: def __init__(self, name): self.name = name conn = Pyro4.socketutil.SocketConnection(socket.socket()) d = Pyro4.core.Daemon() instance1 = d._getInstance(TestClass, conn) instance2 = d._getInstance(TestClass, conn) self.assertEqual("testname", instance1.name) self.assertIs(instance1, instance2) self.assertTrue(TestClass in d._pyroInstances) self.assertIs(instance1, d._pyroInstances[TestClass]) self.assertFalse(TestClass in conn.pyroInstances) def testBehaviorDefaultsIsSession(self): class ClassWithDefaults: def __init__(self): self.name = "yep" conn1 = Pyro4.socketutil.SocketConnection(socket.socket()) conn2 = Pyro4.socketutil.SocketConnection(socket.socket()) d = Pyro4.core.Daemon() d.register(ClassWithDefaults) instance1a = d._getInstance(ClassWithDefaults, conn1) instance1b = d._getInstance(ClassWithDefaults, conn1) instance2a = d._getInstance(ClassWithDefaults, conn2) instance2b = d._getInstance(ClassWithDefaults, conn2) self.assertIs(instance1a, instance1b) self.assertIs(instance2a, instance2b) self.assertIsNot(instance1a, instance2a) self.assertFalse(ClassWithDefaults in d._pyroInstances) self.assertTrue(ClassWithDefaults in conn1.pyroInstances) self.assertTrue(ClassWithDefaults in conn2.pyroInstances) self.assertIs(instance1a, conn1.pyroInstances[ClassWithDefaults]) self.assertIs(instance2a, conn2.pyroInstances[ClassWithDefaults]) def testInstanceCreationSession(self): def creator(clazz): return clazz("testname") @Pyro4.core.behavior(instance_mode="session", instance_creator=creator) class ClassWithDecorator: def __init__(self, name): self.name = name conn1 = Pyro4.socketutil.SocketConnection(socket.socket()) conn2 = Pyro4.socketutil.SocketConnection(socket.socket()) d = Pyro4.core.Daemon() d.register(ClassWithDecorator) # check the class with the decorator first instance1a = d._getInstance(ClassWithDecorator, conn1) instance1b = d._getInstance(ClassWithDecorator, conn1) instance2a = d._getInstance(ClassWithDecorator, conn2) instance2b = d._getInstance(ClassWithDecorator, conn2) self.assertIs(instance1a, instance1b) self.assertIs(instance2a, instance2b) self.assertIsNot(instance1a, instance2a) self.assertFalse(ClassWithDecorator in d._pyroInstances) self.assertTrue(ClassWithDecorator in conn1.pyroInstances) self.assertTrue(ClassWithDecorator in conn2.pyroInstances) self.assertIs(instance1a, conn1.pyroInstances[ClassWithDecorator]) self.assertIs(instance2a, conn2.pyroInstances[ClassWithDecorator]) def testInstanceCreationPerCall(self): def creator(clazz): return clazz("testname") @Pyro4.core.behavior(instance_mode="percall", instance_creator=creator) class TestClass: def __init__(self, name): self.name = name with Pyro4.socketutil.SocketConnection(socket.socket()) as conn: with Pyro4.core.Daemon() as d: instance1 = d._getInstance(TestClass, conn) instance2 = d._getInstance(TestClass, conn) self.assertIsNot(instance1, instance2) self.assertFalse(TestClass in d._pyroInstances) self.assertFalse(TestClass in conn.pyroInstances) def testInstanceCreationWrongType(self): def creator(clazz): return Pyro4.core.URI("PYRO:test@localhost:9999") @Pyro4.core.behavior(instance_creator=creator) class TestClass: def method(self): pass with Pyro4.socketutil.SocketConnection(socket.socket()) as conn: with Pyro4.core.Daemon() as d: with self.assertRaises(TypeError): d._getInstance(TestClass, conn) def testCombine(self): d1 = Pyro4.core.Daemon() d2 = Pyro4.core.Daemon() with self.assertRaises(TypeError): d1.combine(d2) d1.close() d2.close() try: config.SERVERTYPE = "multiplex" d1 = Pyro4.core.Daemon() d2 = Pyro4.core.Daemon() nsuri, nsd, bcd = Pyro4.naming.startNS(host="", bchost="") d1_selector = d1.transportServer.selector d1.combine(d2) d1.combine(nsd) d1.combine(bcd) self.assertIs(d1_selector, d1.transportServer.selector) self.assertIs(d1_selector, d2.transportServer.selector) self.assertIs(d1_selector, nsd.transportServer.selector) self.assertIs(d1_selector, bcd.transportServer.selector) self.assertEqual(4, len(d1.sockets)) self.assertIn(d1.sock, d1.sockets) self.assertIn(d2.sock, d1.sockets) self.assertIn(nsd.sock, d1.sockets) self.assertIn(bcd, d1.sockets) bcd.close() nsd.close() d2.close() d1.close() finally: config.SERVERTYPE = "thread" class MetaInfoTests(unittest.TestCase): def testMeta(self): with Pyro4.core.Daemon() as d: daemon_obj = d.objectsById[Pyro4.constants.DAEMON_NAME] self.assertTrue(len(daemon_obj.info()) > 10) meta = daemon_obj.get_metadata(Pyro4.constants.DAEMON_NAME) self.assertEqual({"get_metadata", "get_next_stream_item", "close_stream", "info", "ping", "registered"}, meta["methods"]) def testMetaSerialization(self): with Pyro4.core.Daemon() as d: daemon_obj = d.objectsById[Pyro4.constants.DAEMON_NAME] meta = daemon_obj.get_metadata(Pyro4.constants.DAEMON_NAME) for ser_id in [Pyro4.util.JsonSerializer.serializer_id, Pyro4.util.MarshalSerializer.serializer_id, Pyro4.util.PickleSerializer.serializer_id, Pyro4.util.SerpentSerializer.serializer_id]: serializer = Pyro4.util.get_serializer_by_id(ser_id) data = serializer.dumps(meta) _ = serializer.loads(data) try: serializer = Pyro4.util.get_serializer_by_id(Pyro4.util.DillSerializer.serializer_id) except SerializeError: # dill doesn't work with ironpython so we allow an error here in that case if sys.platform != "cli": raise else: data = serializer.dumps(meta) _ = serializer.loads(data) def testMetaResetCache(self): class Dummy: @Pyro4.core.expose def method(self): pass with Pyro4.core.Daemon() as d: dummy = Dummy() uri = d.register(dummy) daemon_obj = d.objectsById[Pyro4.constants.DAEMON_NAME] meta = daemon_obj.get_metadata(uri.object) self.assertNotIn("newly_added_method", meta["methods"]) self.assertNotIn("newly_added_method_two", meta["methods"]) Dummy.newly_added_method = Pyro4.core.expose(lambda self: None) meta = daemon_obj.get_metadata(uri.object) self.assertNotIn("newly_added_method", meta["methods"]) d.resetMetadataCache(uri.object) meta = daemon_obj.get_metadata(uri.object) self.assertIn("newly_added_method", meta["methods"]) Dummy.newly_added_method_two = Pyro4.core.expose(lambda self: None) d.resetMetadataCache(dummy) meta = daemon_obj.get_metadata(uri.object) self.assertIn("newly_added_method_two", meta["methods"]) if __name__ == "__main__": # import sys;sys.argv = ['', 'Test.testName'] unittest.main()