StringIO.py (5476B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\StringIO.py 2 try: 3 from errno import EINVAL 4 except ImportError: 5 EINVAL = 22 6 7 __all__ = ['StringIO'] 8 9 def _complain_ifclosed(closed): 10 if closed: 11 raise ValueError, 'I/O operation on closed file' 12 13 14 class StringIO: 15 16 def __init__(self, buf = ''): 17 if not isinstance(buf, basestring): 18 buf = str(buf) 19 self.buf = buf 20 self.len = len(buf) 21 self.buflist = [] 22 self.pos = 0 23 self.closed = False 24 self.softspace = 0 25 26 def __iter__(self): 27 return self 28 29 def next(self): 30 _complain_ifclosed(self.closed) 31 r = self.readline() 32 if not r: 33 raise StopIteration 34 return r 35 36 def close(self): 37 if not self.closed: 38 self.closed = True 39 del self.buf 40 del self.pos 41 42 def isatty(self): 43 _complain_ifclosed(self.closed) 44 return False 45 46 def seek(self, pos, mode = 0): 47 _complain_ifclosed(self.closed) 48 if self.buflist: 49 self.buf += ''.join(self.buflist) 50 self.buflist = [] 51 if mode == 1: 52 pos += self.pos 53 elif mode == 2: 54 pos += self.len 55 self.pos = max(0, pos) 56 57 def tell(self): 58 _complain_ifclosed(self.closed) 59 return self.pos 60 61 def read(self, n = -1): 62 _complain_ifclosed(self.closed) 63 if self.buflist: 64 self.buf += ''.join(self.buflist) 65 self.buflist = [] 66 if n is None or n < 0: 67 newpos = self.len 68 else: 69 newpos = min(self.pos + n, self.len) 70 r = self.buf[self.pos:newpos] 71 self.pos = newpos 72 return r 73 74 def readline(self, length = None): 75 _complain_ifclosed(self.closed) 76 if self.buflist: 77 self.buf += ''.join(self.buflist) 78 self.buflist = [] 79 i = self.buf.find('\n', self.pos) 80 if i < 0: 81 newpos = self.len 82 else: 83 newpos = i + 1 84 if length is not None and length > 0: 85 if self.pos + length < newpos: 86 newpos = self.pos + length 87 r = self.buf[self.pos:newpos] 88 self.pos = newpos 89 return r 90 91 def readlines(self, sizehint = 0): 92 total = 0 93 lines = [] 94 line = self.readline() 95 while line: 96 lines.append(line) 97 total += len(line) 98 if 0 < sizehint <= total: 99 break 100 line = self.readline() 101 102 return lines 103 104 def truncate(self, size = None): 105 _complain_ifclosed(self.closed) 106 if size is None: 107 size = self.pos 108 elif size < 0: 109 raise IOError(EINVAL, 'Negative size not allowed') 110 elif size < self.pos: 111 self.pos = size 112 self.buf = self.getvalue()[:size] 113 self.len = size 114 115 def write(self, s): 116 _complain_ifclosed(self.closed) 117 if not s: 118 return 119 if not isinstance(s, basestring): 120 s = str(s) 121 spos = self.pos 122 slen = self.len 123 if spos == slen: 124 self.buflist.append(s) 125 self.len = self.pos = spos + len(s) 126 return 127 if spos > slen: 128 self.buflist.append('\x00' * (spos - slen)) 129 slen = spos 130 newpos = spos + len(s) 131 if spos < slen: 132 if self.buflist: 133 self.buf += ''.join(self.buflist) 134 self.buflist = [self.buf[:spos], s, self.buf[newpos:]] 135 self.buf = '' 136 if newpos > slen: 137 slen = newpos 138 else: 139 self.buflist.append(s) 140 slen = newpos 141 self.len = slen 142 self.pos = newpos 143 144 def writelines(self, iterable): 145 write = self.write 146 for line in iterable: 147 write(line) 148 149 def flush(self): 150 _complain_ifclosed(self.closed) 151 152 def getvalue(self): 153 if self.buflist: 154 self.buf += ''.join(self.buflist) 155 self.buflist = [] 156 return self.buf 157 158 159 def test(): 160 import sys 161 if sys.argv[1:]: 162 file = sys.argv[1] 163 else: 164 file = '/etc/passwd' 165 lines = open(file, 'r').readlines() 166 text = open(file, 'r').read() 167 f = StringIO() 168 for line in lines[:-2]: 169 f.write(line) 170 171 f.writelines(lines[-2:]) 172 if f.getvalue() != text: 173 raise RuntimeError, 'write failed' 174 length = f.tell() 175 print 'File length =', length 176 f.seek(len(lines[0])) 177 f.write(lines[1]) 178 f.seek(0) 179 print 'First line =', repr(f.readline()) 180 print 'Position =', f.tell() 181 line = f.readline() 182 print 'Second line =', repr(line) 183 f.seek(-len(line), 1) 184 line2 = f.read(len(line)) 185 if line != line2: 186 raise RuntimeError, 'bad result after seek back' 187 f.seek(len(line2), 1) 188 list = f.readlines() 189 line = list[-1] 190 f.seek(f.tell() - len(line)) 191 line2 = f.read() 192 if line != line2: 193 raise RuntimeError, 'bad result after seek back from EOF' 194 print 'Read', len(list), 'more lines' 195 print 'File length =', f.tell() 196 if f.tell() != length: 197 raise RuntimeError, 'bad length' 198 f.truncate(length / 2) 199 f.seek(0, 2) 200 print 'Truncated length =', f.tell() 201 if f.tell() != length / 2: 202 raise RuntimeError, 'truncate did not adjust length' 203 f.close() 204 205 206 if __name__ == '__main__': 207 test()