[pypy-svn] r15127 - in pypy/dist/pypy/tool: . test

[rxe at codespeak.net]

Author: rxe
Date: Tue Jul 26 16:21:21 2005
New Revision: 15127

Added:
   pypy/dist/pypy/tool/osfilewrapper.py
   pypy/dist/pypy/tool/test/test_osfilewrapper.py
Log:
Add a very simple os file wrapper.
  


Added: pypy/dist/pypy/tool/osfilewrapper.py
==============================================================================
--- (empty file)
+++ pypy/dist/pypy/tool/osfilewrapper.py	Tue Jul 26 16:21:21 2005
@@ -0,0 +1,36 @@
+import os
+
+class OsFileWrapper(object):
+    """ Very simple os file wrapper.
+    Note user is responsible for closing.
+    XXX Could add a simple buffer mechanism. """
+    
+    def __init__(self, fd):
+        self.fd = fd
+        
+    def read(self, expected):
+        readcount = 0
+        bufs = []
+        while readcount < expected:
+            # os.read will raise an error itself
+            buf = os.read(self.fd, expected)
+            readcount += len(buf)
+            bufs.append(buf)
+        return "".join(bufs)
+
+    def write(self, buf):
+        writecount = 0
+        towrite = len(buf)
+        while writecount < towrite:
+            # os.write will raise an error itself
+            writecount += os.write(self.fd, buf)
+            buf = buf[writecount:]
+
+    def close(self):
+        os.close(self.fd)
+
+    def create_wrapper(cls, filename, flag, mode=0777):
+        fd = os.open(filename, flag, mode)
+        return cls(fd)
+    
+    create_wrapper = classmethod(create_wrapper)

Added: pypy/dist/pypy/tool/test/test_osfilewrapper.py
==============================================================================
--- (empty file)
+++ pypy/dist/pypy/tool/test/test_osfilewrapper.py	Tue Jul 26 16:21:21 2005
@@ -0,0 +1,32 @@
+import autopath
+from pypy.tool.osfilewrapper import OsFileWrapper 
+from pypy.tool.udir import udir 
+import os
+
+def test_reads():
+    
+    p = str(udir.join('test.dat'))
+
+    # As we are testing file writes, only using udir to create a path 
+    buf = "1234567890"
+    f = open(p, "w")
+    f.write(buf)
+    f.close()
+
+    for ii in range(10):
+        f = OsFileWrapper.create_wrapper(p, os.O_RDONLY)
+        assert f.read(ii) == buf[:ii]
+    
+def test_writes_reads():
+
+    # As we are testing file writes, only using udir to create a path 
+    buf = "1234567890"
+    for ii in range(10):
+        p = str(udir.join('test.dat'))
+        f1 = OsFileWrapper.create_wrapper(p, os.O_WRONLY)
+        f1.write(buf[:ii])
+        f1.close()
+
+        f2 = OsFileWrapper.create_wrapper(p, os.O_RDONLY)
+        assert f2.read(ii) == buf[:ii]
+        f2.close()