Python code coverage for Lib/mimetools.py

#	count	content
1	1	"""Various tools used by MIME-reading or MIME-writing programs."""
2	n/a
3	n/a
4	1	import os
5	1	import sys
6	1	import tempfile
7	1	from warnings import filterwarnings, catch_warnings
8	1	with catch_warnings():
9	1	if sys.py3kwarning:
10	0	filterwarnings("ignore", ".*rfc822 has been removed", DeprecationWarning)
11	1	import rfc822
12	n/a
13	1	from warnings import warnpy3k
14	1	warnpy3k("in 3.x, mimetools has been removed in favor of the email package",
15	1	stacklevel=2)
16	n/a
17	1	__all__ = ["Message","choose_boundary","encode","decode","copyliteral",
18	1	"copybinary"]
19	n/a
20	2	class Message(rfc822.Message):
21	n/a	"""A derived class of rfc822.Message that knows about MIME headers and
22	1	contains some hooks for decoding encoded and multipart messages."""
23	n/a
24	1	def __init__(self, fp, seekable = 1):
25	437	rfc822.Message.__init__(self, fp, seekable)
26	n/a	self.encodingheader = \
27	437	self.getheader('content-transfer-encoding')
28	n/a	self.typeheader = \
29	437	self.getheader('content-type')
30	437	self.parsetype()
31	437	self.parseplist()
32	n/a
33	1	def parsetype(self):
34	437	str = self.typeheader
35	437	if str is None:
36	252	str = 'text/plain'
37	437	if ';' in str:
38	16	i = str.index(';')
39	16	self.plisttext = str[i:]
40	16	str = str[:i]
41	n/a	else:
42	421	self.plisttext = ''
43	437	fields = str.split('/')
44	1311	for i in range(len(fields)):
45	874	fields[i] = fields[i].strip().lower()
46	437	self.type = '/'.join(fields)
47	437	self.maintype = fields[0]
48	437	self.subtype = '/'.join(fields[1:])
49	n/a
50	1	def parseplist(self):
51	437	str = self.plisttext
52	437	self.plist = []
53	456	while str[:1] == ';':
54	19	str = str[1:]
55	19	if ';' in str:
56	n/a	# XXX Should parse quotes!
57	3	end = str.index(';')
58	n/a	else:
59	16	end = len(str)
60	19	f = str[:end]
61	19	if '=' in f:
62	19	i = f.index('=')
63	n/a	f = f[:i].strip().lower() + \
64	19	'=' + f[i+1:].strip()
65	19	self.plist.append(f.strip())
66	19	str = str[end:]
67	n/a
68	1	def getplist(self):
69	1	return self.plist
70	n/a
71	1	def getparam(self, name):
72	5	name = name.lower() + '='
73	5	n = len(name)
74	8	for p in self.plist:
75	7	if p[:n] == name:
76	4	return rfc822.unquote(p[n:])
77	1	return None
78	n/a
79	1	def getparamnames(self):
80	1	result = []
81	3	for p in self.plist:
82	2	i = p.find('=')
83	2	if i >= 0:
84	2	result.append(p[:i].lower())
85	1	return result
86	n/a
87	1	def getencoding(self):
88	3	if self.encodingheader is None:
89	2	return '7bit'
90	1	return self.encodingheader.lower()
91	n/a
92	1	def gettype(self):
93	1	return self.type
94	n/a
95	1	def getmaintype(self):
96	6	return self.maintype
97	n/a
98	1	def getsubtype(self):
99	3	return self.subtype
100	n/a
101	n/a
102	n/a
103	n/a
104	n/a	# Utility functions
105	n/a	# -----------------
106	n/a
107	1	try:
108	1	import thread
109	0	except ImportError:
110	0	import dummy_thread as thread
111	1	_counter_lock = thread.allocate_lock()
112	1	del thread
113	n/a
114	1	_counter = 0
115	1	def _get_next_counter():
116	n/a	global _counter
117	100	_counter_lock.acquire()
118	100	_counter += 1
119	100	result = _counter
120	100	_counter_lock.release()
121	100	return result
122	n/a
123	1	_prefix = None
124	n/a
125	1	def choose_boundary():
126	n/a	"""Return a string usable as a multipart boundary.
127	n/a
128	n/a	The string chosen is unique within a single program run, and
129	n/a	incorporates the user id (if available), process id (if available),
130	n/a	and current time. So it's very unlikely the returned string appears
131	n/a	in message text, but there's no guarantee.
132	n/a
133	n/a	The boundary contains dots so you have to quote it in the header."""
134	n/a
135	n/a	global _prefix
136	100	import time
137	100	if _prefix is None:
138	1	import socket
139	1	try:
140	1	hostid = socket.gethostbyname(socket.gethostname())
141	0	except socket.gaierror:
142	0	hostid = '127.0.0.1'
143	1	try:
144	1	uid = repr(os.getuid())
145	0	except AttributeError:
146	0	uid = '1'
147	1	try:
148	1	pid = repr(os.getpid())
149	0	except AttributeError:
150	0	pid = '1'
151	1	_prefix = hostid + '.' + uid + '.' + pid
152	100	return "%s.%.3f.%d" % (_prefix, time.time(), _get_next_counter())
153	n/a
154	n/a
155	n/a	# Subroutines for decoding some common content-transfer-types
156	n/a
157	1	def decode(input, output, encoding):
158	n/a	"""Decode common content-transfer-encodings (base64, quopri, uuencode)."""
159	8	if encoding == 'base64':
160	1	import base64
161	1	return base64.decode(input, output)
162	7	if encoding == 'quoted-printable':
163	1	import quopri
164	1	return quopri.decode(input, output)
165	6	if encoding in ('uuencode', 'x-uuencode', 'uue', 'x-uue'):
166	4	import uu
167	4	return uu.decode(input, output)
168	2	if encoding in ('7bit', '8bit'):
169	2	return output.write(input.read())
170	0	if encoding in decodetab:
171	0	pipethrough(input, decodetab[encoding], output)
172	n/a	else:
173	0	raise ValueError, \
174	0	'unknown Content-Transfer-Encoding: %s' % encoding
175	n/a
176	1	def encode(input, output, encoding):
177	n/a	"""Encode common content-transfer-encodings (base64, quopri, uuencode)."""
178	8	if encoding == 'base64':
179	1	import base64
180	1	return base64.encode(input, output)
181	7	if encoding == 'quoted-printable':
182	1	import quopri
183	1	return quopri.encode(input, output, 0)
184	6	if encoding in ('uuencode', 'x-uuencode', 'uue', 'x-uue'):
185	4	import uu
186	4	return uu.encode(input, output)
187	2	if encoding in ('7bit', '8bit'):
188	2	return output.write(input.read())
189	0	if encoding in encodetab:
190	0	pipethrough(input, encodetab[encoding], output)
191	n/a	else:
192	0	raise ValueError, \
193	0	'unknown Content-Transfer-Encoding: %s' % encoding
194	n/a
195	n/a	# The following is no longer used for standard encodings
196	n/a
197	n/a	# XXX This requires that uudecode and mmencode are in $PATH
198	n/a
199	n/a	uudecode_pipe = '''(
200	n/a	TEMP=/tmp/@uu.$$
201	n/a	sed "s%^begin [0-7][0-7]* .*%begin 600 $TEMP%" \| uudecode
202	n/a	cat $TEMP
203	n/a	rm $TEMP
204	1	)'''
205	n/a
206	1	decodetab = {
207	1	'uuencode': uudecode_pipe,
208	1	'x-uuencode': uudecode_pipe,
209	1	'uue': uudecode_pipe,
210	1	'x-uue': uudecode_pipe,
211	1	'quoted-printable': 'mmencode -u -q',
212	1	'base64': 'mmencode -u -b',
213	n/a	}
214	n/a
215	1	encodetab = {
216	1	'x-uuencode': 'uuencode tempfile',
217	1	'uuencode': 'uuencode tempfile',
218	1	'x-uue': 'uuencode tempfile',
219	1	'uue': 'uuencode tempfile',
220	1	'quoted-printable': 'mmencode -q',
221	1	'base64': 'mmencode -b',
222	n/a	}
223	n/a
224	1	def pipeto(input, command):
225	0	pipe = os.popen(command, 'w')
226	0	copyliteral(input, pipe)
227	0	pipe.close()
228	n/a
229	1	def pipethrough(input, command, output):
230	0	(fd, tempname) = tempfile.mkstemp()
231	0	temp = os.fdopen(fd, 'w')
232	0	copyliteral(input, temp)
233	0	temp.close()
234	0	pipe = os.popen(command + ' <' + tempname, 'r')
235	0	copybinary(pipe, output)
236	0	pipe.close()
237	0	os.unlink(tempname)
238	n/a
239	1	def copyliteral(input, output):
240	0	while 1:
241	0	line = input.readline()
242	0	if not line: break
243	0	output.write(line)
244	n/a
245	1	def copybinary(input, output):
246	0	BUFSIZE = 8192
247	0	while 1:
248	0	line = input.read(BUFSIZE)
249	0	if not line: break
250	0	output.write(line)