test_pickle.py 文件源码

def test_round_trip_current(self):

        try:
            import cPickle as c_pickle

            def c_pickler(obj, path):
                with open(path, 'wb') as fh:
                    c_pickle.dump(obj, fh, protocol=-1)

            def c_unpickler(path):
                with open(path, 'rb') as fh:
                    fh.seek(0)
                    return c_pickle.load(fh)
        except:
            c_pickler = None
            c_unpickler = None

        import pickle as python_pickle

        def python_pickler(obj, path):
            with open(path, 'wb') as fh:
                python_pickle.dump(obj, fh, protocol=-1)

        def python_unpickler(path):
            with open(path, 'rb') as fh:
                fh.seek(0)
                return python_pickle.load(fh)

        for typ, dv in self.data.items():
            for dt, expected in dv.items():

                for writer in [pd.to_pickle, c_pickler, python_pickler]:
                    if writer is None:
                        continue

                    with tm.ensure_clean(self.path) as path:

                        # test writing with each pickler
                        writer(expected, path)

                        # test reading with each unpickler
                        result = pd.read_pickle(path)
                        self.compare_element(result, expected, typ)

                        if c_unpickler is not None:
                            result = c_unpickler(path)
                            self.compare_element(result, expected, typ)

                        result = python_unpickler(path)
                        self.compare_element(result, expected, typ)