Python code coverage for Lib/test/test_email/test_policy.py

#	count	content
1	n/a	import io
2	n/a	import types
3	n/a	import textwrap
4	n/a	import unittest
5	n/a	import email.policy
6	n/a	import email.parser
7	n/a	import email.generator
8	n/a	import email.message
9	n/a	from email import headerregistry
10	n/a
11	n/a	def make_defaults(base_defaults, differences):
12	n/a	defaults = base_defaults.copy()
13	n/a	defaults.update(differences)
14	n/a	return defaults
15	n/a
16	n/a	class PolicyAPITests(unittest.TestCase):
17	n/a
18	n/a	longMessage = True
19	n/a
20	n/a	# Base default values.
21	n/a	compat32_defaults = {
22	n/a	'max_line_length': 78,
23	n/a	'linesep': '\n',
24	n/a	'cte_type': '8bit',
25	n/a	'raise_on_defect': False,
26	n/a	'mangle_from_': True,
27	n/a	'message_factory': None,
28	n/a	}
29	n/a	# These default values are the ones set on email.policy.default.
30	n/a	# If any of these defaults change, the docs must be updated.
31	n/a	policy_defaults = compat32_defaults.copy()
32	n/a	policy_defaults.update({
33	n/a	'utf8': False,
34	n/a	'raise_on_defect': False,
35	n/a	'header_factory': email.policy.EmailPolicy.header_factory,
36	n/a	'refold_source': 'long',
37	n/a	'content_manager': email.policy.EmailPolicy.content_manager,
38	n/a	'mangle_from_': False,
39	n/a	'message_factory': email.message.EmailMessage,
40	n/a	})
41	n/a
42	n/a	# For each policy under test, we give here what we expect the defaults to
43	n/a	# be for that policy. The second argument to make defaults is the
44	n/a	# difference between the base defaults and that for the particular policy.
45	n/a	new_policy = email.policy.EmailPolicy()
46	n/a	policies = {
47	n/a	email.policy.compat32: make_defaults(compat32_defaults, {}),
48	n/a	email.policy.default: make_defaults(policy_defaults, {}),
49	n/a	email.policy.SMTP: make_defaults(policy_defaults,
50	n/a	{'linesep': '\r\n'}),
51	n/a	email.policy.SMTPUTF8: make_defaults(policy_defaults,
52	n/a	{'linesep': '\r\n',
53	n/a	'utf8': True}),
54	n/a	email.policy.HTTP: make_defaults(policy_defaults,
55	n/a	{'linesep': '\r\n',
56	n/a	'max_line_length': None}),
57	n/a	email.policy.strict: make_defaults(policy_defaults,
58	n/a	{'raise_on_defect': True}),
59	n/a	new_policy: make_defaults(policy_defaults, {}),
60	n/a	}
61	n/a	# Creating a new policy creates a new header factory. There is a test
62	n/a	# later that proves this.
63	n/a	policies[new_policy]['header_factory'] = new_policy.header_factory
64	n/a
65	n/a	def test_defaults(self):
66	n/a	for policy, expected in self.policies.items():
67	n/a	for attr, value in expected.items():
68	n/a	with self.subTest(policy=policy, attr=attr):
69	n/a	self.assertEqual(getattr(policy, attr), value,
70	n/a	("change {} docs/docstrings if defaults have "
71	n/a	"changed").format(policy))
72	n/a
73	n/a	def test_all_attributes_covered(self):
74	n/a	for policy, expected in self.policies.items():
75	n/a	for attr in dir(policy):
76	n/a	with self.subTest(policy=policy, attr=attr):
77	n/a	if (attr.startswith('_') or
78	n/a	isinstance(getattr(email.policy.EmailPolicy, attr),
79	n/a	types.FunctionType)):
80	n/a	continue
81	n/a	else:
82	n/a	self.assertIn(attr, expected,
83	n/a	"{} is not fully tested".format(attr))
84	n/a
85	n/a	def test_abc(self):
86	n/a	with self.assertRaises(TypeError) as cm:
87	n/a	email.policy.Policy()
88	n/a	msg = str(cm.exception)
89	n/a	abstract_methods = ('fold',
90	n/a	'fold_binary',
91	n/a	'header_fetch_parse',
92	n/a	'header_source_parse',
93	n/a	'header_store_parse')
94	n/a	for method in abstract_methods:
95	n/a	self.assertIn(method, msg)
96	n/a
97	n/a	def test_policy_is_immutable(self):
98	n/a	for policy, defaults in self.policies.items():
99	n/a	for attr in defaults:
100	n/a	with self.assertRaisesRegex(AttributeError, attr+".*read-only"):
101	n/a	setattr(policy, attr, None)
102	n/a	with self.assertRaisesRegex(AttributeError, 'no attribute.*foo'):
103	n/a	policy.foo = None
104	n/a
105	n/a	def test_set_policy_attrs_when_cloned(self):
106	n/a	# None of the attributes has a default value of None, so we set them
107	n/a	# all to None in the clone call and check that it worked.
108	n/a	for policyclass, defaults in self.policies.items():
109	n/a	testattrdict = {attr: None for attr in defaults}
110	n/a	policy = policyclass.clone(**testattrdict)
111	n/a	for attr in defaults:
112	n/a	self.assertIsNone(getattr(policy, attr))
113	n/a
114	n/a	def test_reject_non_policy_keyword_when_called(self):
115	n/a	for policyclass in self.policies:
116	n/a	with self.assertRaises(TypeError):
117	n/a	policyclass(this_keyword_should_not_be_valid=None)
118	n/a	with self.assertRaises(TypeError):
119	n/a	policyclass(newtline=None)
120	n/a
121	n/a	def test_policy_addition(self):
122	n/a	expected = self.policy_defaults.copy()
123	n/a	p1 = email.policy.default.clone(max_line_length=100)
124	n/a	p2 = email.policy.default.clone(max_line_length=50)
125	n/a	added = p1 + p2
126	n/a	expected.update(max_line_length=50)
127	n/a	for attr, value in expected.items():
128	n/a	self.assertEqual(getattr(added, attr), value)
129	n/a	added = p2 + p1
130	n/a	expected.update(max_line_length=100)
131	n/a	for attr, value in expected.items():
132	n/a	self.assertEqual(getattr(added, attr), value)
133	n/a	added = added + email.policy.default
134	n/a	for attr, value in expected.items():
135	n/a	self.assertEqual(getattr(added, attr), value)
136	n/a
137	n/a	def test_register_defect(self):
138	n/a	class Dummy:
139	n/a	def __init__(self):
140	n/a	self.defects = []
141	n/a	obj = Dummy()
142	n/a	defect = object()
143	n/a	policy = email.policy.EmailPolicy()
144	n/a	policy.register_defect(obj, defect)
145	n/a	self.assertEqual(obj.defects, [defect])
146	n/a	defect2 = object()
147	n/a	policy.register_defect(obj, defect2)
148	n/a	self.assertEqual(obj.defects, [defect, defect2])
149	n/a
150	n/a	class MyObj:
151	n/a	def __init__(self):
152	n/a	self.defects = []
153	n/a
154	n/a	class MyDefect(Exception):
155	n/a	pass
156	n/a
157	n/a	def test_handle_defect_raises_on_strict(self):
158	n/a	foo = self.MyObj()
159	n/a	defect = self.MyDefect("the telly is broken")
160	n/a	with self.assertRaisesRegex(self.MyDefect, "the telly is broken"):
161	n/a	email.policy.strict.handle_defect(foo, defect)
162	n/a
163	n/a	def test_handle_defect_registers_defect(self):
164	n/a	foo = self.MyObj()
165	n/a	defect1 = self.MyDefect("one")
166	n/a	email.policy.default.handle_defect(foo, defect1)
167	n/a	self.assertEqual(foo.defects, [defect1])
168	n/a	defect2 = self.MyDefect("two")
169	n/a	email.policy.default.handle_defect(foo, defect2)
170	n/a	self.assertEqual(foo.defects, [defect1, defect2])
171	n/a
172	n/a	class MyPolicy(email.policy.EmailPolicy):
173	n/a	defects = None
174	n/a	def __init__(self, args, *kw):
175	n/a	super().__init__(args, defects=[], *kw)
176	n/a	def register_defect(self, obj, defect):
177	n/a	self.defects.append(defect)
178	n/a
179	n/a	def test_overridden_register_defect_still_raises(self):
180	n/a	foo = self.MyObj()
181	n/a	defect = self.MyDefect("the telly is broken")
182	n/a	with self.assertRaisesRegex(self.MyDefect, "the telly is broken"):
183	n/a	self.MyPolicy(raise_on_defect=True).handle_defect(foo, defect)
184	n/a
185	n/a	def test_overridden_register_defect_works(self):
186	n/a	foo = self.MyObj()
187	n/a	defect1 = self.MyDefect("one")
188	n/a	my_policy = self.MyPolicy()
189	n/a	my_policy.handle_defect(foo, defect1)
190	n/a	self.assertEqual(my_policy.defects, [defect1])
191	n/a	self.assertEqual(foo.defects, [])
192	n/a	defect2 = self.MyDefect("two")
193	n/a	my_policy.handle_defect(foo, defect2)
194	n/a	self.assertEqual(my_policy.defects, [defect1, defect2])
195	n/a	self.assertEqual(foo.defects, [])
196	n/a
197	n/a	def test_default_header_factory(self):
198	n/a	h = email.policy.default.header_factory('Test', 'test')
199	n/a	self.assertEqual(h.name, 'Test')
200	n/a	self.assertIsInstance(h, headerregistry.UnstructuredHeader)
201	n/a	self.assertIsInstance(h, headerregistry.BaseHeader)
202	n/a
203	n/a	class Foo:
204	n/a	parse = headerregistry.UnstructuredHeader.parse
205	n/a
206	n/a	def test_each_Policy_gets_unique_factory(self):
207	n/a	policy1 = email.policy.EmailPolicy()
208	n/a	policy2 = email.policy.EmailPolicy()
209	n/a	policy1.header_factory.map_to_type('foo', self.Foo)
210	n/a	h = policy1.header_factory('foo', 'test')
211	n/a	self.assertIsInstance(h, self.Foo)
212	n/a	self.assertNotIsInstance(h, headerregistry.UnstructuredHeader)
213	n/a	h = policy2.header_factory('foo', 'test')
214	n/a	self.assertNotIsInstance(h, self.Foo)
215	n/a	self.assertIsInstance(h, headerregistry.UnstructuredHeader)
216	n/a
217	n/a	def test_clone_copies_factory(self):
218	n/a	policy1 = email.policy.EmailPolicy()
219	n/a	policy2 = policy1.clone()
220	n/a	policy1.header_factory.map_to_type('foo', self.Foo)
221	n/a	h = policy1.header_factory('foo', 'test')
222	n/a	self.assertIsInstance(h, self.Foo)
223	n/a	h = policy2.header_factory('foo', 'test')
224	n/a	self.assertIsInstance(h, self.Foo)
225	n/a
226	n/a	def test_new_factory_overrides_default(self):
227	n/a	mypolicy = email.policy.EmailPolicy()
228	n/a	myfactory = mypolicy.header_factory
229	n/a	newpolicy = mypolicy + email.policy.strict
230	n/a	self.assertEqual(newpolicy.header_factory, myfactory)
231	n/a	newpolicy = email.policy.strict + mypolicy
232	n/a	self.assertEqual(newpolicy.header_factory, myfactory)
233	n/a
234	n/a	def test_adding_default_policies_preserves_default_factory(self):
235	n/a	newpolicy = email.policy.default + email.policy.strict
236	n/a	self.assertEqual(newpolicy.header_factory,
237	n/a	email.policy.EmailPolicy.header_factory)
238	n/a	self.assertEqual(newpolicy.__dict__, {'raise_on_defect': True})
239	n/a
240	n/a	# XXX: Need subclassing tests.
241	n/a	# For adding subclassed objects, make sure the usual rules apply (subclass
242	n/a	# wins), but that the order still works (right overrides left).
243	n/a
244	n/a
245	n/a	class TestException(Exception):
246	n/a	pass
247	n/a
248	n/a	class TestPolicyPropagation(unittest.TestCase):
249	n/a
250	n/a	# The abstract methods are used by the parser but not by the wrapper
251	n/a	# functions that call it, so if the exception gets raised we know that the
252	n/a	# policy was actually propagated all the way to feedparser.
253	n/a	class MyPolicy(email.policy.Policy):
254	n/a	def badmethod(self, args, *kw):
255	n/a	raise TestException("test")
256	n/a	fold = fold_binary = header_fetch_parser = badmethod
257	n/a	header_source_parse = header_store_parse = badmethod
258	n/a
259	n/a	def test_message_from_string(self):
260	n/a	with self.assertRaisesRegex(TestException, "^test$"):
261	n/a	email.message_from_string("Subject: test\n\n",
262	n/a	policy=self.MyPolicy)
263	n/a
264	n/a	def test_message_from_bytes(self):
265	n/a	with self.assertRaisesRegex(TestException, "^test$"):
266	n/a	email.message_from_bytes(b"Subject: test\n\n",
267	n/a	policy=self.MyPolicy)
268	n/a
269	n/a	def test_message_from_file(self):
270	n/a	f = io.StringIO('Subject: test\n\n')
271	n/a	with self.assertRaisesRegex(TestException, "^test$"):
272	n/a	email.message_from_file(f, policy=self.MyPolicy)
273	n/a
274	n/a	def test_message_from_binary_file(self):
275	n/a	f = io.BytesIO(b'Subject: test\n\n')
276	n/a	with self.assertRaisesRegex(TestException, "^test$"):
277	n/a	email.message_from_binary_file(f, policy=self.MyPolicy)
278	n/a
279	n/a	# These are redundant, but we need them for black-box completeness.
280	n/a
281	n/a	def test_parser(self):
282	n/a	p = email.parser.Parser(policy=self.MyPolicy)
283	n/a	with self.assertRaisesRegex(TestException, "^test$"):
284	n/a	p.parsestr('Subject: test\n\n')
285	n/a
286	n/a	def test_bytes_parser(self):
287	n/a	p = email.parser.BytesParser(policy=self.MyPolicy)
288	n/a	with self.assertRaisesRegex(TestException, "^test$"):
289	n/a	p.parsebytes(b'Subject: test\n\n')
290	n/a
291	n/a	# Now that we've established that all the parse methods get the
292	n/a	# policy in to feedparser, we can use message_from_string for
293	n/a	# the rest of the propagation tests.
294	n/a
295	n/a	def _make_msg(self, source='Subject: test\n\n', policy=None):
296	n/a	self.policy = email.policy.default.clone() if policy is None else policy
297	n/a	return email.message_from_string(source, policy=self.policy)
298	n/a
299	n/a	def test_parser_propagates_policy_to_message(self):
300	n/a	msg = self._make_msg()
301	n/a	self.assertIs(msg.policy, self.policy)
302	n/a
303	n/a	def test_parser_propagates_policy_to_sub_messages(self):
304	n/a	msg = self._make_msg(textwrap.dedent("""\
305	n/a	Subject: mime test
306	n/a	MIME-Version: 1.0
307	n/a	Content-Type: multipart/mixed, boundary="XXX"
308	n/a
309	n/a	--XXX
310	n/a	Content-Type: text/plain
311	n/a
312	n/a	test
313	n/a	--XXX
314	n/a	Content-Type: text/plain
315	n/a
316	n/a	test2
317	n/a	--XXX--
318	n/a	"""))
319	n/a	for part in msg.walk():
320	n/a	self.assertIs(part.policy, self.policy)
321	n/a
322	n/a	def test_message_policy_propagates_to_generator(self):
323	n/a	msg = self._make_msg("Subject: test\nTo: foo\n\n",
324	n/a	policy=email.policy.default.clone(linesep='X'))
325	n/a	s = io.StringIO()
326	n/a	g = email.generator.Generator(s)
327	n/a	g.flatten(msg)
328	n/a	self.assertEqual(s.getvalue(), "Subject: testXTo: fooXX")
329	n/a
330	n/a	def test_message_policy_used_by_as_string(self):
331	n/a	msg = self._make_msg("Subject: test\nTo: foo\n\n",
332	n/a	policy=email.policy.default.clone(linesep='X'))
333	n/a	self.assertEqual(msg.as_string(), "Subject: testXTo: fooXX")
334	n/a
335	n/a
336	n/a	class TestConcretePolicies(unittest.TestCase):
337	n/a
338	n/a	def test_header_store_parse_rejects_newlines(self):
339	n/a	instance = email.policy.EmailPolicy()
340	n/a	self.assertRaises(ValueError,
341	n/a	instance.header_store_parse,
342	n/a	'From', 'spam\negg@foo.py')
343	n/a
344	n/a
345	n/a	if __name__ == '__main__':
346	n/a	unittest.main()