test cross version serialization by launching subprocesses; much cleaner!
--HG-- branch : trunk
This commit is contained in:
Benjamin Peterson
parent
8f69d23f18
commit
4d598370b4
|
|
@ -35,8 +35,6 @@ if _INPY3:
|
|||
pass
|
||||
bytes = bytes
|
||||
else:
|
||||
class bytes(str):
|
||||
pass
|
||||
b = str
|
||||
_b = bytes
|
||||
_unicode = unicode
|
||||
|
|
@ -80,25 +78,28 @@ class Serializer(object):
|
|||
raise UnserializableType("can't serialize %s" % (tp,))
|
||||
dispatch(self, obj)
|
||||
|
||||
dispatch = {}
|
||||
|
||||
def save_bytes(self, bytes_):
|
||||
self.stream.write(BYTES)
|
||||
self._write_byte_sequence(bytes_)
|
||||
dispatch[bytes] = save_bytes
|
||||
|
||||
def save_unicode(self, s):
|
||||
self.stream.write(UNICODE)
|
||||
self._write_unicode_string(s)
|
||||
|
||||
def save_string(self, s):
|
||||
if _INPY3:
|
||||
if _INPY3:
|
||||
def save_string(self, s):
|
||||
self.stream.write(PY3STRING)
|
||||
self._write_unicode_string(s)
|
||||
else:
|
||||
# Case for tests
|
||||
if _REALLY_PY3 and isinstance(s, str):
|
||||
s = s.encode("latin-1")
|
||||
else:
|
||||
def save_string(self, s):
|
||||
self.stream.write(PY2STRING)
|
||||
self._write_byte_sequence(s)
|
||||
|
||||
def save_unicode(self, s):
|
||||
self.stream.write(UNICODE)
|
||||
self._write_unicode_string(s)
|
||||
dispatch[unicode] = save_unicode
|
||||
dispatch[str] = save_string
|
||||
|
||||
def _write_unicode_string(self, s):
|
||||
try:
|
||||
as_bytes = s.encode("utf-8")
|
||||
|
|
@ -113,6 +114,7 @@ class Serializer(object):
|
|||
def save_int(self, i):
|
||||
self.stream.write(INT)
|
||||
self._write_int4(i)
|
||||
dispatch[int] = save_int
|
||||
|
||||
def _write_int4(self, i, error="int must be less than %i" %
|
||||
(FOUR_BYTE_INT_MAX,)):
|
||||
|
|
@ -125,6 +127,7 @@ class Serializer(object):
|
|||
self._write_int4(len(L), "list is too long")
|
||||
for i, item in enumerate(L):
|
||||
self._write_setitem(i, item)
|
||||
dispatch[list] = save_list
|
||||
|
||||
def _write_setitem(self, key, value):
|
||||
self._save(key)
|
||||
|
|
@ -135,12 +138,14 @@ class Serializer(object):
|
|||
self.stream.write(NEWDICT)
|
||||
for key, value in d.items():
|
||||
self._write_setitem(key, value)
|
||||
dispatch[dict] = save_dict
|
||||
|
||||
def save_tuple(self, tup):
|
||||
for item in tup:
|
||||
self._save(item)
|
||||
self.stream.write(BUILDTUPLE)
|
||||
self._write_int4(len(tup), "tuple is too long")
|
||||
dispatch[tuple] = save_tuple
|
||||
|
||||
|
||||
class _UnserializationOptions(object):
|
||||
|
|
@ -156,35 +161,10 @@ class _Py3UnserializationOptions(_UnserializationOptions):
|
|||
def __init__(self, py2_strings_as_str=False):
|
||||
self.py2_strings_as_str = py2_strings_as_str
|
||||
|
||||
|
||||
_unchanging_dispatch = {}
|
||||
for tp in (dict, list, tuple, int):
|
||||
name = "save_%s" % (tp.__name__,)
|
||||
_unchanging_dispatch[tp] = getattr(Serializer, name)
|
||||
del tp, name
|
||||
|
||||
def _setup_dispatch():
|
||||
dispatch = _unchanging_dispatch.copy()
|
||||
# This is sutble. bytes is aliased to str in 2.6, so
|
||||
# dispatch[bytes] is overwritten. Additionally, we alias unicode
|
||||
# to str in 3.x, so dispatch[unicode] is overwritten with
|
||||
# save_string.
|
||||
dispatch[bytes] = Serializer.save_bytes
|
||||
dispatch[unicode] = Serializer.save_unicode
|
||||
dispatch[str] = Serializer.save_string
|
||||
Serializer.dispatch = dispatch
|
||||
|
||||
def _setup_version_dependent_constants(leave_unicode_alone=False):
|
||||
global unicode, UnserializationOptions
|
||||
if _INPY3:
|
||||
unicode = str
|
||||
UnserializationOptions = _Py3UnserializationOptions
|
||||
else:
|
||||
UnserializationOptions = _Py2UnserializationOptions
|
||||
unicode = _unicode
|
||||
_setup_dispatch()
|
||||
_setup_version_dependent_constants()
|
||||
|
||||
if _INPY3:
|
||||
UnserializationOptions = _Py3UnserializationOptions
|
||||
else:
|
||||
UnserializationOptions = _Py2UnserializationOptions
|
||||
|
||||
class _Stop(Exception):
|
||||
pass
|
||||
|
|
@ -236,7 +216,7 @@ class Unserializer(object):
|
|||
|
||||
def load_py3string(self):
|
||||
as_bytes = self._read_byte_string()
|
||||
if (not _INPY3 and self.options.py3_strings_as_str) and not _REALLY_PY3:
|
||||
if not _INPY3 and self.options.py3_strings_as_str:
|
||||
# XXX Should we try to decode into latin-1?
|
||||
self.stack.append(as_bytes)
|
||||
else:
|
||||
|
|
@ -245,8 +225,7 @@ class Unserializer(object):
|
|||
|
||||
def load_py2string(self):
|
||||
as_bytes = self._read_byte_string()
|
||||
if (_INPY3 and self.options.py2_strings_as_str) or \
|
||||
(_REALLY_PY3 and not _INPY3):
|
||||
if _INPY3 and self.options.py2_strings_as_str:
|
||||
s = as_bytes.decode("latin-1")
|
||||
else:
|
||||
s = as_bytes
|
||||
|
|
@ -254,7 +233,7 @@ class Unserializer(object):
|
|||
opcodes[PY2STRING] = load_py2string
|
||||
|
||||
def load_bytes(self):
|
||||
s = bytes(self._read_byte_string())
|
||||
s = self._read_byte_string()
|
||||
self.stack.append(s)
|
||||
opcodes[BYTES] = load_bytes
|
||||
|
||||
|
|
|
|||
|
|
@ -1,50 +1,78 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
import shutil
|
||||
import sys
|
||||
import os
|
||||
import tempfile
|
||||
import subprocess
|
||||
import py
|
||||
from py.__.execnet import serializer
|
||||
|
||||
|
||||
def _find_version(suffix=""):
|
||||
name = "python" + suffix
|
||||
executable = py.path.local.sysfind(name)
|
||||
if executable is None:
|
||||
py.test.skip("can't find a %r executable" % (name,))
|
||||
return executable
|
||||
|
||||
def setup_module(mod):
|
||||
mod._save_python3 = serializer._INPY3
|
||||
mod.TEMPDIR = py.path.local(tempfile.mkdtemp())
|
||||
if sys.version_info > (3, 0):
|
||||
mod._py3_wrapper = PythonWrapper(py.path.local(sys.executable))
|
||||
mod._py2_wrapper = PythonWrapper(_find_version())
|
||||
else:
|
||||
mod._py3_wrapper = PythonWrapper(_find_version("3"))
|
||||
mod._py2_wrapper = PythonWrapper(py.path.local(sys.executable))
|
||||
mod._old_pypath = os.environ.get("PYTHONPATH")
|
||||
pylib = str(py.path.local(py.__file__).dirpath().join(".."))
|
||||
os.environ["PYTHONPATH"] = pylib
|
||||
|
||||
def teardown_module(mod):
|
||||
serializer._setup_version_dependent_constants()
|
||||
TEMPDIR.remove(True)
|
||||
if _old_pypath is not None:
|
||||
os.environ["PYTHONPATH"] = _old_pypath
|
||||
|
||||
def _dump(obj):
|
||||
stream = py.io.BytesIO()
|
||||
saver = serializer.Serializer(stream)
|
||||
saver.save(obj)
|
||||
return stream.getvalue()
|
||||
|
||||
def _load(serialized, str_coerion):
|
||||
stream = py.io.BytesIO(serialized)
|
||||
opts = serializer.UnserializationOptions(str_coerion)
|
||||
unserializer = serializer.Unserializer(stream, opts)
|
||||
return unserializer.load()
|
||||
class PythonWrapper(object):
|
||||
|
||||
def _run_in_version(is_py3, func, *args):
|
||||
serializer._INPY3 = is_py3
|
||||
serializer._setup_version_dependent_constants()
|
||||
try:
|
||||
return func(*args)
|
||||
finally:
|
||||
serializer._INPY3 = _save_python3
|
||||
def __init__(self, executable):
|
||||
self.executable = executable
|
||||
|
||||
def dump_py2(obj):
|
||||
return _run_in_version(False, _dump, obj)
|
||||
def dump(self, obj_rep):
|
||||
script_file = TEMPDIR.join("dump.py")
|
||||
script_file.write("""
|
||||
from py.__.execnet import serializer
|
||||
import sys
|
||||
if sys.version_info > (3, 0): # Need binary output
|
||||
sys.stdout = sys.stdout.detach()
|
||||
saver = serializer.Serializer(sys.stdout)
|
||||
saver.save(%s)""" % (obj_rep,))
|
||||
return self.executable.sysexec(script_file)
|
||||
|
||||
def dump_py3(obj):
|
||||
return _run_in_version(True, _dump, obj)
|
||||
def load(self, data, option_args=""):
|
||||
script_file = TEMPDIR.join("load.py")
|
||||
script_file.write(r"""
|
||||
from py.__.execnet import serializer
|
||||
import sys
|
||||
if sys.version_info > (3, 0):
|
||||
sys.stdin = sys.stdin.detach()
|
||||
options = serializer.UnserializationOptions(%s)
|
||||
loader = serializer.Unserializer(sys.stdin, options)
|
||||
obj = loader.load()
|
||||
sys.stdout.write(type(obj).__name__ + "\n")
|
||||
sys.stdout.write(repr(obj))""" % (option_args,))
|
||||
popen = subprocess.Popen([str(self.executable), str(script_file)],
|
||||
stdin=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE)
|
||||
stdout, stderr = popen.communicate(data.encode("latin-1"))
|
||||
ret = popen.returncode
|
||||
if ret:
|
||||
raise py.process.cmdexec.Error(ret, ret, str(self.executable),
|
||||
stdout, stderr)
|
||||
return [s.decode("ascii") for s in stdout.splitlines()]
|
||||
|
||||
def load_py2(serialized, str_coercion=False):
|
||||
return _run_in_version(False, _load, serialized, str_coercion)
|
||||
|
||||
def load_py3(serialized, str_coercion=False):
|
||||
return _run_in_version(True, _load, serialized, str_coercion)
|
||||
|
||||
try:
|
||||
bytes
|
||||
except NameError:
|
||||
bytes = str
|
||||
def __repr__(self):
|
||||
return "<PythonWrapper for %s>" % (self.executable,)
|
||||
|
||||
|
||||
def pytest_funcarg__py2(request):
|
||||
|
|
@ -55,73 +83,86 @@ def pytest_funcarg__py3(request):
|
|||
|
||||
class TestSerializer:
|
||||
|
||||
def test_int(self):
|
||||
for dump in dump_py2, dump_py3:
|
||||
p = dump_py2(4)
|
||||
for load in load_py2, load_py3:
|
||||
i = load(p)
|
||||
assert isinstance(i, int)
|
||||
assert i == 4
|
||||
py.test.raises(serializer.SerializationError, dump, 123456678900)
|
||||
def test_int(self, py2, py3):
|
||||
for dump in py2.dump, py3.dump:
|
||||
p = dump(4)
|
||||
for load in py2.load, py3.load:
|
||||
tp, v = load(p)
|
||||
assert tp == "int"
|
||||
assert int(v) == 4
|
||||
py.test.raises(serializer.SerializationError,
|
||||
serializer.Serializer(py.io.BytesIO()).save,
|
||||
123456678900)
|
||||
|
||||
def test_bytes(self):
|
||||
for dump in dump_py2, dump_py3:
|
||||
p = dump(serializer._b('hi'))
|
||||
for load in load_py2, load_py3:
|
||||
s = load(p)
|
||||
assert isinstance(s, serializer.bytes)
|
||||
assert s == serializer._b('hi')
|
||||
def test_bytes(self, py2, py3):
|
||||
p = py3.dump("b'hi'")
|
||||
tp, v = py2.load(p)
|
||||
assert tp == "str"
|
||||
assert v == "'hi'"
|
||||
tp, v = py3.load(p)
|
||||
assert tp == "bytes"
|
||||
assert v == "b'hi'"
|
||||
|
||||
def check_sequence(self, seq):
|
||||
for dump in dump_py2, dump_py3:
|
||||
def check_sequence(self, seq, tp_name, rep, py2, py3):
|
||||
for dump in py2.dump, py3.dump:
|
||||
p = dump(seq)
|
||||
for load in load_py2, load_py3:
|
||||
l = load(p)
|
||||
assert l == seq
|
||||
for load in py2.load, py3.load:
|
||||
tp, v = load(p)
|
||||
assert tp == tp_name
|
||||
assert v == rep
|
||||
|
||||
def test_list(self):
|
||||
self.check_sequence([1, 2, 3])
|
||||
def test_list(self, py2, py3):
|
||||
self.check_sequence([1, 2, 3], "list", "[1, 2, 3]", py2, py3)
|
||||
|
||||
@py.test.mark.xfail
|
||||
# I'm not sure if we need the complexity.
|
||||
def test_recursive_list(self):
|
||||
def test_recursive_list(self, py2, py3):
|
||||
l = [1, 2, 3]
|
||||
l.append(l)
|
||||
self.check_sequence(l)
|
||||
p = py2.dump(l)
|
||||
tp, rep = py2.load(l)
|
||||
assert tp == "list"
|
||||
|
||||
def test_tuple(self):
|
||||
self.check_sequence((1, 2, 3))
|
||||
def test_tuple(self, py2, py3):
|
||||
self.check_sequence((1, 2, 3), "tuple", "(1, 2, 3)", py2, py3)
|
||||
|
||||
def test_dict(self):
|
||||
for dump in dump_py2, dump_py3:
|
||||
p = dump({"hi" : 2, (1, 2, 3) : 32})
|
||||
for load in load_py2, load_py3:
|
||||
d = load(p, True)
|
||||
assert d == {"hi" : 2, (1, 2, 3) : 32}
|
||||
def test_dict(self, py2, py3):
|
||||
for dump in py2.dump, py3.dump:
|
||||
p = dump({6 : 2, (1, 2, 3) : 32})
|
||||
for load in py2.load, py3.load:
|
||||
tp, v = load(p)
|
||||
assert tp == "dict"
|
||||
# XXX comparing dict reprs
|
||||
assert v == "{6: 2, (1, 2, 3): 32}"
|
||||
|
||||
def test_string(self):
|
||||
py.test.skip("will rewrite")
|
||||
p = dump_py2("xyz")
|
||||
s = load_py2(p)
|
||||
assert isinstance(s, str)
|
||||
assert s == "xyz"
|
||||
s = load_py3(p)
|
||||
assert isinstance(s, bytes)
|
||||
assert s == serializer.b("xyz")
|
||||
p = dump_py2("xyz")
|
||||
s = load_py3(p, True)
|
||||
assert isinstance(s, serializer._unicode)
|
||||
assert s == serializer.unicode("xyz")
|
||||
p = dump_py3("xyz")
|
||||
s = load_py2(p, True)
|
||||
assert isinstance(s, str)
|
||||
assert s == "xyz"
|
||||
def test_string(self, py2, py3):
|
||||
p = py2.dump("'xyz'")
|
||||
tp, s = py2.load(p)
|
||||
assert tp == "str"
|
||||
assert s == "'xyz'"
|
||||
tp, s = py3.load(p)
|
||||
assert tp == "bytes"
|
||||
assert s == "b'xyz'"
|
||||
tp, s = py3.load(p, "True")
|
||||
assert tp == "str"
|
||||
assert s == "'xyz'"
|
||||
p = py3.dump("'xyz'")
|
||||
tp, s = py2.load(p, True)
|
||||
assert tp == "str"
|
||||
assert s == "'xyz'"
|
||||
|
||||
def test_unicode(self):
|
||||
py.test.skip("will rewrite")
|
||||
for dump, uni in (dump_py2, serializer._unicode), (dump_py3, str):
|
||||
p = dump(uni("xyz"))
|
||||
for load in load_py2, load_py3:
|
||||
s = load(p)
|
||||
assert isinstance(s, serializer._unicode)
|
||||
assert s == serializer._unicode("xyz")
|
||||
def test_unicode(self, py2, py3):
|
||||
p = py2.dump("u'hi'")
|
||||
tp, s = py2.load(p)
|
||||
assert tp == "unicode"
|
||||
assert s == "u'hi'"
|
||||
tp, s = py3.load(p)
|
||||
assert tp == "str"
|
||||
assert s == "'hi'"
|
||||
p = py3.dump("'hi'")
|
||||
tp, s = py3.load(p)
|
||||
assert tp == "str"
|
||||
assert s == "'hi'"
|
||||
tp, s = py2.load(p)
|
||||
assert tp == "unicode"
|
||||
assert s == "u'hi'"
|
||||
|
|
|
|||
Loading…
Reference in New Issue