[pypy-svn] r77588 - in pypy/branch/fast-forward/pypy/module: _io _io/test _multiprocessing/test

afa at codespeak.net afa at codespeak.net
Mon Oct 4 23:11:50 CEST 2010


Author: afa
Date: Mon Oct  4 23:11:48 2010
New Revision: 77588

Modified:
   pypy/branch/fast-forward/pypy/module/_io/interp_stringio.py   (contents, props changed)
   pypy/branch/fast-forward/pypy/module/_io/test/test_stringio.py   (contents, props changed)
   pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py   (contents, props changed)
Log:
fixeol


Modified: pypy/branch/fast-forward/pypy/module/_io/interp_stringio.py
==============================================================================
--- pypy/branch/fast-forward/pypy/module/_io/interp_stringio.py	(original)
+++ pypy/branch/fast-forward/pypy/module/_io/interp_stringio.py	Mon Oct  4 23:11:48 2010
@@ -1,68 +1,68 @@
-from pypy.module._io.interp_io import W_TextIOBase
-from pypy.interpreter.typedef import TypeDef
-from pypy.interpreter.gateway import interp2app, unwrap_spec
-from pypy.interpreter.error import operationerrfmt
-from pypy.interpreter.baseobjspace import ObjSpace, W_Root
-
-class W_StringIO(W_TextIOBase):
-    def __init__(self):
-        self.buf = []
-        self.pos = 0
-
-    def _check_closed(self):
-        pass
-    def _check_initialized(self):
-        pass
-
-    @unwrap_spec(ObjSpace, W_Root)
-    def descr_new(space, w_subtype):
-        self = space.allocate_instance(W_StringIO, w_subtype)
-        W_StringIO.__init__(self)
-        return space.wrap(self)
-
-    def resize_buffer(self, newlength):
-        if len(self.buf) > newlength:
-            self.buf = self.buf[:newlength]
-        if len(self.buf) < newlength:
-            self.buf.extend([u'\0'] * (newlength - len(self.buf)))
-
-    def write(self, string):
-        # XXX self.decoder
-        decoded = string
-        # XXX writenl
-
-        length = len(decoded)
-        if self.pos + length > len(self.buf):
-            self.resize_buffer(self.pos + length)
-
-        for i in range(length):
-            self.buf[self.pos + i] = string[i]
-        self.pos += length
-
-    @unwrap_spec('self', ObjSpace, W_Root)
-    def write_w(self, space, w_obj):
-        self._check_initialized()
-        if not space.isinstance_w(w_obj, space.w_unicode):
-            raise operationerrfmt(space.w_TypeError,
-                                  "string argument expected, got '%s'",
-                                  space.type(self).getname(space, '?'))
-        self._check_closed()
-        string = space.unicode_w(w_obj)
-        size = len(string)
-        if size:
-            self.write(string)
-        return space.wrap(size)
-
-    @unwrap_spec('self', ObjSpace)
-    def getvalue_w(self, space):
-        self._check_initialized()
-        self._check_closed()
-        return space.wrap(u''.join(self.buf))
-
-W_StringIO.typedef = TypeDef(
-    'StringIO', W_TextIOBase.typedef,
-    __new__  = interp2app(W_StringIO.descr_new.im_func),
-    write=interp2app(W_StringIO.write_w),
-    getvalue=interp2app(W_StringIO.getvalue_w),
-    )
-
+from pypy.module._io.interp_io import W_TextIOBase
+from pypy.interpreter.typedef import TypeDef
+from pypy.interpreter.gateway import interp2app, unwrap_spec
+from pypy.interpreter.error import operationerrfmt
+from pypy.interpreter.baseobjspace import ObjSpace, W_Root
+
+class W_StringIO(W_TextIOBase):
+    def __init__(self):
+        self.buf = []
+        self.pos = 0
+
+    def _check_closed(self):
+        pass
+    def _check_initialized(self):
+        pass
+
+    @unwrap_spec(ObjSpace, W_Root)
+    def descr_new(space, w_subtype):
+        self = space.allocate_instance(W_StringIO, w_subtype)
+        W_StringIO.__init__(self)
+        return space.wrap(self)
+
+    def resize_buffer(self, newlength):
+        if len(self.buf) > newlength:
+            self.buf = self.buf[:newlength]
+        if len(self.buf) < newlength:
+            self.buf.extend([u'\0'] * (newlength - len(self.buf)))
+
+    def write(self, string):
+        # XXX self.decoder
+        decoded = string
+        # XXX writenl
+
+        length = len(decoded)
+        if self.pos + length > len(self.buf):
+            self.resize_buffer(self.pos + length)
+
+        for i in range(length):
+            self.buf[self.pos + i] = string[i]
+        self.pos += length
+
+    @unwrap_spec('self', ObjSpace, W_Root)
+    def write_w(self, space, w_obj):
+        self._check_initialized()
+        if not space.isinstance_w(w_obj, space.w_unicode):
+            raise operationerrfmt(space.w_TypeError,
+                                  "string argument expected, got '%s'",
+                                  space.type(self).getname(space, '?'))
+        self._check_closed()
+        string = space.unicode_w(w_obj)
+        size = len(string)
+        if size:
+            self.write(string)
+        return space.wrap(size)
+
+    @unwrap_spec('self', ObjSpace)
+    def getvalue_w(self, space):
+        self._check_initialized()
+        self._check_closed()
+        return space.wrap(u''.join(self.buf))
+
+W_StringIO.typedef = TypeDef(
+    'StringIO', W_TextIOBase.typedef,
+    __new__  = interp2app(W_StringIO.descr_new.im_func),
+    write=interp2app(W_StringIO.write_w),
+    getvalue=interp2app(W_StringIO.getvalue_w),
+    )
+

Modified: pypy/branch/fast-forward/pypy/module/_io/test/test_stringio.py
==============================================================================
--- pypy/branch/fast-forward/pypy/module/_io/test/test_stringio.py	(original)
+++ pypy/branch/fast-forward/pypy/module/_io/test/test_stringio.py	Mon Oct  4 23:11:48 2010
@@ -1,7 +1,7 @@
-class AppTestStringIO:
-    def test_stringio(self):
-        import io
-        sio = io.StringIO()
-        sio.write(u'Hello ')
-        sio.write(u'world')
-        assert sio.getvalue() == u'Hello world'
+class AppTestStringIO:
+    def test_stringio(self):
+        import io
+        sio = io.StringIO()
+        sio.write(u'Hello ')
+        sio.write(u'world')
+        assert sio.getvalue() == u'Hello world'

Modified: pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py
==============================================================================
--- pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py	(original)
+++ pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py	Mon Oct  4 23:11:48 2010
@@ -1,48 +1,48 @@
-import py
-import sys
-from pypy.conftest import gettestobjspace
-
-class TestConnection:
-    def test_simple(self):
-        from pypy.module._multiprocessing import interp_connection
-
-class AppTestConnection:
-    def setup_class(cls):
-        space = gettestobjspace(usemodules=('_multiprocessing', 'thread'))
-        cls.space = space
-        if sys.platform == "win32":
-            # stubs for some modules,
-            # just for multiprocessing to import correctly.
-            w_modules = space.sys.get('modules')
-            space.setitem(w_modules, space.wrap('msvcrt'), space.sys)
-            space.setitem(w_modules, space.wrap('_subprocess'), space.sys)
-
-        # import multiprocessing once
-        space.appexec([], """(): import multiprocessing""")
-
-    def test_winpipe_connection(self):
-        import sys
-        if sys.platform != "win32":
-            skip("win32 only")
-
-        import multiprocessing
-        rhandle, whandle = multiprocessing.Pipe()
-
-        obj = [1, 2.0, "hello"]
-        whandle.send(obj)
-        obj2 = rhandle.recv()
-        assert obj == obj2
-
-    def test_ospipe_connection(self):
-        import _multiprocessing
-        import os
-        fd1, fd2 = os.pipe()
-        rhandle = _multiprocessing.Connection(fd1, writable=False)
-        whandle = _multiprocessing.Connection(fd2, readable=False)
-
-        obj = [1, 2.0, "hello"]
-        whandle.send(obj)
-        obj2 = rhandle.recv()
-        assert obj == obj2
-
-
+import py
+import sys
+from pypy.conftest import gettestobjspace
+
+class TestConnection:
+    def test_simple(self):
+        from pypy.module._multiprocessing import interp_connection
+
+class AppTestConnection:
+    def setup_class(cls):
+        space = gettestobjspace(usemodules=('_multiprocessing', 'thread'))
+        cls.space = space
+        if sys.platform == "win32":
+            # stubs for some modules,
+            # just for multiprocessing to import correctly.
+            w_modules = space.sys.get('modules')
+            space.setitem(w_modules, space.wrap('msvcrt'), space.sys)
+            space.setitem(w_modules, space.wrap('_subprocess'), space.sys)
+
+        # import multiprocessing once
+        space.appexec([], """(): import multiprocessing""")
+
+    def test_winpipe_connection(self):
+        import sys
+        if sys.platform != "win32":
+            skip("win32 only")
+
+        import multiprocessing
+        rhandle, whandle = multiprocessing.Pipe()
+
+        obj = [1, 2.0, "hello"]
+        whandle.send(obj)
+        obj2 = rhandle.recv()
+        assert obj == obj2
+
+    def test_ospipe_connection(self):
+        import _multiprocessing
+        import os
+        fd1, fd2 = os.pipe()
+        rhandle = _multiprocessing.Connection(fd1, writable=False)
+        whandle = _multiprocessing.Connection(fd2, readable=False)
+
+        obj = [1, 2.0, "hello"]
+        whandle.send(obj)
+        obj2 = rhandle.recv()
+        assert obj == obj2
+
+



More information about the Pypy-commit mailing list