"""
Tests for the data serializer.

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

import array
import sys
import collections
import copy
import pprint
import pickle
import base64
import unittest
import math
import uuid
import Pyro4.util
import Pyro4.errors
import Pyro4.core
import Pyro4.futures
from Pyro4.configuration import config
from testsupport import *


class SerializeTests_pickle(unittest.TestCase):
    SERIALIZER = "pickle"

    def setUp(self):
        self.previous_serializer = config.SERIALIZER
        config.SERIALIZER = self.SERIALIZER
        self.serializer = Pyro4.util.get_serializer(config.SERIALIZER)
        config.REQUIRE_EXPOSE = True

    def tearDown(self):
        config.SERIALIZER = self.previous_serializer

    def testSerItself(self):
        s = Pyro4.util.get_serializer(config.SERIALIZER)
        p, _ = self.serializer.serializeData(s)
        s2 = self.serializer.deserializeData(p)
        self.assertEqual(s, s2)
        self.assertTrue(s == s2)
        self.assertFalse(s != s2)

    def testSerUnicode(self):
        data = unicode("x")
        self.serializer.serializeData(data)
        self.serializer.serializeCall(data, unicode("method"), [], {})

    def testSerCompression(self):
        d1, c1 = self.serializer.serializeData("small data", compress=True)
        d2, c2 = self.serializer.serializeData("small data", compress=False)
        self.assertFalse(c1)
        self.assertEqual(d1, d2)
        bigdata = "x" * 1000
        d1, c1 = self.serializer.serializeData(bigdata, compress=False)
        d2, c2 = self.serializer.serializeData(bigdata, compress=True)
        self.assertFalse(c1)
        self.assertTrue(c2)
        self.assertTrue(len(d2) < len(d1))
        self.assertEqual(bigdata, self.serializer.deserializeData(d1, compressed=False))
        self.assertEqual(bigdata, self.serializer.deserializeData(d2, compressed=True))

    def testSerErrors(self):
        e1 = Pyro4.errors.NamingError(unicode("x"))
        e1._pyroTraceback = ["this is the remote traceback"]
        orig_e = copy.copy(e1)
        e2 = Pyro4.errors.PyroError(unicode("x"))
        e3 = Pyro4.errors.ProtocolError(unicode("x"))
        if sys.platform == "cli":
            Pyro4.util.fixIronPythonExceptionForPickle(e1, True)
        p, _ = self.serializer.serializeData(e1)
        e = self.serializer.deserializeData(p)
        if sys.platform == "cli":
            Pyro4.util.fixIronPythonExceptionForPickle(e, False)
        self.assertIsInstance(e, Pyro4.errors.NamingError)
        self.assertEqual(repr(orig_e), repr(e))
        self.assertEqual(["this is the remote traceback"], e._pyroTraceback, "remote traceback info should be present")
        p, _ = self.serializer.serializeData(e2)
        e = self.serializer.deserializeData(p)
        self.assertIsInstance(e, Pyro4.errors.PyroError)
        self.assertEqual(repr(e2), repr(e))
        p, _ = self.serializer.serializeData(e3)
        e = self.serializer.deserializeData(p)
        self.assertIsInstance(e, Pyro4.errors.ProtocolError)
        self.assertEqual(repr(e3), repr(e))

    def testSerializeExceptionWithAttr(self):
        ex = ZeroDivisionError("test error")
        ex._pyroTraceback = ["test traceback payload"]
        Pyro4.util.fixIronPythonExceptionForPickle(ex, True)  # hack for ironpython
        data, compressed = self.serializer.serializeData(ex)
        ex2 = self.serializer.deserializeData(data, compressed)
        Pyro4.util.fixIronPythonExceptionForPickle(ex2, False)  # hack for ironpython
        self.assertEqual(ZeroDivisionError, type(ex2))
        self.assertTrue(hasattr(ex2, "_pyroTraceback"))
        self.assertEqual(["test traceback payload"], ex2._pyroTraceback)

    def testSerCoreOffline(self):
        uri = Pyro4.core.URI("PYRO:9999@host.com:4444")
        p, _ = self.serializer.serializeData(uri)
        uri2 = self.serializer.deserializeData(p)
        self.assertEqual(uri, uri2)
        self.assertEqual("PYRO", uri2.protocol)
        self.assertEqual("9999", uri2.object)
        self.assertEqual("host.com:4444", uri2.location)
        self.assertEqual(4444, uri2.port)
        self.assertIsNone(uri2.sockname)

        uri = Pyro4.core.URI("PYRO:12345@./u:/tmp/socketname")
        p, _ = self.serializer.serializeData(uri)
        uri2 = self.serializer.deserializeData(p)
        self.assertEqual(uri, uri2)
        self.assertEqual("PYRO", uri2.protocol)
        self.assertEqual("12345", uri2.object)
        self.assertEqual("./u:/tmp/socketname", uri2.location)
        self.assertIsNone(uri2.port)
        self.assertEqual("/tmp/socketname", uri2.sockname)

        proxy = Pyro4.core.Proxy("PYRO:9999@host.com:4444")
        proxy._pyroTimeout = 42
        proxy._pyroMaxRetries = 78
        self.assertIsNone(proxy._pyroConnection)
        p, _ = self.serializer.serializeData(proxy)
        proxy2 = self.serializer.deserializeData(p)
        self.assertIsNone(proxy._pyroConnection)
        self.assertIsNone(proxy2._pyroConnection)
        self.assertEqual(proxy2._pyroUri, proxy._pyroUri)
        self.assertEqual(0, proxy2._pyroTimeout, "must be reset to defaults")
        self.assertEqual(0, proxy2._pyroMaxRetries, "must be reset to defaults")

    def testNested(self):
        if self.SERIALIZER == "marshal":
            self.skipTest("marshal can't serialize custom objects")
        uri1 = Pyro4.core.URI("PYRO:1111@host.com:111")
        uri2 = Pyro4.core.URI("PYRO:2222@host.com:222")
        _ = self.serializer.serializeData(uri1)
        data = [uri1, uri2]
        p, _ = self.serializer.serializeData(data)
        [u1, u2] = self.serializer.deserializeData(p)
        self.assertEqual(uri1, u1)
        self.assertEqual(uri2, u2)

    def testSerDaemonHack(self):
        # This tests the hack that a Daemon should be serializable,
        # but only to support serializing Pyro objects.
        # The serialized form of a Daemon should be empty (and thus, useless)
        with Pyro4.core.Daemon(port=0) as daemon:
            d, _ = self.serializer.serializeData(daemon)
            d2 = self.serializer.deserializeData(d)
            self.assertTrue(len(d2.__dict__) == 0, "deserialized daemon should be empty")
            self.assertTrue("Pyro4.core.Daemon" in repr(d2))
            self.assertTrue("unusable" in repr(d2))
            try:
                config.AUTOPROXY = False
                obj = pprint.PrettyPrinter(stream="dummy", width=42)
                obj.name = "hello"
                daemon.register(obj)
                o, _ = self.serializer.serializeData(obj)
                if self.SERIALIZER in ("pickle", "cloudpickle", "dill"):
                    # only pickle, cloudpickle and dill can deserialize the PrettyPrinter class without the need of explicit deserialization function
                    o2 = self.serializer.deserializeData(o)
                    self.assertEqual("hello", o2.name)
                    self.assertEqual(42, o2._width)
            finally:
                config.AUTOPROXY = True

    def testPyroClasses(self):
        uri = Pyro4.core.URI("PYRO:object@host:4444")
        s, c = self.serializer.serializeData(uri)
        x = self.serializer.deserializeData(s, c)
        self.assertIsInstance(x, Pyro4.core.URI)
        self.assertEqual(uri, x)
        self.assertTrue("Pyro4.core.URI" in repr(uri))
        self.assertEqual("PYRO:object@host:4444", str(uri))
        uri = Pyro4.core.URI("PYRO:12345@./u:/tmp/socketname")
        s, c = self.serializer.serializeData(uri)
        x = self.serializer.deserializeData(s, c)
        self.assertIsInstance(x, Pyro4.core.URI)
        self.assertEqual(uri, x)
        proxy = Pyro4.core.Proxy(uri)
        proxy._pyroAttrs = set("abc")
        proxy._pyroMethods = set("def")
        proxy._pyroOneway = set("ghi")
        proxy._pyroTimeout = 42
        proxy._pyroHmacKey = b"secret"
        proxy._pyroHandshake = "apples"
        proxy._pyroMaxRetries = 78
        proxy._pyroSerializer = "serializer"
        s, c = self.serializer.serializeData(proxy)
        x = self.serializer.deserializeData(s, c)
        self.assertIsInstance(x, Pyro4.core.Proxy)
        self.assertEqual(proxy._pyroUri, x._pyroUri)
        self.assertEqual(set("abc"), x._pyroAttrs)
        self.assertEqual(set("def"), x._pyroMethods)
        self.assertEqual(set("ghi"), x._pyroOneway)
        self.assertEqual(b"secret", x._pyroHmacKey)
        self.assertEqual("apples", x._pyroHandshake)
        self.assertEqual("serializer", x._pyroSerializer)
        self.assertEqual(0, x._pyroTimeout, "must be reset to defaults")
        self.assertEqual(0, x._pyroMaxRetries, "must be reset to defaults")
        self.assertTrue("Pyro4.core.Proxy" in repr(x))
        self.assertTrue("Pyro4.core.Proxy" in str(x))
        daemon = Pyro4.core.Daemon()
        s, c = self.serializer.serializeData(daemon)
        x = self.serializer.deserializeData(s, c)
        self.assertIsInstance(x, Pyro4.core.Daemon)
        self.assertTrue("Pyro4.core.Daemon" in repr(x))
        self.assertTrue("unusable" in repr(x))
        self.assertTrue("Pyro4.core.Daemon" in str(x))
        self.assertTrue("unusable" in str(x))
        wrapper = Pyro4.futures._ExceptionWrapper(ZeroDivisionError("divided by zero"))
        s, c = self.serializer.serializeData(wrapper)
        x = self.serializer.deserializeData(s, c)
        self.assertIsInstance(x, Pyro4.futures._ExceptionWrapper)
        self.assertEqual("divided by zero", str(x.exception))
        self.assertTrue("ExceptionWrapper" in repr(x))
        self.assertTrue("ExceptionWrapper" in str(x))

    def testPyroClassesForDict(self):
        uri = Pyro4.core.URI("PYRO:object@host:4444")
        state = uri.__getstate_for_dict__()
        self.assertEqual(('PYRO', 'object', None, 'host', 4444), state)
        uri2 = Pyro4.core.URI("PYRONAME:xxx")
        uri2.__setstate_from_dict__(state)
        self.assertEqual(uri, uri2)
        proxy = Pyro4.core.Proxy(uri)
        proxy._pyroAttrs = set("abc")
        proxy._pyroMethods = set("def")
        proxy._pyroOneway = set("ghi")
        proxy._pyroTimeout = 42
        proxy._pyroHmacKey = b"secret"
        proxy._pyroHandshake = "apples"
        proxy._pyroMaxRetries = 78
        proxy._pyroSerializer = "serializer"
        state = proxy.__getstate_for_dict__()
        b64_secret = "b64:"+base64.b64encode(b"secret").decode("utf-8")
        self.assertEqual(('PYRO:object@host:4444', tuple(set("ghi")), tuple(set("def")), tuple(set("abc")), 42, b64_secret, "apples", 78, "serializer"), state)
        proxy2 = Pyro4.core.Proxy("PYRONAME:xxx")
        proxy2.__setstate_from_dict__(state)
        self.assertEqual(proxy, proxy2)
        self.assertEqual(proxy._pyroUri, proxy2._pyroUri)
        self.assertEqual(proxy._pyroAttrs, proxy2._pyroAttrs)
        self.assertEqual(proxy._pyroMethods, proxy2._pyroMethods)
        self.assertEqual(proxy._pyroOneway, proxy2._pyroOneway)
        self.assertEqual(proxy._pyroHmacKey, proxy2._pyroHmacKey)
        self.assertEqual(proxy._pyroHandshake, proxy2._pyroHandshake)
        self.assertEqual(proxy._pyroSerializer, proxy2._pyroSerializer)
        self.assertEqual(0, proxy2._pyroTimeout, "must be reset to defaults")
        self.assertEqual(0, proxy2._pyroMaxRetries, "must be reset to defaults")
        daemon = Pyro4.core.Daemon()
        state = daemon.__getstate_for_dict__()
        self.assertEqual(tuple(), state)
        daemon2 = Pyro4.core.Daemon()
        daemon2.__setstate_from_dict__(state)

    def testProxySerializationCompat(self):
        proxy = Pyro4.core.Proxy("PYRO:object@host:4444")
        proxy._pyroSerializer = "serializer"
        pickle_state = proxy.__getstate__()
        self.assertEqual(9, len(pickle_state))
        pickle_state = pickle_state[:8]
        proxy.__setstate__(pickle_state)
        self.assertIsNone(proxy._pyroSerializer)
        proxy._pyroSerializer = "serializer"
        serpent_state = proxy.__getstate_for_dict__()
        self.assertEqual(9, len(serpent_state))
        serpent_state = serpent_state[:8]
        proxy.__setstate_from_dict__(serpent_state)
        self.assertIsNone(proxy._pyroSerializer)

    def testAutoProxyPartlyExposed(self):
        if self.SERIALIZER == "marshal":
            self.skipTest("marshal can't serialize custom objects")
        self.serializer.register_type_replacement(MyThingPartlyExposed, Pyro4.core.pyroObjectToAutoProxy)
        t1 = MyThingPartlyExposed("1")
        t2 = MyThingPartlyExposed("2")
        with Pyro4.core.Daemon() as d:
            d.register(t1, "thingy1")
            d.register(t2, "thingy2")
            data = [t1, ["apple", t2]]
            s, c = self.serializer.serializeData(data)
            data = self.serializer.deserializeData(s, c)
            self.assertEqual("apple", data[1][0])
            p1 = data[0]
            p2 = data[1][1]
            self.assertIsInstance(p1, Pyro4.core.Proxy)
            self.assertIsInstance(p2, Pyro4.core.Proxy)
            self.assertEqual("thingy1", p1._pyroUri.object)
            self.assertEqual("thingy2", p2._pyroUri.object)
            self.assertEqual({"prop1", "readonly_prop1"}, p1._pyroAttrs)
            self.assertEqual({"exposed", "oneway"}, p1._pyroMethods)
            self.assertEqual({'oneway'}, p1._pyroOneway)

    def testAutoProxyFullExposed(self):
        if self.SERIALIZER == "marshal":
            self.skipTest("marshal can't serialize custom objects")
        self.serializer.register_type_replacement(MyThingPartlyExposed, Pyro4.core.pyroObjectToAutoProxy)
        t1 = MyThingFullExposed("1")
        t2 = MyThingFullExposed("2")
        with Pyro4.core.Daemon() as d:
            d.register(t1, "thingy1")
            d.register(t2, "thingy2")
            data = [t1, ["apple", t2]]
            s, c = self.serializer.serializeData(data)
            data = self.serializer.deserializeData(s, c)
            self.assertEqual("apple", data[1][0])
            p1 = data[0]
            p2 = data[1][1]
            self.assertIsInstance(p1, Pyro4.core.Proxy)
            self.assertIsInstance(p2, Pyro4.core.Proxy)
            self.assertEqual("thingy1", p1._pyroUri.object)
            self.assertEqual("thingy2", p2._pyroUri.object)
            self.assertEqual({"prop1", "prop2", "readonly_prop1"}, p1._pyroAttrs)
            self.assertEqual({'classmethod', 'method', 'oneway', 'staticmethod', 'exposed', "__dunder__"}, p1._pyroMethods)
            self.assertEqual({'oneway'}, p1._pyroOneway)

    def testRegisterTypeReplacementSanity(self):
        if self.SERIALIZER == "marshal":
            self.skipTest("marshal can't serialize custom objects")
        self.serializer.register_type_replacement(int, lambda: None)
        with self.assertRaises(ValueError):
            self.serializer.register_type_replacement(type, lambda: None)
        with self.assertRaises(ValueError):
            self.serializer.register_type_replacement(42, lambda: None)

    @unittest.skip("skip tetes on python3.11")
    def testCustomClassFail(self):
        if self.SERIALIZER in ("pickle", "cloudpickle", "dill"):
            self.skipTest("pickle, cloudpickle and dill simply serialize custom classes")
        o = pprint.PrettyPrinter(stream="dummy", width=42)
        s, c = self.serializer.serializeData(o)
        try:
            _ = self.serializer.deserializeData(s, c)
            self.fail("error expected, shouldn't deserialize unknown class")
        except Pyro4.errors.ProtocolError:
            pass

    @unittest.skip("skip tetes on python3.11")
    def testCustomClassOk(self):
        if self.SERIALIZER in ("pickle", "cloudpickle", "dill"):
            self.skipTest("pickle, cloudpickle and dill simply serialize custom classes just fine")
        o = MyThingPartlyExposed("test")
        Pyro4.util.SerializerBase.register_class_to_dict(MyThingPartlyExposed, mything_dict)
        Pyro4.util.SerializerBase.register_dict_to_class("CUSTOM-Mythingymabob", mything_creator)
        s, c = self.serializer.serializeData(o)
        o2 = self.serializer.deserializeData(s, c)
        self.assertIsInstance(o2, MyThingPartlyExposed)
        self.assertEqual("test", o2.name)
        # unregister the deserializer
        Pyro4.util.SerializerBase.unregister_dict_to_class("CUSTOM-Mythingymabob")
        try:
            self.serializer.deserializeData(s, c)
            self.fail("must fail")
        except Pyro4.errors.ProtocolError:
            pass  # ok
        # unregister the serializer
        Pyro4.util.SerializerBase.unregister_class_to_dict(MyThingPartlyExposed)
        s, c = self.serializer.serializeData(o)
        try:
            self.serializer.deserializeData(s, c)
            self.fail("must fail")
        except Pyro4.errors.SerializeError as x:
            msg = str(x)
            self.assertIn(msg, ["unsupported serialized class: testsupport.MyThingPartlyExposed",
                                "unsupported serialized class: PyroTests.testsupport.MyThingPartlyExposed"])

    def testData(self):
        data = [42, "hello"]
        ser, compressed = self.serializer.serializeData(data)
        self.assertFalse(compressed)
        data2 = self.serializer.deserializeData(ser, compressed=False)
        self.assertEqual(data, data2)

    def testUnicodeData(self):
        data = u"euro\u20aclowbytes\u0000\u0001\u007f\u0080\u00ff"
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(data, data2)

    def testUUID(self):
        data = uuid.uuid1()
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        uuid_as_str = str(data)
        self.assertTrue(data2==data or data2==uuid_as_str)

    def testSet(self):
        data = {111, 222, 333}
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(data, data2)

    def testCircular(self):
        data = [42, "hello", Pyro4.core.Proxy("PYRO:dummy@dummy:4444")]
        data.append(data)
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed)
        self.assertIs(data2, data2[3])
        self.assertEqual(42, data2[0])

    def testCallPlain(self):
        ser, compressed = self.serializer.serializeCall("object", "method", ("vargs1", "vargs2"), {"kwargs": 999})
        self.assertFalse(compressed)
        obj, method, vargs, kwargs = self.serializer.deserializeCall(ser, compressed=False)
        self.assertEqual("object", obj)
        self.assertEqual("method", method)
        self.assertTrue(len(vargs) == 2)
        self.assertTrue(vargs[0] == "vargs1")
        self.assertTrue(vargs[1] == "vargs2")
        self.assertDictEqual({"kwargs": 999}, kwargs)

    def testCallPyroObjAsArg(self):
        if self.SERIALIZER == "marshal":
            self.skipTest("marshal can't serialize custom objects")
        uri = Pyro4.core.URI("PYRO:555@localhost:80")
        ser, compressed = self.serializer.serializeCall("object", "method", [uri], {"thing": uri})
        self.assertFalse(compressed)
        obj, method, vargs, kwargs = self.serializer.deserializeCall(ser, compressed=False)
        self.assertEqual("object", obj)
        self.assertEqual("method", method)
        self.assertEqual([uri], vargs)
        self.assertEqual({"thing": uri}, kwargs)

    def testCallCustomObjAsArg(self):
        if self.SERIALIZER == "marshal":
            self.skipTest("marshal can't serialize custom objects")
        e = ZeroDivisionError("hello")
        ser, compressed = self.serializer.serializeCall("object", "method", [e], {"thing": e})
        self.assertFalse(compressed)
        obj, method, vargs, kwargs = self.serializer.deserializeCall(ser, compressed=False)
        self.assertEqual("object", obj)
        self.assertEqual("method", method)
        self.assertIsInstance(vargs, list)
        self.assertIsInstance(vargs[0], ZeroDivisionError)
        self.assertEqual("hello", str(vargs[0]))
        self.assertIsInstance(kwargs["thing"], ZeroDivisionError)
        self.assertEqual("hello", str(kwargs["thing"]))

    def testSerializeException(self):
        e = ZeroDivisionError()
        d, c = self.serializer.serializeData(e)
        e2 = self.serializer.deserializeData(d, c)
        self.assertIsInstance(e2, ZeroDivisionError)
        self.assertEqual("", str(e2))
        e = ZeroDivisionError("hello")
        d, c = self.serializer.serializeData(e)
        e2 = self.serializer.deserializeData(d, c)
        self.assertIsInstance(e2, ZeroDivisionError)
        self.assertEqual("hello", str(e2))
        e = ZeroDivisionError("hello", 42)
        d, c = self.serializer.serializeData(e)
        e2 = self.serializer.deserializeData(d, c)
        self.assertIsInstance(e2, ZeroDivisionError)
        self.assertIn(str(e2), ("('hello', 42)", "(u'hello', 42)"))
        e.custom_attribute = 999
        if sys.platform == "cli":
            Pyro4.util.fixIronPythonExceptionForPickle(e, True)
        ser, compressed = self.serializer.serializeData(e)
        e2 = self.serializer.deserializeData(ser, compressed)
        if sys.platform == "cli":
            Pyro4.util.fixIronPythonExceptionForPickle(e2, False)
        self.assertIsInstance(e2, ZeroDivisionError)
        self.assertIn(str(e2), ("('hello', 42)", "(u'hello', 42)"))
        self.assertEqual(999, e2.custom_attribute)

    def testSerializeSpecialException(self):
        self.assertIn("GeneratorExit", Pyro4.util.all_exceptions)
        e = GeneratorExit()
        d, c = self.serializer.serializeData(e)
        e2 = self.serializer.deserializeData(d, c)
        self.assertIsInstance(e2, GeneratorExit)

    def testRecreateClasses(self):
        self.assertEqual([1, 2, 3], self.serializer.recreate_classes([1, 2, 3]))
        d = {"__class__": "invalid"}
        try:
            self.serializer.recreate_classes(d)
            self.fail("error expected")
        except Pyro4.errors.ProtocolError:
            pass  # ok
        d = {"__class__": "Pyro4.core.URI", "state": ['PYRO', '555', None, 'localhost', 80]}
        uri = self.serializer.recreate_classes(d)
        self.assertEqual(Pyro4.core.URI("PYRO:555@localhost:80"), uri)
        number, uri = self.serializer.recreate_classes([1, {"uri": d}])
        self.assertEqual(1, number)
        self.assertEqual(Pyro4.core.URI("PYRO:555@localhost:80"), uri["uri"])

    def testProtocolVersion(self):
        self.assertGreaterEqual(config.PICKLE_PROTOCOL_VERSION, 2)
        self.assertEqual(pickle.HIGHEST_PROTOCOL, config.PICKLE_PROTOCOL_VERSION)

    def testUriSerializationWithoutSlots(self):
        orig_protocol = config.PICKLE_PROTOCOL_VERSION
        config.PICKLE_PROTOCOL_VERSION = 2
        try:
            u = Pyro4.core.URI("PYRO:obj@localhost:1234")
            d, compr = self.serializer.serializeData(u)
            self.assertFalse(compr)
            import pickletools
            d = pickletools.optimize(d)
            result1 = b'\x80\x02cPyro4.core\nURI\n)\x81(U\x04PYROU\x03objNU\tlocalhostM\xd2\x04tb.'
            result2 = b'\x80\x02cPyro4.core\nURI\n)\x81(X\x04\x00\x00\x00PYROX\x03\x00\x00\x00objNX\t\x00\x00\x00localhostM\xd2\x04tb.'
            self.assertTrue(d in (result1, result2))
        finally:
            config.PICKLE_PROTOCOL_VERSION = orig_protocol

    def testFloatPrecision(self):
        f1 = 1482514078.54635912345
        f2 = 9876543212345.12345678987654321
        f3 = 11223344.556677889988776655e33
        floats = [f1, f2, f3]
        d, compr = self.serializer.serializeData(floats)
        v = self.serializer.deserializeData(d, compr)
        self.assertEqual(floats, v, "float precision must not be compromised in any serializer")

    def testSourceByteTypes_deserialize(self):
        # uncompressed
        call_ser, _ = self.serializer.serializeCall("object", "method", [1, 2, 3], {"kwarg": 42}, False)
        ser, _ = self.serializer.serializeData([4, 5, 6], False)
        _, _, vargs, _ = self.serializer.deserializeCall(bytearray(call_ser), False)
        self.assertEqual([1, 2, 3], vargs)
        d = self.serializer.deserializeData(bytearray(ser), False)
        self.assertEqual([4, 5, 6], d)
        if sys.version_info < (3, 0):
            _, _, vargs, _ = self.serializer.deserializeCall(buffer(call_ser), False)
            self.assertEqual([1, 2, 3], vargs)
            d = self.serializer.deserializeData(buffer(ser), False)
            self.assertEqual([4, 5, 6], d)
        # compressed
        call_ser, _ = self.serializer.serializeCall("object", "method", [1, 2, 3] * 100, {"kwarg": 42}, True)
        ser, _ = self.serializer.serializeData([4, 5, 6] * 100, True)
        _, _, vargs, _ = self.serializer.deserializeCall(bytearray(call_ser), True)
        self.assertEqual(300, len(vargs))
        d = self.serializer.deserializeData(bytearray(ser), True)
        self.assertEqual(300, len(d))
        if sys.version_info < (3, 0):
            _, _, vargs, _ = self.serializer.deserializeCall(buffer(call_ser), True)
            self.assertEqual(300, len(vargs))
            d = self.serializer.deserializeData(buffer(ser), True)
            self.assertEqual(300, len(d))

    @unittest.skipIf(sys.platform == "cli", "ironpython can't properly create memoryviews from serialized data")
    def testSourceByteTypes_deserialize_memoryview(self):
        # uncompressed
        call_ser, _ = self.serializer.serializeCall("object", "method", [1, 2, 3], {"kwarg": 42}, False)
        ser, _ = self.serializer.serializeData([4, 5, 6], False)
        _, _, vargs, _ = self.serializer.deserializeCall(memoryview(call_ser), False)
        self.assertEqual([1, 2, 3], vargs)
        d = self.serializer.deserializeData(memoryview(ser), False)
        self.assertEqual([4, 5, 6], d)
        # compressed
        call_ser, _ = self.serializer.serializeCall("object", "method", [1, 2, 3] * 100, {"kwarg": 42}, True)
        ser, _ = self.serializer.serializeData([4, 5, 6] * 100, True)
        _, _, vargs, _ = self.serializer.deserializeCall(memoryview(call_ser), True)
        self.assertEqual(300, len(vargs))
        d = self.serializer.deserializeData(memoryview(ser), True)
        self.assertEqual(300, len(d))

    def testSourceByteTypes_loads(self):
        call_ser, _ = self.serializer.serializeCall("object", "method", [1, 2, 3], {"kwarg": 42}, False)
        ser, _ = self.serializer.serializeData([4, 5, 6], False)
        _, _, vargs, _ = self.serializer.loadsCall(bytearray(call_ser))
        self.assertEqual([1, 2, 3], vargs)
        d = self.serializer.loads(bytearray(ser))
        self.assertEqual([4, 5, 6], d)
        if sys.version_info < (3, 0):
            _, _, vargs, _ = self.serializer.loadsCall(buffer(call_ser))
            self.assertEqual([1, 2, 3], vargs)
            d = self.serializer.loads(buffer(ser))
            self.assertEqual([4, 5, 6], d)

    @unittest.skipIf(sys.platform == "cli", "ironpython can't properly create memoryviews from serialized data")
    def testSourceByteTypes_loads_memoryview(self):
        call_ser, _ = self.serializer.serializeCall("object", "method", [1, 2, 3], {"kwarg": 42}, False)
        ser, _ = self.serializer.serializeData([4, 5, 6], False)
        _, _, vargs, _ = self.serializer.loadsCall(memoryview(call_ser))
        self.assertEqual([1, 2, 3], vargs)
        d = self.serializer.loads(memoryview(ser))
        self.assertEqual([4, 5, 6], d)

    def testSerializeDumpsAndDumpsCall(self):
        self.serializer.dumps(uuid.uuid4())
        self.serializer.dumps(Pyro4.URI("PYRO:test@test:4444"))
        self.serializer.dumps(Pyro4.Proxy("PYRONAME:foobar"))
        self.serializer.dumpsCall("obj", "method", (1, 2, 3), {"arg1": 999})
        self.serializer.dumpsCall("obj", "method", (1, 2, array.array('i', [1, 2, 3])), {"arg1": 999})
        self.serializer.dumpsCall("obj", "method", (1, 2, array.array('i', [1, 2, 3])), {"arg1": array.array('i', [1, 2, 3])})
        self.serializer.dumpsCall("obj", "method", (1, 2, Pyro4.URI("PYRO:test@test:4444")), {"arg1": 999})
        self.serializer.dumpsCall("obj", "method", (1, 2, Pyro4.URI("PYRO:test@test:4444")), {"arg1": Pyro4.URI("PYRO:test@test:4444")})
        self.serializer.dumpsCall("obj", "method", (1, 2, Pyro4.Proxy("PYRONAME:foobar")), {"arg1": 999})
        self.serializer.dumpsCall("obj", "method", (1, 2, Pyro4.Proxy("PYRONAME:foobar")), {"arg1": Pyro4.Proxy("PYRONAME:foobar")})

    def testArrays(self):
        if sys.version_info < (3, 0):
            a1 = array.array('c', b"hello")
            ser = self.serializer.dumps(a1)
            a2 = self.serializer.loads(ser)
            if type(a2) is array.array:
                self.assertEqual(a1, a2)
            else:
                self.assertEqual(b"hello", a2)
        a1 = array.array('u', unicode("hello"))
        ser = self.serializer.dumps(a1)
        a2 = self.serializer.loads(ser)
        if type(a2) is array.array:
            self.assertEqual(a1, a2)
        else:
            self.assertEqual(unicode("hello"), a2)
        a1 = array.array('h', [222, 333, 444, 555])
        ser = self.serializer.dumps(a1)
        a2 = self.serializer.loads(ser)
        if type(a2) is array.array:
            self.assertEqual(a1, a2)
        else:
            self.assertEqual([222, 333, 444, 555], a2)

    def testArrays2(self):
        if sys.version_info < (3, 0):
            a1 = array.array('c', b"hello")
            ser = self.serializer.dumpsCall("obj", "method", [a1], {})
            a2 = self.serializer.loads(ser)
            a2 = a2["params"][0] if self.SERIALIZER == "json" else a2[2][0]
            if type(a2) is array.array:
                self.assertEqual(a1, a2)
            else:
                self.assertEqual(b"hello", a2)
        a1 = array.array('u', unicode("hello"))
        ser = self.serializer.dumpsCall("obj", "method", [a1], {})
        a2 = self.serializer.loads(ser)
        a2 = a2["params"][0] if self.SERIALIZER == "json" else a2[2][0]
        if type(a2) is array.array:
            self.assertEqual(a1, a2)
        else:
            self.assertEqual(unicode("hello"), a2)
        a1 = array.array('h', [222, 333, 444, 555])
        ser = self.serializer.dumpsCall("obj", "method", [a1], {})
        a2 = self.serializer.loads(ser)
        a2 = a2["params"][0] if self.SERIALIZER == "json" else a2[2][0]
        if type(a2) is array.array:
            self.assertEqual(a1, a2)
        else:
            self.assertEqual([222, 333, 444, 555], a2)


class SerializeTests_cloudpickle(SerializeTests_pickle):
    SERIALIZER = "cloudpickle"

    @unittest.skip('not implemented')
    def testUriSerializationWithoutSlots(self):
        pass

    def testSerializeLambda(self):
        l = lambda x: x * x
        ser, compressed = self.serializer.serializeData(l)
        l2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(l2(3.), 9.)

    def testSerializeLocalFunction(self):
        def f(x):
            return x * x
        ser, compressed = self.serializer.serializeData(f)
        f2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(f2(3.), 9.)


is_ironpython_without_dill = False
try:
    import dill
except ImportError:
    if sys.platform == "cli":
        is_ironpython_without_dill = True


@unittest.skipIf(is_ironpython_without_dill, "dill with ironpython has issues so it's fine if we don't test this")
class SerializeTests_dill(SerializeTests_pickle):
    SERIALIZER = "dill"

    def testProtocolVersion(self):
        import dill
        self.assertEqual(dill.HIGHEST_PROTOCOL, config.DILL_PROTOCOL_VERSION)

    @unittest.skip('not implemented')
    def testUriSerializationWithoutSlots(self):
        pass

    def testSerializeLambda(self):
        l = lambda x: x * x
        ser, compressed = self.serializer.serializeData(l)
        l2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(l2(3.), 9.)

    def testSerializeLocalFunction(self):
        def f(x):
            return x * x
        ser, compressed = self.serializer.serializeData(f)
        f2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(f2(3.), 9.)


class SerializeTests_serpent(SerializeTests_pickle):
    SERIALIZER = "serpent"

    def testCircular(self):
        # serpent doesn't support object graphs (since serpent 1.7 reports ValueError instead of crashing)
        with self.assertRaises(ValueError):
            super(SerializeTests_serpent, self).testCircular()

    def testSet(self):
        data = {111, 222, 333}
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(tuple(data), tuple(data2))

    def testDeque(self):
        # serpent converts a deque into a primitive list
        deq = collections.deque([1, 2, 3, 4])
        ser, compressed = self.serializer.serializeData(deq)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual([1, 2, 3, 4], data2)

    @unittest.skipIf(sys.version_info < (2, 7), "ordereddict is in Python 2.7+")
    def testOrderedDict(self):
        od = collections.OrderedDict()
        od["a"] = 1
        od["b"] = 2
        od["c"] = 3
        def recreate_OrderedDict(name, values):
            self.assertEqual("collections.OrderedDict", name)
            return collections.OrderedDict(values["items"])
        Pyro4.util.SerializerBase.register_dict_to_class("collections.OrderedDict", recreate_OrderedDict)
        ser, compressed = self.serializer.serializeData(od)
        self.assertIn(b"collections.OrderedDict", ser)
        self.assertIn(b"[('a',1),('b',2),('c',3)]", ser)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(od, data2)

    def testUriSerializationWithoutSlots(self):
        u = Pyro4.core.URI("PYRO:obj@localhost:1234")
        d, compr = self.serializer.serializeData(u)
        self.assertFalse(compr)
        result1 = b"# serpent utf-8 python3.2\n{'__class__':'Pyro4.core.URI','state':('PYRO','obj',None,'localhost',1234)}"
        result2 = b"# serpent utf-8 python3.2\n{'state':('PYRO','obj',None,'localhost',1234),'__class__':'Pyro4.core.URI'}"
        result3 = b"# serpent utf-8 python2.6\n{'state':('PYRO','obj',None,'localhost',1234),'__class__':'Pyro4.core.URI'}"
        result4 = b"# serpent utf-8 python2.6\n{'__class__':'Pyro4.core.URI','state':('PYRO','obj',None,'localhost',1234)}"
        self.assertTrue(d in (result1, result2, result3, result4))


class SerializeTests_json(SerializeTests_pickle):
    SERIALIZER = "json"

    def testCircular(self):
        with self.assertRaises(ValueError):  # json doesn't support object graphs
            super(SerializeTests_json, self).testCircular()

    def testSet(self):
        # json serializes a set into a list, so we override this
        data = {111, 222, 333}
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(list(data), data2)

    def testUriSerializationWithoutSlots(self):
        u = Pyro4.core.URI("PYRO:obj@localhost:1234")
        d, compr = self.serializer.serializeData(u)
        self.assertFalse(compr)
        result1 = b'{"__class__": "Pyro4.core.URI", "state": ["PYRO", "obj", null, "localhost", 1234]}'
        result2 = b'{"state": ["PYRO", "obj", null, "localhost", 1234], "__class__": "Pyro4.core.URI"}'
        self.assertTrue(d in (result1, result2))


class SerializeTests_marshal(SerializeTests_pickle):
    SERIALIZER = "marshal"

    def testCircular(self):
        with self.assertRaises(ValueError):  # marshal doesn't support object graphs
            super(SerializeTests_marshal, self).testCircular()

    @unittest.skip("marshaling is implementation dependent")
    def testUriSerializationWithoutSlots(self):
        pass


class SerializeTests_msgpack(SerializeTests_pickle):
    SERIALIZER = "msgpack"

    @unittest.skip("circular will crash msgpack")
    def testCircular(self):
        pass

    def testSet(self):
        # msgpack serializes a set into a list, so we override this
        data = {111, 222, 333}
        ser, compressed = self.serializer.serializeData(data)
        data2 = self.serializer.deserializeData(ser, compressed=compressed)
        self.assertEqual(list(data), data2)

    @unittest.skip("msgpack is implementation dependent")
    def testUriSerializationWithoutSlots(self):
        pass


class GenericTests(unittest.TestCase):
    def testSerializersAvailable(self):
        Pyro4.util.get_serializer("pickle")
        Pyro4.util.get_serializer("marshal")
        try:
            import json
            Pyro4.util.get_serializer("json")
        except ImportError:
            pass
        try:
            import serpent
            Pyro4.util.get_serializer("serpent")
        except ImportError:
            pass
        try:
            import cloudpickle
            Pyro4.util.get_serializer("cloudpickle")
        except ImportError:
            pass
        try:
            import dill
            Pyro4.util.get_serializer("dill")
        except ImportError:
            pass

    def testAssignedSerializerIds(self):
        self.assertEqual(1, Pyro4.util.SerpentSerializer.serializer_id)
        self.assertEqual(2, Pyro4.util.JsonSerializer.serializer_id)
        self.assertEqual(3, Pyro4.util.MarshalSerializer.serializer_id)
        self.assertEqual(4, Pyro4.util.PickleSerializer.serializer_id)
        self.assertEqual(5, Pyro4.util.DillSerializer.serializer_id)
        self.assertEqual(6, Pyro4.util.MsgpackSerializer.serializer_id)
        self.assertEqual(7, Pyro4.util.CloudpickleSerializer.serializer_id)

    def testSerializersAvailableById(self):
        Pyro4.util.get_serializer_by_id(1)  # serpent
        Pyro4.util.get_serializer_by_id(2)  # json
        Pyro4.util.get_serializer_by_id(3)  # marshal
        Pyro4.util.get_serializer_by_id(4)  # pickle
        # ids 5, 6 and 7 (dill, msgpack, cloudpickle) are not always available, so we skip those.
        self.assertRaises(Pyro4.errors.SerializeError, lambda: Pyro4.util.get_serializer_by_id(0))
        self.assertRaises(Pyro4.errors.SerializeError, lambda: Pyro4.util.get_serializer_by_id(8))

    @unittest.skip("skip tetes on python3.11")
    def testDictClassFail(self):
        o = pprint.PrettyPrinter(stream="dummy", width=42)
        d = Pyro4.util.SerializerBase.class_to_dict(o)
        self.assertEqual(42, d["_width"])
        self.assertEqual("pprint.PrettyPrinter", d["__class__"])
        try:
            _ = Pyro4.util.SerializerBase.dict_to_class(d)
            self.fail("error expected")
        except Pyro4.errors.ProtocolError:
            pass

    def testDictException(self):
        x = ZeroDivisionError("hello", 42)
        expected = {
            "__class__": None,
            "__exception__": True,
            "args": ("hello", 42),
            "attributes": {}
        }
        if sys.version_info < (3, 0):
            expected["__class__"] = "exceptions.ZeroDivisionError"
        else:
            expected["__class__"] = "builtins.ZeroDivisionError"
        d = Pyro4.util.SerializerBase.class_to_dict(x)
        self.assertEqual(expected, d)
        x.custom_attribute = 999
        expected["attributes"] = {"custom_attribute": 999}
        d = Pyro4.util.SerializerBase.class_to_dict(x)
        self.assertEqual(expected, d)

    def testDictClassOk(self):
        uri = Pyro4.core.URI("PYRO:object@host:4444")
        d = Pyro4.util.SerializerBase.class_to_dict(uri)
        self.assertEqual("Pyro4.core.URI", d["__class__"])
        self.assertIn("state", d)
        x = Pyro4.util.SerializerBase.dict_to_class(d)
        self.assertIsInstance(x, Pyro4.core.URI)
        self.assertEqual(uri, x)
        self.assertEqual(4444, x.port)
        uri = Pyro4.core.URI("PYRO:12345@./u:/tmp/socketname")
        d = Pyro4.util.SerializerBase.class_to_dict(uri)
        self.assertEqual("Pyro4.core.URI", d["__class__"])
        self.assertIn("state", d)
        x = Pyro4.util.SerializerBase.dict_to_class(d)
        self.assertIsInstance(x, Pyro4.core.URI)
        self.assertEqual(uri, x)
        self.assertEqual("/tmp/socketname", x.sockname)

    @unittest.skip("skip tetes on python3.11")
    def testCustomDictClass(self):
        o = MyThingPartlyExposed("test")
        Pyro4.util.SerializerBase.register_class_to_dict(MyThingPartlyExposed, mything_dict)
        Pyro4.util.SerializerBase.register_dict_to_class("CUSTOM-Mythingymabob", mything_creator)
        d = Pyro4.util.SerializerBase.class_to_dict(o)
        self.assertEqual("CUSTOM-Mythingymabob", d["__class__"])
        self.assertEqual("test", d["name"])
        x = Pyro4.util.SerializerBase.dict_to_class(d)
        self.assertIsInstance(x, MyThingPartlyExposed)
        self.assertEqual("test", x.name)
        # unregister the conversion functions and try again
        Pyro4.util.SerializerBase.unregister_class_to_dict(MyThingPartlyExposed)
        Pyro4.util.SerializerBase.unregister_dict_to_class("CUSTOM-Mythingymabob")
        d_orig = Pyro4.util.SerializerBase.class_to_dict(o)
        clsname = d_orig["__class__"]
        self.assertTrue(clsname.endswith("testsupport.MyThingPartlyExposed"))
        try:
            _ = Pyro4.util.SerializerBase.dict_to_class(d)
            self.fail("should crash")
        except Pyro4.errors.ProtocolError:
            pass  # ok

    def testExceptionNamespacePy2(self):
        data = {'__class__': 'exceptions.ZeroDivisionError',
                '__exception__': True,
                'args': ('hello', 42),
                'attributes': {"test_attribute": 99}}
        exc = Pyro4.util.SerializerBase.dict_to_class(data)
        self.assertIsInstance(exc, ZeroDivisionError)
        self.assertEqual("ZeroDivisionError('hello', 42)", repr(exc))
        self.assertEqual(99, exc.test_attribute)

    def testExceptionNamespacePy3(self):
        data = {'__class__': 'builtins.ZeroDivisionError',
                '__exception__': True,
                'args': ('hello', 42),
                'attributes': {"test_attribute": 99}}
        exc = Pyro4.util.SerializerBase.dict_to_class(data)
        self.assertIsInstance(exc, ZeroDivisionError)
        self.assertEqual("ZeroDivisionError('hello', 42)", repr(exc))
        self.assertEqual(99, exc.test_attribute)

    def testExceptionNotTagged(self):
        data = {'__class__': 'builtins.ZeroDivisionError',
                'args': ('hello', 42),
                'attributes': {}}
        with self.assertRaises(Pyro4.errors.SerializeError) as cm:
            _ = Pyro4.util.SerializerBase.dict_to_class(data)
        self.assertEqual("unsupported serialized class: builtins.ZeroDivisionError", str(cm.exception))

    def testWeirdFloats(self):
        ser = Pyro4.util.get_serializer(config.SERIALIZER)
        p, _ = ser.serializeData([float("+inf"), float("-inf"), float("nan")])
        s2 = ser.deserializeData(p)
        self.assertTrue(math.isinf(s2[0]))
        self.assertEqual(1.0, math.copysign(1, s2[0]))
        self.assertTrue(math.isinf(s2[1]))
        self.assertEqual(-1.0, math.copysign(1, s2[1]))
        self.assertTrue(math.isnan(s2[2]))


def mything_dict(obj):
    return {
        "__class__": "CUSTOM-Mythingymabob",
        "name": obj.name
    }


def mything_creator(classname, d):
    assert classname == "CUSTOM-Mythingymabob"
    assert d["__class__"] == "CUSTOM-Mythingymabob"
    return MyThingPartlyExposed(d["name"])


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