"""
Tests for a running Pyro server, without timeouts.

Pyro - Python Remote Objects.  Copyright by Irmen de Jong (irmen@razorvine.net).
"""

from __future__ import print_function
import time
import sys
import threading
import uuid
import unittest
import serpent
import Pyro4.core
import Pyro4.errors
import Pyro4.util
import Pyro4.message
import Pyro4.socketutil
from Pyro4.configuration import config
from testsupport import *


@Pyro4.core.expose
class ServerTestObject(object):
    something = 99
    dict_attr = {}

    def __init__(self):
        self._dictionary = {"number": 42}
        self.dict_attr = {"number2": 43}
        self._value = 12345

    def getDict(self):
        return self._dictionary

    def getDictAttr(self):
        return self.dict_attr

    def multiply(self, x, y):
        return x * y

    def divide(self, x, y):
        return x // y

    def ping(self):
        pass

    def echo(self, obj):
        return obj

    def blob(self, blob):
        return blob.info, blob.deserialized()

    @Pyro4.core.oneway
    def oneway_delay(self, delay):
        time.sleep(delay)

    def delay(self, delay):
        time.sleep(delay)
        return "slept %d seconds" % delay

    def delayAndId(self, delay, id):
        time.sleep(delay)
        return "slept for " + str(id)

    def testargs(self, x, *args, **kwargs):
        return [x, list(args), kwargs]  # don't return tuples, this enables us to test json serialization as well.

    def nonserializableException(self):
        raise NonserializableError(("xantippe", lambda x: 0))

    @Pyro4.core.oneway
    def oneway_multiply(self, x, y):
        return x * y

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, newvalue):
        self._value = newvalue

    @property
    def dictionary(self):
        return self._dictionary

    def iterator(self):
        return iter(["one", "two", "three"])

    def generator(self):
        yield "one"
        yield "two"
        yield "three"
        yield "four"
        yield "five"

    def response_annotation(self):
        # part of the annotations tests
        if "XYZZ" not in Pyro4.core.current_context.annotations:
            raise ValueError("XYZZ should be present in annotations in the daemon")
        if Pyro4.core.current_context.annotations["XYZZ"] != b"data from proxy via new api":
            raise ValueError("XYZZ annotation has wrong data")
        Pyro4.core.current_context.response_annotations["ANN2"] = b"daemon annotation via new api"
        return {"annotations_in_daemon": Pyro4.core.current_context.annotations}

    def new_test_object(self):
        return ServerTestObject()


class NotEverythingExposedClass(object):
    def __init__(self, name):
        self.name = name

    @Pyro4.core.expose
    def getName(self):
        return self.name

    def unexposed(self):
        return "you should not see this"    # .... only when REQUIRE_EXPOSE is set to True is this valid


class DaemonLoopThread(threading.Thread):
    def __init__(self, pyrodaemon):
        super(DaemonLoopThread, self).__init__()
        self.setDaemon(True)
        self.pyrodaemon = pyrodaemon
        self.running = threading.Event()
        self.running.clear()

    def run(self):
        self.running.set()
        try:
            self.pyrodaemon.requestLoop()
        except Pyro4.errors.CommunicationError:
            pass  # ignore pyro communication errors


class DaemonWithSabotagedHandshake(Pyro4.core.Daemon):
    def _handshake(self, conn, denied_reason=None):
        # receive the client's handshake data
        msg = Pyro4.message.Message.recv(conn, [Pyro4.message.MSG_CONNECT], self._pyroHmacKey)
        # return a CONNECTFAIL always
        serializer = Pyro4.util.get_serializer_by_id(msg.serializer_id)
        data, _ = serializer.serializeData("rigged connection failure", compress=False)
        msg = Pyro4.message.Message(Pyro4.message.MSG_CONNECTFAIL, data, serializer.serializer_id, 0, 1, hmac_key=self._pyroHmacKey)
        conn.send(msg.to_bytes())
        return False


class ServerTestsBrokenHandshake(unittest.TestCase):
    def setUp(self):
        config.LOGWIRE = True
        config.SERIALIZERS_ACCEPTED.add("pickle")
        self.daemon = DaemonWithSabotagedHandshake(port=0)
        obj = ServerTestObject()
        uri = self.daemon.register(obj, "something")
        self.objectUri = uri
        self.daemonthread = DaemonLoopThread(self.daemon)
        self.daemonthread.start()
        self.daemonthread.running.wait()
        time.sleep(0.05)

    def tearDown(self):
        time.sleep(0.05)
        self.daemon.shutdown()
        self.daemonthread.join()
        config.SERIALIZERS_ACCEPTED.discard("pickle")

    def testDaemonConnectFail(self):
        # check what happens when the daemon responds with a failed connection msg
        with Pyro4.core.Proxy(self.objectUri) as p:
            try:
                p.ping()
                self.fail("expected CommunicationError")
            except Pyro4.errors.CommunicationError as x:
                message = str(x)
                self.assertIn("rejected:", message)
                self.assertIn("rigged connection failure", message)


class ServerTestsOnce(unittest.TestCase):
    """tests that are fine to run with just a single server type"""

    def setUp(self):
        config.LOGWIRE = True
        config.SERIALIZERS_ACCEPTED.add("pickle")
        self.daemon = Pyro4.core.Daemon(port=0)
        obj = ServerTestObject()
        uri = self.daemon.register(obj, "something")
        self.objectUri = uri
        obj2 = NotEverythingExposedClass("hello")
        self.daemon.register(obj2, "unexposed")
        self.daemonthread = DaemonLoopThread(self.daemon)
        self.daemonthread.start()
        self.daemonthread.running.wait()
        time.sleep(0.05)

    def tearDown(self):
        time.sleep(0.05)
        if self.daemon is not None:
            self.daemon.shutdown()
            self.daemonthread.join()
        config.SERIALIZERS_ACCEPTED.discard("pickle")

    def testPingMessage(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            p._pyroBind()
            conn = p._pyroConnection
            msg = Pyro4.message.Message(Pyro4.message.MSG_PING, b"something", 42, 0, 999, hmac_key=p._pyroHmacKey)
            conn.send(msg.to_bytes())
            msg = Pyro4.message.Message.recv(conn, [Pyro4.message.MSG_PING], hmac_key=p._pyroHmacKey)
            self.assertEqual(Pyro4.message.MSG_PING, msg.type)
            self.assertEqual(999, msg.seq)
            self.assertEqual(b"pong", msg.data)
            Pyro4.message.Message.ping(p._pyroConnection)  # the convenience method that does the above

    def testSequence(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            p.echo(1)
            p.echo(2)
            p.echo(3)
            self.assertEqual(3, p._pyroSeq, "should have 3 method calls")
            p._pyroSeq = 999   # hacking the seq nr won't have any effect because it is the reply from the server that is checked
            self.assertEqual(42, p.echo(42))

    def testMetaOffAttrs(self):
        try:
            old_meta = config.METADATA
            config.METADATA = False
            # should fail here, because there is no meta info about attributes
            with Pyro4.core.Proxy(self.objectUri) as p:
                self.assertEqual(55, p.multiply(5, 11))
                x = p.getDict()
                self.assertEqual({"number": 42}, x)
                # property
                with self.assertRaises(AttributeError):
                    p.dictionary.update({"more": 666})
                # attribute
                with self.assertRaises(AttributeError):
                    p.dict_attr.update({"more": 666})
                x = p.getDict()
                self.assertEqual({"number": 42}, x)
        finally:
            config.METADATA = old_meta

    def testMetaOnAttrs(self):
        try:
            old_meta = config.METADATA
            config.METADATA = True
            with Pyro4.core.Proxy(self.objectUri) as p:
                self.assertEqual(55, p.multiply(5, 11))
                # property
                x = p.getDict()
                self.assertEqual({"number": 42}, x)
                p.dictionary.update({"more": 666})  # should not fail because metadata is enabled and the dictionary property is retrieved as local copy
                x = p.getDict()
                self.assertEqual({"number": 42}, x)  # not updated remotely because we had a local copy
            with Pyro4.core.Proxy(self.objectUri) as p:
                with self.assertRaises(AttributeError):
                    # attribute should fail (meta only works for exposed properties)
                    p.dict_attr.update({"more": 666})
        finally:
            config.METADATA = old_meta

    def testSomeArgumentTypes(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            self.assertEqual([1, [], {}], p.testargs(1))
            self.assertEqual([1, [2, 3], {'a': 4}], p.testargs(1, 2, 3, a=4))
            self.assertEqual([1, [], {'a': 2}], p.testargs(1, **{'a': 2}))

    def testUnicodeKwargs(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            self.assertEqual([1, [], {unichr(65): 2}], p.testargs(1, **{unichr(65): 2}))
            result = p.testargs(unichr(0x20ac), **{unichr(0x20ac): 2})
            self.assertEqual(result[0], unichr(0x20ac))
            key = list(result[2].keys())[0]
            self.assertTrue(type(key) is unicode)
            self.assertEqual(key, unichr(0x20ac))

    def testNormalProxy(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            self.assertEqual(42, p.multiply(7, 6))

    def testExceptions(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            try:
                p.divide(1, 0)
                self.fail("should crash")
            except ZeroDivisionError:
                pass
            try:
                p.multiply("a", "b")
                self.fail("should crash")
            except TypeError:
                pass

    def testProxyMetadata(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            # unconnected proxies have empty metadata
            self.assertEqual(set(), p._pyroAttrs)
            self.assertEqual(set(), p._pyroMethods)
            self.assertEqual(set(), p._pyroOneway)
            # connecting it should obtain metadata (as long as METADATA is true)
            p._pyroBind()
            self.assertEqual({'value', 'dictionary'}, p._pyroAttrs)
            self.assertEqual({'echo', 'getDict', 'divide', 'nonserializableException', 'ping', 'oneway_delay', 'delayAndId', 'delay', 'testargs',
                                  'multiply', 'oneway_multiply', 'getDictAttr', 'iterator', 'generator', 'response_annotation', 'blob', 'new_test_object'}, p._pyroMethods)
            self.assertEqual({'oneway_multiply', 'oneway_delay'}, p._pyroOneway)
            p._pyroAttrs = None
            p._pyroGetMetadata()
            self.assertEqual({'value', 'dictionary'}, p._pyroAttrs)
            p._pyroAttrs = None
            p._pyroGetMetadata(self.objectUri.object)
            self.assertEqual({'value', 'dictionary'}, p._pyroAttrs)
            p._pyroAttrs = None
            p._pyroGetMetadata(known_metadata={"attrs": set(), "oneway": set(), "methods": {"ping"}})
            self.assertEqual(set(), p._pyroAttrs)

    def testProxyRepr(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            p._pyroBind()
            expected = "<Pyro4.core.Proxy at 0x%x; connected %s; for %s>" % (id(p), Pyro4.socketutil.family_str(self.daemon.sock), self.objectUri)
            self.assertEqual(expected, str(p))
            self.assertEqual(expected, repr(p))

    def testProxyAttrsMetadataOff(self):
        try:
            config.METADATA = False
            # read attributes
            with Pyro4.core.Proxy(self.objectUri) as p:
                a = p.multiply
                self.assertIsInstance(a, Pyro4.core._RemoteMethod)
                a = p.value
                self.assertIsInstance(a, Pyro4.core._RemoteMethod)
                a = p.non_existing_attribute
                self.assertIsInstance(a, Pyro4.core._RemoteMethod)
            # set attributes
            with Pyro4.core.Proxy(self.objectUri) as p:
                p.some_weird_attribute = 42
                self.assertEqual(42, p.some_weird_attribute)
        finally:
            config.METADATA = True

    def testProxyAttrsMetadataOn(self):
        try:
            config.METADATA = True
            # read attributes
            with Pyro4.core.Proxy(self.objectUri) as p:
                # unconnected proxy still has empty metadata.
                # but, as soon as an attribute is used, the metadata is obtained (as long as METADATA is true)
                a = p.value
                self.assertEqual(12345, a)
                a = p.multiply
                self.assertIsInstance(a, Pyro4.core._RemoteMethod)  # multiply is still a regular method
                with self.assertRaises(AttributeError):
                    _ = p.non_existing_attribute
            # set attributes, should also trigger getting metadata
            with Pyro4.core.Proxy(self.objectUri) as p:
                p.value = 42
                self.assertEqual(42, p.value)
                self.assertTrue("value" in p._pyroAttrs)
        finally:
            config.METADATA = True

    def testProxyAnnotationsOldApi(self):
        class CustomAnnotationsProxy(Pyro4.core.Proxy):
            def __init__(self, uri, response):
                self.__dict__["response"] = response
                super(CustomAnnotationsProxy, self).__init__(uri)
            def _pyroAnnotations(self):
                ann = {"XYZZ": b"some data sent by old annotations api"}
                self.__dict__["response"]["annotations_sent"] = ann
                return ann
            def _pyroResponseAnnotations(self, annotations, msgtype):
                self.__dict__["response"]["annotations"] = annotations
                self.__dict__["response"]["msgtype"] = msgtype
        response = {}
        corr_id = Pyro4.core.current_context.correlation_id = uuid.uuid4()
        with CustomAnnotationsProxy(self.objectUri, response) as p:
            p.ping()
        self.assertDictEqual({"XYZZ": b"some data sent by old annotations api"}, p.__dict__["response"]["annotations_sent"])
        self.assertEqual(Pyro4.message.MSG_RESULT, p.__dict__["response"]["msgtype"])
        self.assertDictEqual({"CORR": corr_id.bytes}, p.__dict__["response"]["annotations"])

    def testProxyAnnotations(self):
        class AnnotationsProxy(Pyro4.core.Proxy):
            def _pyroAnnotations(self):
                return {"QWER": b"data via old api"}
        with AnnotationsProxy(self.objectUri) as p:
            Pyro4.core.current_context.annotations = {"XYZZ": b"invalid test data"}
            try:
                p.response_annotation()
                self.fail("should fail")
            except ValueError:
                pass
            Pyro4.core.current_context.annotations = {"XYZZ": b"data from proxy via new api"}
            response = p.response_annotation()
            self.assertEqual(b"daemon annotation via new api", Pyro4.core.current_context.response_annotations["ANN2"])
            # check that the daemon received both the old and the new annotation api data:
            daemon_annotations = response["annotations_in_daemon"]
            if sys.version_info < (3, 0) and sys.platform != "cli":
                self.assertEqual("data from proxy via new api", daemon_annotations["XYZZ"])
                self.assertEqual("data via old api", daemon_annotations["QWER"])
            else:
                self.assertEqual(b"data from proxy via new api", serpent.tobytes(daemon_annotations["XYZZ"]))
                self.assertEqual(b"data via old api", serpent.tobytes(daemon_annotations["QWER"]))

    def testExposedNotRequired(self):
        try:
            old_require = config.REQUIRE_EXPOSE
            config.REQUIRE_EXPOSE = False
            with self.daemon.proxyFor("unexposed") as p:
                self.assertEqual({"unexposed", "getName"}, p._pyroMethods)
                self.assertEqual("hello", p.getName())
                self.assertEqual("you should not see this", p.unexposed())   # you *should* see it when REQUIRE_EXPOSE is False :)
        finally:
            config.REQUIRE_EXPOSE = old_require

    def testExposedRequired(self):
        try:
            old_require = config.REQUIRE_EXPOSE
            config.REQUIRE_EXPOSE = True
            with self.daemon.proxyFor("unexposed") as p:
                self.assertEqual({"getName"}, p._pyroMethods)
                self.assertEqual("hello", p.getName())
                with self.assertRaises(AttributeError) as e:
                    p.unexposed()
                expected_msg = "remote object '%s' has no exposed attribute or method 'unexposed'" % p._pyroUri
                self.assertEqual(expected_msg, str(e.exception))
                with self.assertRaises(AttributeError) as e:
                    p.unexposed_set = 999
                expected_msg = "remote object '%s' has no exposed attribute 'unexposed_set'" % p._pyroUri
                self.assertEqual(expected_msg, str(e.exception))
        finally:
            config.REQUIRE_EXPOSE = old_require

    def testProperties(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            _ = p.value
            # metadata should be loaded now
            self.assertEqual({"value", "dictionary"}, p._pyroAttrs)
            with self.assertRaises(AttributeError):
                _ = p.something
            with self.assertRaises(AttributeError):
                _ = p._dictionary
            with self.assertRaises(AttributeError):
                _ = p._value
            self.assertEqual(12345, p.value)
            self.assertEqual({"number": 42}, p.dictionary)

    def testHasAttr(self):
        try:
            config.METADATA = False
            with Pyro4.core.Proxy(self.objectUri) as p:
                # with metadata off, all attributes are considered valid (and return a RemoteMethod object)
                self.assertTrue(hasattr(p, "multiply"))
                self.assertTrue(hasattr(p, "oneway_multiply"))
                self.assertTrue(hasattr(p, "value"))
                self.assertTrue(hasattr(p, "_value"))
                self.assertTrue(hasattr(p, "_dictionary"))
                self.assertTrue(hasattr(p, "non_existing_attribute"))
            config.METADATA = True
            with Pyro4.core.Proxy(self.objectUri) as p:
                # with metadata on, hasattr actually gives proper results
                self.assertTrue(hasattr(p, "multiply"))
                self.assertTrue(hasattr(p, "oneway_multiply"))
                self.assertTrue(hasattr(p, "value"))
                self.assertFalse(hasattr(p, "_value"))
                self.assertFalse(hasattr(p, "_dictionary"))
                self.assertFalse(hasattr(p, "non_existing_attribute"))
        finally:
            config.METADATA = True

    def testProxyMetadataKnown(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            # unconnected proxies have empty metadata
            self.assertEqual(set(), p._pyroAttrs)
            self.assertEqual(set(), p._pyroMethods)
            self.assertEqual(set(), p._pyroOneway)
            # set some metadata manually, they should be overwritten at connection time
            p._pyroMethods = set("abc")
            p._pyroAttrs = set("xyz")
            p._pyroBind()
            self.assertNotEqual(set("xyz"), p._pyroAttrs)
            self.assertNotEqual(set("abc"), p._pyroMethods)
            self.assertNotEqual(set(), p._pyroOneway)

    def testNonserializableException_other(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            try:
                p.nonserializableException()
                self.fail("should crash")
            except Exception as x:
                self.assertIsInstance(x, Pyro4.errors.PyroError)
                tblines = "\n".join(Pyro4.util.getPyroTraceback())
                self.assertTrue("unsupported serialized class" in tblines)

    def testNonserializableException_pickle(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            config.SERIALIZER = "pickle"
            try:
                p.nonserializableException()
                self.fail("should crash")
            except Exception as x:
                self.assertIsInstance(x, Pyro4.errors.PyroError)
                tblines = "\n".join(Pyro4.util.getPyroTraceback())
                self.assertTrue("PyroError: Error serializing exception" in tblines)
                s1 = "Original exception: <class 'testsupport.NonserializableError'>:"
                s2 = "Original exception: <class 'PyroTests.testsupport.NonserializableError'>:"
                self.assertTrue(s1 in tblines or s2 in tblines)
                self.assertTrue("raise NonserializableError((\"xantippe" in tblines)
            finally:
                config.SERIALIZER = "serpent"

    def testBatchProxy(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            batch = Pyro4.core.batch(p)
            self.assertIsNone(batch.multiply(7, 6))
            self.assertIsNone(batch.divide(999, 3))
            self.assertIsNone(batch.ping())
            self.assertIsNone(batch.divide(999, 0))  # force an exception here
            self.assertIsNone(batch.multiply(3, 4))  # this call should not be performed after the error
            results = batch()
            self.assertEqual(42, next(results))
            self.assertEqual(333, next(results))
            self.assertIsNone(next(results))
            self.assertRaises(ZeroDivisionError, next, results)  # 999//0 should raise this error
            self.assertRaises(StopIteration, next, results)  # no more results should be available after the error

    def testAsyncProxy(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            Pyro4.core.asyncproxy(p)
            p._pyroBind()  # force that any metadata is processed
            begin = time.time()
            result = p.delayAndId(1, 42)
            duration = time.time() - begin
            self.assertTrue(duration < 0.1)
            self.assertFalse(result.ready)
            self.assertFalse(result.wait(0.5))  # not available within 0.5 sec
            self.assertEqual("slept for 42", result.value)
            self.assertTrue(result.ready)
            self.assertTrue(result.wait())

    def testAsyncProxyCallchain(self):
        class FuncHolder(object):
            count = AtomicCounter()

            def function(self, value, increase=1):
                self.count.incr()
                return value + increase

        with Pyro4.core.Proxy(self.objectUri) as p:
            Pyro4.core.asyncproxy(p)
            p._pyroBind()  # force that any metadata is processed
            holder = FuncHolder()
            begin = time.time()
            result = p.multiply(2, 3)
            result.then(holder.function, increase=10) \
                .then(holder.function, increase=5) \
                .then(holder.function)
            duration = time.time() - begin
            self.assertTrue(duration < 0.1)
            value = result.value
            self.assertTrue(result.ready)
            self.assertEqual(22, value)
            self.assertEqual(3, holder.count.value)

    def testBatchOneway(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            batch = Pyro4.core.batch(p)
            self.assertIsNone(batch.multiply(7, 6))
            self.assertIsNone(batch.delay(1))  # a delay shouldn't matter with oneway
            self.assertIsNone(batch.multiply(3, 4))
            begin = time.time()
            results = batch(oneway=True)
            duration = time.time() - begin
            self.assertTrue(duration < 0.1, "oneway batch with delay should return almost immediately")
            self.assertIsNone(results)

    def testBatchAsync(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            batch = Pyro4.core.batch(p)
            self.assertIsNone(batch.multiply(7, 6))
            self.assertIsNone(batch.delay(1))  # a delay shouldn't matter with asynchronous
            self.assertIsNone(batch.multiply(3, 4))
            begin = time.time()
            asyncresult = batch(asynchronous=True)
            duration = time.time() - begin
            self.assertTrue(duration < 0.1, "async batch with delay should return almost immediately")
            results = asyncresult.value
            self.assertEqual(42, next(results))
            self.assertEqual("slept 1 seconds", next(results))
            self.assertEqual(12, next(results))
            self.assertRaises(StopIteration, next, results)  # no more results should be available

    def testBatchAsyncCallchain(self):
        class FuncHolder(object):
            count = AtomicCounter()

            def function(self, values):
                result = [value + 1 for value in values]
                self.count.incr()
                return result

        with Pyro4.core.Proxy(self.objectUri) as p:
            batch = Pyro4.core.batch(p)
            self.assertIsNone(batch.multiply(7, 6))
            self.assertIsNone(batch.multiply(3, 4))
            result = batch(asynchronous=True)
            holder = FuncHolder()
            result.then(holder.function).then(holder.function)
            value = result.value
            self.assertTrue(result.ready)
            self.assertEqual([44, 14], value)
            self.assertEqual(2, holder.count.value)

    def testPyroTracebackNormal(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            try:
                p.divide(999, 0)  # force error here
                self.fail("expected error")
            except ZeroDivisionError:
                # going to check if the magic pyro traceback attribute is available for batch methods too
                tb = "".join(Pyro4.util.getPyroTraceback())
                self.assertIn("Remote traceback:", tb)  # validate if remote tb is present
                self.assertIn("ZeroDivisionError", tb)  # the error
                self.assertIn("return x // y", tb)  # the statement

    def testPyroTracebackBatch(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            batch = Pyro4.core.batch(p)
            self.assertIsNone(batch.divide(999, 0))  # force an exception here
            results = batch()
            try:
                next(results)
                self.fail("expected error")
            except ZeroDivisionError:
                # going to check if the magic pyro traceback attribute is available for batch methods too
                tb = "".join(Pyro4.util.getPyroTraceback())
                self.assertIn("Remote traceback:", tb)  # validate if remote tb is present
                self.assertIn("ZeroDivisionError", tb)  # the error
                self.assertIn("return x // y", tb)  # the statement
            self.assertRaises(StopIteration, next, results)  # no more results should be available after the error

    def testAutoProxy(self):
        obj = ServerTestObject()
        config.SERIALIZER = "pickle"
        try:
            with Pyro4.core.Proxy(self.objectUri) as p:
                config.AUTOPROXY = False  # make sure autoproxy is disabled
                result = p.echo(obj)
                self.assertIsInstance(result, ServerTestObject)
                self.daemon.register(obj)
                result = p.echo(obj)
                self.assertIsInstance(result, ServerTestObject, "with autoproxy off the object should be an instance of the class")
                self.daemon.unregister(obj)
                result = p.echo(obj)
                self.assertIsInstance(result, ServerTestObject, "serialized object must still be normal object")
                self.daemon.register(ServerTestObject)
                new_result = result.new_test_object()
                self.assertIsInstance(new_result, ServerTestObject, "serialized pyro object must be a normal object")
                self.daemon.unregister(ServerTestObject)
                config.AUTOPROXY = True  # make sure autoproxying is enabled
                result = p.echo(obj)
                self.assertIsInstance(result, ServerTestObject, "non-pyro object must be returned as normal class")
                self.daemon.register(obj)
                result = p.echo(obj)
                self.assertIsInstance(result, Pyro4.core.Proxy, "serialized pyro object must be a proxy")
                self.daemon.register(ServerTestObject)
                new_result = result.new_test_object()
                self.assertIsInstance(new_result, Pyro4.core.Proxy, "serialized pyro object must be a proxy")
                self.daemon.unregister(ServerTestObject)
                self.daemon.unregister(obj)
                result = p.echo(obj)
                self.assertIsInstance(result, ServerTestObject, "unregistered pyro object must be normal class again")
                # note: the custom serializer may still be active but it should be smart enough to see
                # that the object is no longer a pyro object, and therefore, no proxy should be created.
        finally:
            config.AUTOPROXY = True
            config.SERIALIZER = "serpent"

    def testConnectOnce(self):
        with Pyro4.core.Proxy(self.objectUri) as proxy:
            self.assertTrue(proxy._pyroBind(), "first bind should always connect")
            self.assertFalse(proxy._pyroBind(), "second bind should not connect again")

    def testConnectingThreads(self):
        class ConnectingThread(threading.Thread):
            new_connections = AtomicCounter()

            def __init__(self, proxy, event):
                threading.Thread.__init__(self)
                self.proxy = proxy
                self.event = event
                self.setDaemon(True)
                self.new_connections.reset()

            def run(self):
                self.event.wait()
                if self.proxy._pyroBind():
                    ConnectingThread.new_connections.incr()  # 1 more new connection done

        with Pyro4.core.Proxy(self.objectUri) as proxy:
            event = threading.Event()
            threads = [ConnectingThread(proxy, event) for _ in range(20)]
            for t in threads:
                t.start()
            event.set()
            for t in threads:
                t.join()
            self.assertEqual(1, ConnectingThread.new_connections.value)  # proxy shared among threads must still have only 1 connect done

    def testMaxMsgSize(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            bigobject = [42] * 1000
            result = p.echo(bigobject)
            self.assertEqual(result, bigobject)
            config.MAX_MESSAGE_SIZE = 999
            try:
                _ = p.echo(bigobject)
                self.fail("should fail with ProtocolError msg too large")
            except Pyro4.errors.ProtocolError:
                pass
            config.MAX_MESSAGE_SIZE = 0

    def testIterator(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            iterator = p.iterator()
            self.assertIsInstance(iterator, Pyro4.core._StreamResultIterator)
            self.assertEqual("one", next(iterator))
            self.assertEqual("two", next(iterator))
            self.assertEqual("three", next(iterator))
            with self.assertRaises(StopIteration):
                next(iterator)
            iterator.close()

    def testGenerator(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            generator = p.generator()
            self.assertIsInstance(generator, Pyro4.core._StreamResultIterator)
            self.assertEqual("one", next(generator))
            self.assertEqual("two", next(generator))
            self.assertEqual("three", next(generator))
            self.assertEqual("four", next(generator))
            self.assertEqual("five", next(generator))
            with self.assertRaises(StopIteration):
                next(generator)
            with self.assertRaises(StopIteration):
                next(generator)
            generator.close()
            generator = p.generator()
            [v for v in generator]
            with self.assertRaises(StopIteration):
                next(generator)
            generator.close()

    def testCleanup(self):
        p1 = Pyro4.core.Proxy(self.objectUri)
        p2 = Pyro4.core.Proxy(self.objectUri)
        p3 = Pyro4.core.Proxy(self.objectUri)
        p1.echo(42)
        p2.echo(42)
        p3.echo(42)
        # we have several active connections still up, see if we can cleanly shutdown the daemon
        # (it should interrupt the worker's socket connections)
        time.sleep(0.1)
        self.daemon.shutdown()
        self.daemon = None
        p1._pyroRelease()
        p2._pyroRelease()
        p3._pyroRelease()

    def testSerializedBlob(self):
        sb = Pyro4.core.SerializedBlob("blobname", [1, 2, 3])
        self.assertEqual("blobname", sb.info)
        self.assertEqual([1, 2, 3],  sb.deserialized())

    def testSerializedBlobMessage(self):
        serializer = Pyro4.util.get_serializer("serpent")
        data, _ = serializer.serializeCall("object", "method", ([1, 2, 3],), False)
        msg = Pyro4.message.Message(Pyro4.message.MSG_INVOKE, data, serializer.serializer_id, 0, 42)
        sb = Pyro4.core.SerializedBlob("blobname", msg, is_blob=True)
        self.assertEqual("blobname", sb.info)
        self.assertEqual(([1, 2, 3],), sb.deserialized())

    def testProxySerializedBlobArg(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            blobinfo, blobdata = p.blob(Pyro4.core.SerializedBlob("blobname", [1, 2, 3]))
            self.assertEqual("blobname", blobinfo)
            self.assertEqual([1, 2, 3], blobdata)

    def testResourceFreeing(self):
        rsvc = ResourceService()
        uri = self.daemon.register(rsvc)
        with Pyro4.core.Proxy(uri) as p:
            p.allocate("r1")
            p.allocate("r2")
            resources = {r.name: r for r in rsvc.resources}
            p.free("r1")
            rsc = p.list()
            self.assertEqual(["r2"], rsc)
            self.assertTrue(resources["r1"].close_called)
            self.assertFalse(resources["r2"].close_called)
        time.sleep(0.02)
        self.assertTrue(resources["r1"].close_called)
        self.assertTrue(resources["r2"].close_called)
        with Pyro4.core.Proxy(uri) as p:
            rsc = p.list()
            self.assertEqual([], rsc, "r2 must now be freed due to connection loss earlier")


class ServerTestsThreadNoTimeout(unittest.TestCase):
    SERVERTYPE = "thread"
    COMMTIMEOUT = None

    def setUp(self):
        config.LOGWIRE = True
        config.POLLTIMEOUT = 0.1
        config.SERVERTYPE = self.SERVERTYPE
        config.COMMTIMEOUT = self.COMMTIMEOUT
        config.SERIALIZERS_ACCEPTED.add("pickle")
        self.daemon = Pyro4.core.Daemon(port=0)
        obj = ServerTestObject()
        uri = self.daemon.register(obj, "something")
        self.objectUri = uri
        self.daemonthread = DaemonLoopThread(self.daemon)
        self.daemonthread.start()
        self.daemonthread.running.wait()
        time.sleep(0.05)

    def tearDown(self):
        time.sleep(0.05)
        self.daemon.shutdown()
        self.daemonthread.join()
        config.SERVERTYPE = "thread"
        config.COMMTIMEOUT = None
        config.SERIALIZERS_ACCEPTED.discard("pickle")

    def testConnectionStuff(self):
        p1 = Pyro4.core.Proxy(self.objectUri)
        p2 = Pyro4.core.Proxy(self.objectUri)
        self.assertIsNone(p1._pyroConnection)
        self.assertIsNone(p2._pyroConnection)
        p1.ping()
        p2.ping()
        _ = p1.multiply(11, 5)
        _ = p2.multiply(11, 5)
        self.assertIsNotNone(p1._pyroConnection)
        self.assertIsNotNone(p2._pyroConnection)
        p1._pyroRelease()
        p1._pyroRelease()
        p2._pyroRelease()
        p2._pyroRelease()
        self.assertIsNone(p1._pyroConnection)
        self.assertIsNone(p2._pyroConnection)
        p1._pyroBind()
        _ = p1.multiply(11, 5)
        _ = p2.multiply(11, 5)
        self.assertIsNotNone(p1._pyroConnection)
        self.assertIsNotNone(p2._pyroConnection)
        self.assertEqual("PYRO", p1._pyroUri.protocol)
        self.assertEqual("PYRO", p2._pyroUri.protocol)
        p1._pyroRelease()
        p2._pyroRelease()

    def testReconnectAndCompression(self):
        # try reconnects
        with Pyro4.core.Proxy(self.objectUri) as p:
            self.assertIsNone(p._pyroConnection)
            p._pyroReconnect(tries=100)
            self.assertIsNotNone(p._pyroConnection)
        self.assertIsNone(p._pyroConnection)
        # test compression:
        try:
            with Pyro4.core.Proxy(self.objectUri) as p:
                config.COMPRESSION = True
                self.assertEqual(55, p.multiply(5, 11))
                self.assertEqual("*" * 1000, p.multiply("*" * 500, 2))
        finally:
            config.COMPRESSION = False

    def testOnewayMetaOn(self):
        config.METADATA = True
        with Pyro4.core.Proxy(self.objectUri) as p:
            self.assertEqual(set(), p._pyroOneway)  # when not bound, no meta info exchange has been done
            p._pyroBind()
            self.assertIn("oneway_multiply", p._pyroOneway)  # after binding, meta info has been processed
            self.assertEqual(55, p.multiply(5, 11))  # not tagged as @Pyro4.oneway
            self.assertIsNone(p.oneway_multiply(5, 11))  # tagged as @Pyro4.oneway
            p._pyroOneway = set()
            self.assertEqual(55, p.multiply(5, 11))
            self.assertEqual(55, p.oneway_multiply(5, 11))
            # check nonexisting method behavoir for oneway methods
            with self.assertRaises(AttributeError):
                p.nonexisting_method()
            p._pyroOneway.add("nonexisting_method")
            # now it should still fail because of metadata telling Pyro what methods actually exist
            with self.assertRaises(AttributeError):
                p.nonexisting_method()

    def testOnewayMetaOff(self):
        config.METADATA = False
        with Pyro4.core.Proxy(self.objectUri) as p:
            self.assertEqual(set(), p._pyroOneway)  # when not bound, no meta info exchange has been done
            p._pyroBind()
            self.assertEqual(set(), p._pyroOneway)  # after binding, no meta info exchange has been done because disabled
            self.assertEqual(55, p.multiply(5, 11))
            self.assertEqual(55, p.oneway_multiply(5, 11))
            # check nonexisting method behavoir for oneway methods
            with self.assertRaises(AttributeError):
                p.nonexisting_method()
            p._pyroOneway.add("nonexisting_method")
            # now it shouldn't fail because of oneway semantics (!) (and becaue there's no metadata to tell Pyro that the method doesn't exist)
            p.nonexisting_method()
        config.METADATA = True

    def testOnewayWithProxySubclass(self):
        config.METADATA = False

        class ProxyWithOneway(Pyro4.core.Proxy):
            def __init__(self, arg):
                super(ProxyWithOneway, self).__init__(arg)
                self._pyroOneway = {"oneway_multiply", "multiply"}

        with ProxyWithOneway(self.objectUri) as p:
            self.assertIsNone(p.oneway_multiply(5, 11))
            self.assertIsNone(p.multiply(5, 11))
            p._pyroOneway = set()
            self.assertEqual(55, p.oneway_multiply(5, 11))
            self.assertEqual(55, p.multiply(5, 11))
        config.METADATA = True

    def testOnewayDelayed(self):
        try:
            with Pyro4.core.Proxy(self.objectUri) as p:
                p.ping()
                config.ONEWAY_THREADED = True  # the default
                now = time.time()
                p.oneway_delay(1)  # oneway so we should continue right away
                time.sleep(0.01)
                self.assertTrue(time.time() - now < 0.2, "delay should be running as oneway")
                now = time.time()
                self.assertEqual(55, p.multiply(5, 11), "expected a normal result from a non-oneway call")
                self.assertTrue(time.time() - now < 0.2, "delay should be running in its own thread")
                # make oneway calls run in the server thread
                # we can change the config here and the server will pick it up on the fly
                config.ONEWAY_THREADED = False
                now = time.time()
                p.oneway_delay(1)  # oneway so we should continue right away
                time.sleep(0.01)
                self.assertTrue(time.time() - now < 0.2, "delay should be running as oneway")
                now = time.time()
                self.assertEqual(55, p.multiply(5, 11), "expected a normal result from a non-oneway call")
                self.assertFalse(time.time() - now < 0.2, "delay should be running in the server thread")
        finally:
            config.ONEWAY_THREADED = True  # back to normal

    def testSerializeConnected(self):
        # online serialization tests
        ser = Pyro4.util.get_serializer(config.SERIALIZER)
        proxy = Pyro4.core.Proxy(self.objectUri)
        proxy._pyroBind()
        self.assertIsNotNone(proxy._pyroConnection)
        p, _ = ser.serializeData(proxy)
        proxy2 = ser.deserializeData(p)
        self.assertIsNone(proxy2._pyroConnection)
        self.assertIsNotNone(proxy._pyroConnection)
        self.assertEqual(proxy2._pyroUri, proxy._pyroUri)
        proxy2._pyroBind()
        self.assertIsNotNone(proxy2._pyroConnection)
        self.assertIsNot(proxy2._pyroConnection, proxy._pyroConnection)
        proxy._pyroRelease()
        proxy2._pyroRelease()
        self.assertIsNone(proxy._pyroConnection)
        self.assertIsNone(proxy2._pyroConnection)
        proxy.ping()
        proxy2.ping()
        # try copying a connected proxy
        import copy
        proxy3 = copy.copy(proxy)
        self.assertIsNone(proxy3._pyroConnection)
        self.assertIsNotNone(proxy._pyroConnection)
        self.assertEqual(proxy3._pyroUri, proxy._pyroUri)
        self.assertIsNot(proxy3._pyroUri, proxy._pyroUri)
        proxy._pyroRelease()
        proxy2._pyroRelease()
        proxy3._pyroRelease()

    def testException(self):
        with Pyro4.core.Proxy(self.objectUri) as p:
            try:
                p.divide(1, 0)
            except:
                et, ev, tb = sys.exc_info()
                self.assertEqual(ZeroDivisionError, et)
                pyrotb = "".join(Pyro4.util.getPyroTraceback(et, ev, tb))
                self.assertIn("Remote traceback", pyrotb)
                self.assertIn("ZeroDivisionError", pyrotb)
                del tb

    def testTimeoutCall(self):
        config.COMMTIMEOUT = None
        with Pyro4.core.Proxy(self.objectUri) as p:
            p.ping()
            start = time.time()
            p.delay(0.5)
            duration = time.time() - start
            self.assertTrue(0.4 < duration < 0.6)
            p._pyroTimeout = 0.1
            start = time.time()
            self.assertRaises(Pyro4.errors.TimeoutError, p.delay, 1)
            duration = time.time() - start
            if sys.platform != "cli":
                self.assertLess(duration, 0.3)
            else:
                # ironpython's time is weird
                self.assertTrue(0.0 < duration < 0.7)

    def testTimeoutConnect(self):
        # set up a unresponsive daemon
        with Pyro4.core.Daemon(port=0) as d:
            time.sleep(0.5)
            obj = ServerTestObject()
            uri = d.register(obj)
            # we're not going to start the daemon's event loop
            p = Pyro4.core.Proxy(uri)
            p._pyroTimeout = 0.2
            start = time.time()
            with self.assertRaises(Pyro4.errors.TimeoutError) as e:
                p.ping()
            self.assertEqual("receiving: timeout", str(e.exception))

    def testProxySharing(self):
        class SharedProxyThread(threading.Thread):
            def __init__(self, proxy):
                super(SharedProxyThread, self).__init__()
                self.proxy = proxy
                self.terminate = False
                self.error = True
                self.setDaemon(True)

            def run(self):
                try:
                    while not self.terminate:
                        reply = self.proxy.multiply(5, 11)
                        assert reply == 55
                        time.sleep(0.001)
                    self.error = False
                except:
                    print("Something went wrong in the thread (SharedProxyThread):")
                    print("".join(Pyro4.util.getPyroTraceback()))

        with Pyro4.core.Proxy(self.objectUri) as p:
            threads = []
            for i in range(5):
                t = SharedProxyThread(p)
                threads.append(t)
                t.start()
            time.sleep(1)
            for t in threads:
                t.terminate = True
                t.join()
            for t in threads:
                self.assertFalse(t.error, "all threads should report no errors")

    def testServerConnections(self):
        # check if the server allows to grow the number of connections
        proxies = [Pyro4.core.Proxy(self.objectUri) for _ in range(10)]
        try:
            for p in proxies:
                p._pyroTimeout = 0.5
                p._pyroBind()
            for p in proxies:
                p.ping()
        finally:
            for p in proxies:
                p._pyroRelease()

    def testServerParallelism(self):
        class ClientThread(threading.Thread):
            def __init__(self, uri, name):
                super(ClientThread, self).__init__()
                self.setDaemon(True)
                self.proxy = Pyro4.core.Proxy(uri)
                self.name = name
                self.error = True
                self.proxy._pyroTimeout = 5.0
                self.proxy._pyroBind()

            def run(self):
                try:
                    reply = self.proxy.delayAndId(0.5, self.name)
                    assert reply == "slept for " + self.name
                    self.error = False
                finally:
                    self.proxy._pyroRelease()

        threads = []
        start = time.time()
        try:
            for i in range(6):
                t = ClientThread(self.objectUri, "t%d" % i)
                threads.append(t)
        except:
            # some exception (probably timeout) while creating clients
            # try to clean up some connections first
            for t in threads:
                t.proxy._pyroRelease()
            raise  # re-raise the exception
        for t in threads:
            t.start()
        for t in threads:
            t.join()
            self.assertFalse(t.error, "all threads should report no errors")
        del threads
        duration = time.time() - start
        if config.SERVERTYPE == "multiplex":
            # multiplex based server doesn't execute calls in parallel,
            # so 6 threads times 0.5 seconds =~ 3 seconds
            self.assertTrue(2.5 < duration < 3.5)
        else:
            # thread based server does execute calls in parallel,
            # so 6 threads taking 0.5 seconds =~ 0.5 seconds passed
            self.assertTrue(0.4 < duration < 0.9)

    def testGeneratorProxyClose(self):
        p = Pyro4.core.Proxy(self.objectUri)
        generator = p.generator()
        p._pyroRelease()
        with self.assertRaises(Pyro4.errors.ConnectionClosedError):
            next(generator)

    def testGeneratorLinger(self):
        orig_linger = config.ITER_STREAM_LINGER
        orig_commt = config.COMMTIMEOUT
        orig_pollt = config.POLLTIMEOUT
        try:
            config.ITER_STREAM_LINGER = 0.5
            config.COMMTIMEOUT = 0.2
            config.POLLTIMEOUT = 0.2
            p = Pyro4.core.Proxy(self.objectUri)
            generator = p.generator()
            self.assertEqual("one", next(generator))
            p._pyroRelease()
            with self.assertRaises(Pyro4.errors.ConnectionClosedError):
                next(generator)
            p._pyroReconnect()
            self.assertEqual("two", next(generator), "generator should resume after reconnect")
            # check that after the linger time passes, the generator *is* gone
            p._pyroRelease()
            time.sleep(2)
            p._pyroReconnect()
            with self.assertRaises(Pyro4.errors.PyroError):  # should not be resumable anymore
                next(generator)
        finally:
            config.ITER_STREAM_LINGER = orig_linger
            config.COMMTIMEOUT = orig_commt
            config.POLLTIMEOUT = orig_pollt

    def testGeneratorNoLinger(self):
        orig_linger = config.ITER_STREAM_LINGER
        try:
            p = Pyro4.core.Proxy(self.objectUri)
            config.ITER_STREAM_LINGER = 0  # disable linger
            generator = p.generator()
            self.assertEqual("one", next(generator))
            p._pyroRelease()
            time.sleep(0.2)
            with self.assertRaises(Pyro4.errors.ConnectionClosedError):
                next(generator)
            p._pyroReconnect()
            with self.assertRaises(Pyro4.errors.PyroError):  # should not be resumable after reconnect
                next(generator)
            generator.close()
        finally:
            config.ITER_STREAM_LINGER = orig_linger


class ServerTestsMultiplexNoTimeout(ServerTestsThreadNoTimeout):
    SERVERTYPE = "multiplex"
    COMMTIMEOUT = None

    def testProxySharing(self):
        pass

    def testException(self):
        pass


if __name__ == "__main__":
    # import sys;sys.argv = ['', 'Test.testName']
    unittest.main()