test_newserialized.py
107 lines
| 3.9 KiB
| text/x-python
|
PythonLexer
MinRK
|
r3637 | """test serialization with newserialized""" | |
MinRK
|
r3539 | ||
MinRK
|
r3664 | #------------------------------------------------------------------------------- | |
# Copyright (C) 2011 The IPython Development Team | |||
# | |||
# Distributed under the terms of the BSD License. The full license is in | |||
# the file COPYING, distributed as part of this software. | |||
#------------------------------------------------------------------------------- | |||
#------------------------------------------------------------------------------- | |||
# Imports | |||
#------------------------------------------------------------------------------- | |||
MinRK
|
r3539 | from unittest import TestCase | |
MinRK
|
r3672 | from IPython.testing.decorators import parametric | |
MinRK
|
r3637 | from IPython.utils import newserialized as ns | |
from IPython.utils.pickleutil import can, uncan, CannedObject, CannedFunction | |||
MinRK
|
r3666 | from IPython.parallel.tests.clienttest import skip_without | |
MinRK
|
r3637 | ||
class CanningTestCase(TestCase): | |||
def test_canning(self): | |||
d = dict(a=5,b=6) | |||
cd = can(d) | |||
MinRK
|
r3661 | self.assertTrue(isinstance(cd, dict)) | |
MinRK
|
r3637 | ||
def test_canned_function(self): | |||
f = lambda : 7 | |||
cf = can(f) | |||
MinRK
|
r3661 | self.assertTrue(isinstance(cf, CannedFunction)) | |
MinRK
|
r3637 | ||
@parametric | |||
def test_can_roundtrip(cls): | |||
objs = [ | |||
dict(), | |||
set(), | |||
list(), | |||
['a',1,['a',1],u'e'], | |||
] | |||
return map(cls.run_roundtrip, objs) | |||
@classmethod | |||
MinRK
|
r3661 | def run_roundtrip(self, obj): | |
MinRK
|
r3637 | o = uncan(can(obj)) | |
MinRK
|
r3661 | assert o == obj, "failed assertion: %r == %r"%(o,obj) | |
MinRK
|
r3637 | ||
def test_serialized_interfaces(self): | |||
us = {'a':10, 'b':range(10)} | |||
s = ns.serialize(us) | |||
uus = ns.unserialize(s) | |||
MinRK
|
r3661 | self.assertTrue(isinstance(s, ns.SerializeIt)) | |
self.assertEquals(uus, us) | |||
MinRK
|
r3637 | ||
def test_pickle_serialized(self): | |||
obj = {'a':1.45345, 'b':'asdfsdf', 'c':10000L} | |||
original = ns.UnSerialized(obj) | |||
originalSer = ns.SerializeIt(original) | |||
firstData = originalSer.getData() | |||
firstTD = originalSer.getTypeDescriptor() | |||
firstMD = originalSer.getMetadata() | |||
MinRK
|
r3661 | self.assertEquals(firstTD, 'pickle') | |
self.assertEquals(firstMD, {}) | |||
MinRK
|
r3637 | unSerialized = ns.UnSerializeIt(originalSer) | |
secondObj = unSerialized.getObject() | |||
for k, v in secondObj.iteritems(): | |||
MinRK
|
r3661 | self.assertEquals(obj[k], v) | |
MinRK
|
r3637 | secondSer = ns.SerializeIt(ns.UnSerialized(secondObj)) | |
MinRK
|
r3661 | self.assertEquals(firstData, secondSer.getData()) | |
self.assertEquals(firstTD, secondSer.getTypeDescriptor() ) | |||
self.assertEquals(firstMD, secondSer.getMetadata()) | |||
MinRK
|
r3637 | ||
@skip_without('numpy') | |||
def test_ndarray_serialized(self): | |||
import numpy | |||
a = numpy.linspace(0.0, 1.0, 1000) | |||
unSer1 = ns.UnSerialized(a) | |||
ser1 = ns.SerializeIt(unSer1) | |||
td = ser1.getTypeDescriptor() | |||
MinRK
|
r3661 | self.assertEquals(td, 'ndarray') | |
MinRK
|
r3637 | md = ser1.getMetadata() | |
MinRK
|
r3661 | self.assertEquals(md['shape'], a.shape) | |
self.assertEquals(md['dtype'], a.dtype.str) | |||
MinRK
|
r3637 | buff = ser1.getData() | |
MinRK
|
r3661 | self.assertEquals(buff, numpy.getbuffer(a)) | |
MinRK
|
r3637 | s = ns.Serialized(buff, td, md) | |
final = ns.unserialize(s) | |||
MinRK
|
r3661 | self.assertEquals(numpy.getbuffer(a), numpy.getbuffer(final)) | |
self.assertTrue((a==final).all()) | |||
self.assertEquals(a.dtype.str, final.dtype.str) | |||
self.assertEquals(a.shape, final.shape) | |||
MinRK
|
r3637 | # test non-copying: | |
a[2] = 1e9 | |||
MinRK
|
r3661 | self.assertTrue((a==final).all()) | |
MinRK
|
r3664 | ||
def test_uncan_function_globals(self): | |||
"""test that uncanning a module function restores it into its module""" | |||
from re import search | |||
cf = can(search) | |||
csearch = uncan(cf) | |||
self.assertEqual(csearch.__module__, search.__module__) | |||
self.assertNotEqual(csearch('asd', 'asdf'), None) | |||
csearch = uncan(cf, dict(a=5)) | |||
self.assertEqual(csearch.__module__, search.__module__) | |||
self.assertNotEqual(csearch('asd', 'asdf'), None) | |||
MinRK
|
r3637 | ||