Python code coverage for Lib/bsddb/dbrecio.py

#	count	content
1	n/a
2	n/a	"""
3	n/a	File-like objects that read from or write to a bsddb record.
4	n/a
5	n/a	This implements (nearly) all stdio methods.
6	n/a
7	n/a	f = DBRecIO(db, key, txn=None)
8	n/a	f.close() # explicitly release resources held
9	n/a	flag = f.isatty() # always false
10	n/a	pos = f.tell() # get current position
11	n/a	f.seek(pos) # set current position
12	n/a	f.seek(pos, mode) # mode 0: absolute; 1: relative; 2: relative to EOF
13	n/a	buf = f.read() # read until EOF
14	n/a	buf = f.read(n) # read up to n bytes
15	n/a	f.truncate([size]) # truncate file at to at most size (default: current pos)
16	n/a	f.write(buf) # write at current position
17	n/a	f.writelines(list) # for line in list: f.write(line)
18	n/a
19	n/a	Notes:
20	n/a	- fileno() is left unimplemented so that code which uses it triggers
21	n/a	an exception early.
22	n/a	- There's a simple test set (see end of this file) - not yet updated
23	n/a	for DBRecIO.
24	n/a	- readline() is not implemented yet.
25	n/a
26	n/a
27	n/a	From:
28	n/a	Itamar Shtull-Trauring <itamar@maxnm.com>
29	n/a	"""
30	n/a
31	n/a	import errno
32	n/a	import string
33	n/a
34	n/a	class DBRecIO:
35	n/a	def __init__(self, db, key, txn=None):
36	n/a	self.db = db
37	n/a	self.key = key
38	n/a	self.txn = txn
39	n/a	self.len = None
40	n/a	self.pos = 0
41	n/a	self.closed = 0
42	n/a	self.softspace = 0
43	n/a
44	n/a	def close(self):
45	n/a	if not self.closed:
46	n/a	self.closed = 1
47	n/a	del self.db, self.txn
48	n/a
49	n/a	def isatty(self):
50	n/a	if self.closed:
51	n/a	raise ValueError, "I/O operation on closed file"
52	n/a	return 0
53	n/a
54	n/a	def seek(self, pos, mode = 0):
55	n/a	if self.closed:
56	n/a	raise ValueError, "I/O operation on closed file"
57	n/a	if mode == 1:
58	n/a	pos = pos + self.pos
59	n/a	elif mode == 2:
60	n/a	pos = pos + self.len
61	n/a	self.pos = max(0, pos)
62	n/a
63	n/a	def tell(self):
64	n/a	if self.closed:
65	n/a	raise ValueError, "I/O operation on closed file"
66	n/a	return self.pos
67	n/a
68	n/a	def read(self, n = -1):
69	n/a	if self.closed:
70	n/a	raise ValueError, "I/O operation on closed file"
71	n/a	if n < 0:
72	n/a	newpos = self.len
73	n/a	else:
74	n/a	newpos = min(self.pos+n, self.len)
75	n/a
76	n/a	dlen = newpos - self.pos
77	n/a
78	n/a	r = self.db.get(self.key, txn=self.txn, dlen=dlen, doff=self.pos)
79	n/a	self.pos = newpos
80	n/a	return r
81	n/a
82	n/a	__fixme = """
83	n/a	def readline(self, length=None):
84	n/a	if self.closed:
85	n/a	raise ValueError, "I/O operation on closed file"
86	n/a	if self.buflist:
87	n/a	self.buf = self.buf + string.joinfields(self.buflist, '')
88	n/a	self.buflist = []
89	n/a	i = string.find(self.buf, '\n', self.pos)
90	n/a	if i < 0:
91	n/a	newpos = self.len
92	n/a	else:
93	n/a	newpos = i+1
94	n/a	if length is not None:
95	n/a	if self.pos + length < newpos:
96	n/a	newpos = self.pos + length
97	n/a	r = self.buf[self.pos:newpos]
98	n/a	self.pos = newpos
99	n/a	return r
100	n/a
101	n/a	def readlines(self, sizehint = 0):
102	n/a	total = 0
103	n/a	lines = []
104	n/a	line = self.readline()
105	n/a	while line:
106	n/a	lines.append(line)
107	n/a	total += len(line)
108	n/a	if 0 < sizehint <= total:
109	n/a	break
110	n/a	line = self.readline()
111	n/a	return lines
112	n/a	"""
113	n/a
114	n/a	def truncate(self, size=None):
115	n/a	if self.closed:
116	n/a	raise ValueError, "I/O operation on closed file"
117	n/a	if size is None:
118	n/a	size = self.pos
119	n/a	elif size < 0:
120	n/a	raise IOError(errno.EINVAL,
121	n/a	"Negative size not allowed")
122	n/a	elif size < self.pos:
123	n/a	self.pos = size
124	n/a	self.db.put(self.key, "", txn=self.txn, dlen=self.len-size, doff=size)
125	n/a
126	n/a	def write(self, s):
127	n/a	if self.closed:
128	n/a	raise ValueError, "I/O operation on closed file"
129	n/a	if not s: return
130	n/a	if self.pos > self.len:
131	n/a	self.buflist.append('\0'*(self.pos - self.len))
132	n/a	self.len = self.pos
133	n/a	newpos = self.pos + len(s)
134	n/a	self.db.put(self.key, s, txn=self.txn, dlen=len(s), doff=self.pos)
135	n/a	self.pos = newpos
136	n/a
137	n/a	def writelines(self, list):
138	n/a	self.write(string.joinfields(list, ''))
139	n/a
140	n/a	def flush(self):
141	n/a	if self.closed:
142	n/a	raise ValueError, "I/O operation on closed file"
143	n/a
144	n/a
145	n/a	"""
146	n/a	# A little test suite
147	n/a
148	n/a	def _test():
149	n/a	import sys
150	n/a	if sys.argv[1:]:
151	n/a	file = sys.argv[1]
152	n/a	else:
153	n/a	file = '/etc/passwd'
154	n/a	lines = open(file, 'r').readlines()
155	n/a	text = open(file, 'r').read()
156	n/a	f = StringIO()
157	n/a	for line in lines[:-2]:
158	n/a	f.write(line)
159	n/a	f.writelines(lines[-2:])
160	n/a	if f.getvalue() != text:
161	n/a	raise RuntimeError, 'write failed'
162	n/a	length = f.tell()
163	n/a	print 'File length =', length
164	n/a	f.seek(len(lines[0]))
165	n/a	f.write(lines[1])
166	n/a	f.seek(0)
167	n/a	print 'First line =', repr(f.readline())
168	n/a	here = f.tell()
169	n/a	line = f.readline()
170	n/a	print 'Second line =', repr(line)
171	n/a	f.seek(-len(line), 1)
172	n/a	line2 = f.read(len(line))
173	n/a	if line != line2:
174	n/a	raise RuntimeError, 'bad result after seek back'
175	n/a	f.seek(len(line2), 1)
176	n/a	list = f.readlines()
177	n/a	line = list[-1]
178	n/a	f.seek(f.tell() - len(line))
179	n/a	line2 = f.read()
180	n/a	if line != line2:
181	n/a	raise RuntimeError, 'bad result after seek back from EOF'
182	n/a	print 'Read', len(list), 'more lines'
183	n/a	print 'File length =', f.tell()
184	n/a	if f.tell() != length:
185	n/a	raise RuntimeError, 'bad length'
186	n/a	f.close()
187	n/a
188	n/a	if __name__ == '__main__':
189	n/a	_test()
190	n/a	"""