Python code coverage for Lib/test/test_nntplib.py

#	count	content
1	n/a	import io
2	n/a	import socket
3	n/a	import datetime
4	n/a	import textwrap
5	n/a	import unittest
6	n/a	import functools
7	n/a	import contextlib
8	n/a	import os.path
9	n/a	from test import support
10	n/a	from nntplib import NNTP, GroupInfo
11	n/a	import nntplib
12	n/a	from unittest.mock import patch
13	n/a	try:
14	n/a	import ssl
15	n/a	except ImportError:
16	n/a	ssl = None
17	n/a	try:
18	n/a	import threading
19	n/a	except ImportError:
20	n/a	threading = None
21	n/a
22	n/a	TIMEOUT = 30
23	n/a	certfile = os.path.join(os.path.dirname(__file__), 'keycert3.pem')
24	n/a
25	n/a	# TODO:
26	n/a	# - test the `file` arg to more commands
27	n/a	# - test error conditions
28	n/a	# - test auth and `usenetrc`
29	n/a
30	n/a
31	n/a	class NetworkedNNTPTestsMixin:
32	n/a
33	n/a	def test_welcome(self):
34	n/a	welcome = self.server.getwelcome()
35	n/a	self.assertEqual(str, type(welcome))
36	n/a
37	n/a	def test_help(self):
38	n/a	resp, lines = self.server.help()
39	n/a	self.assertTrue(resp.startswith("100 "), resp)
40	n/a	for line in lines:
41	n/a	self.assertEqual(str, type(line))
42	n/a
43	n/a	def test_list(self):
44	n/a	resp, groups = self.server.list()
45	n/a	if len(groups) > 0:
46	n/a	self.assertEqual(GroupInfo, type(groups[0]))
47	n/a	self.assertEqual(str, type(groups[0].group))
48	n/a
49	n/a	def test_list_active(self):
50	n/a	resp, groups = self.server.list(self.GROUP_PAT)
51	n/a	if len(groups) > 0:
52	n/a	self.assertEqual(GroupInfo, type(groups[0]))
53	n/a	self.assertEqual(str, type(groups[0].group))
54	n/a
55	n/a	def test_unknown_command(self):
56	n/a	with self.assertRaises(nntplib.NNTPPermanentError) as cm:
57	n/a	self.server._shortcmd("XYZZY")
58	n/a	resp = cm.exception.response
59	n/a	self.assertTrue(resp.startswith("500 "), resp)
60	n/a
61	n/a	def test_newgroups(self):
62	n/a	# gmane gets a constant influx of new groups. In order not to stress
63	n/a	# the server too much, we choose a recent date in the past.
64	n/a	dt = datetime.date.today() - datetime.timedelta(days=7)
65	n/a	resp, groups = self.server.newgroups(dt)
66	n/a	if len(groups) > 0:
67	n/a	self.assertIsInstance(groups[0], GroupInfo)
68	n/a	self.assertIsInstance(groups[0].group, str)
69	n/a
70	n/a	def test_description(self):
71	n/a	def _check_desc(desc):
72	n/a	# Sanity checks
73	n/a	self.assertIsInstance(desc, str)
74	n/a	self.assertNotIn(self.GROUP_NAME, desc)
75	n/a	desc = self.server.description(self.GROUP_NAME)
76	n/a	_check_desc(desc)
77	n/a	# Another sanity check
78	n/a	self.assertIn("Python", desc)
79	n/a	# With a pattern
80	n/a	desc = self.server.description(self.GROUP_PAT)
81	n/a	_check_desc(desc)
82	n/a	# Shouldn't exist
83	n/a	desc = self.server.description("zk.brrtt.baz")
84	n/a	self.assertEqual(desc, '')
85	n/a
86	n/a	def test_descriptions(self):
87	n/a	resp, descs = self.server.descriptions(self.GROUP_PAT)
88	n/a	# 215 for LIST NEWSGROUPS, 282 for XGTITLE
89	n/a	self.assertTrue(
90	n/a	resp.startswith("215 ") or resp.startswith("282 "), resp)
91	n/a	self.assertIsInstance(descs, dict)
92	n/a	desc = descs[self.GROUP_NAME]
93	n/a	self.assertEqual(desc, self.server.description(self.GROUP_NAME))
94	n/a
95	n/a	def test_group(self):
96	n/a	result = self.server.group(self.GROUP_NAME)
97	n/a	self.assertEqual(5, len(result))
98	n/a	resp, count, first, last, group = result
99	n/a	self.assertEqual(group, self.GROUP_NAME)
100	n/a	self.assertIsInstance(count, int)
101	n/a	self.assertIsInstance(first, int)
102	n/a	self.assertIsInstance(last, int)
103	n/a	self.assertLessEqual(first, last)
104	n/a	self.assertTrue(resp.startswith("211 "), resp)
105	n/a
106	n/a	def test_date(self):
107	n/a	resp, date = self.server.date()
108	n/a	self.assertIsInstance(date, datetime.datetime)
109	n/a	# Sanity check
110	n/a	self.assertGreaterEqual(date.year, 1995)
111	n/a	self.assertLessEqual(date.year, 2030)
112	n/a
113	n/a	def _check_art_dict(self, art_dict):
114	n/a	# Some sanity checks for a field dictionary returned by OVER / XOVER
115	n/a	self.assertIsInstance(art_dict, dict)
116	n/a	# NNTP has 7 mandatory fields
117	n/a	self.assertGreaterEqual(art_dict.keys(),
118	n/a	{"subject", "from", "date", "message-id",
119	n/a	"references", ":bytes", ":lines"}
120	n/a	)
121	n/a	for v in art_dict.values():
122	n/a	self.assertIsInstance(v, (str, type(None)))
123	n/a
124	n/a	def test_xover(self):
125	n/a	resp, count, first, last, name = self.server.group(self.GROUP_NAME)
126	n/a	resp, lines = self.server.xover(last - 5, last)
127	n/a	if len(lines) == 0:
128	n/a	self.skipTest("no articles retrieved")
129	n/a	# The 'last' article is not necessarily part of the output (cancelled?)
130	n/a	art_num, art_dict = lines[0]
131	n/a	self.assertGreaterEqual(art_num, last - 5)
132	n/a	self.assertLessEqual(art_num, last)
133	n/a	self._check_art_dict(art_dict)
134	n/a
135	n/a	@unittest.skipIf(True, 'temporarily skipped until a permanent solution'
136	n/a	' is found for issue #28971')
137	n/a	def test_over(self):
138	n/a	resp, count, first, last, name = self.server.group(self.GROUP_NAME)
139	n/a	start = last - 10
140	n/a	# The "start-" article range form
141	n/a	resp, lines = self.server.over((start, None))
142	n/a	art_num, art_dict = lines[0]
143	n/a	self._check_art_dict(art_dict)
144	n/a	# The "start-end" article range form
145	n/a	resp, lines = self.server.over((start, last))
146	n/a	art_num, art_dict = lines[-1]
147	n/a	# The 'last' article is not necessarily part of the output (cancelled?)
148	n/a	self.assertGreaterEqual(art_num, start)
149	n/a	self.assertLessEqual(art_num, last)
150	n/a	self._check_art_dict(art_dict)
151	n/a	# XXX The "message_id" form is unsupported by gmane
152	n/a	# 503 Overview by message-ID unsupported
153	n/a
154	n/a	def test_xhdr(self):
155	n/a	resp, count, first, last, name = self.server.group(self.GROUP_NAME)
156	n/a	resp, lines = self.server.xhdr('subject', last)
157	n/a	for line in lines:
158	n/a	self.assertEqual(str, type(line[1]))
159	n/a
160	n/a	def check_article_resp(self, resp, article, art_num=None):
161	n/a	self.assertIsInstance(article, nntplib.ArticleInfo)
162	n/a	if art_num is not None:
163	n/a	self.assertEqual(article.number, art_num)
164	n/a	for line in article.lines:
165	n/a	self.assertIsInstance(line, bytes)
166	n/a	# XXX this could exceptionally happen...
167	n/a	self.assertNotIn(article.lines[-1], (b".", b".\n", b".\r\n"))
168	n/a
169	n/a	def test_article_head_body(self):
170	n/a	resp, count, first, last, name = self.server.group(self.GROUP_NAME)
171	n/a	# Try to find an available article
172	n/a	for art_num in (last, first, last - 1):
173	n/a	try:
174	n/a	resp, head = self.server.head(art_num)
175	n/a	except nntplib.NNTPTemporaryError as e:
176	n/a	if not e.response.startswith("423 "):
177	n/a	raise
178	n/a	# "423 No such article" => choose another one
179	n/a	continue
180	n/a	break
181	n/a	else:
182	n/a	self.skipTest("could not find a suitable article number")
183	n/a	self.assertTrue(resp.startswith("221 "), resp)
184	n/a	self.check_article_resp(resp, head, art_num)
185	n/a	resp, body = self.server.body(art_num)
186	n/a	self.assertTrue(resp.startswith("222 "), resp)
187	n/a	self.check_article_resp(resp, body, art_num)
188	n/a	resp, article = self.server.article(art_num)
189	n/a	self.assertTrue(resp.startswith("220 "), resp)
190	n/a	self.check_article_resp(resp, article, art_num)
191	n/a	# Tolerate running the tests from behind a NNTP virus checker
192	n/a	blacklist = lambda line: line.startswith(b'X-Antivirus')
193	n/a	filtered_head_lines = [line for line in head.lines
194	n/a	if not blacklist(line)]
195	n/a	filtered_lines = [line for line in article.lines
196	n/a	if not blacklist(line)]
197	n/a	self.assertEqual(filtered_lines, filtered_head_lines + [b''] + body.lines)
198	n/a
199	n/a	def test_capabilities(self):
200	n/a	# The server under test implements NNTP version 2 and has a
201	n/a	# couple of well-known capabilities. Just sanity check that we
202	n/a	# got them.
203	n/a	def _check_caps(caps):
204	n/a	caps_list = caps['LIST']
205	n/a	self.assertIsInstance(caps_list, (list, tuple))
206	n/a	self.assertIn('OVERVIEW.FMT', caps_list)
207	n/a	self.assertGreaterEqual(self.server.nntp_version, 2)
208	n/a	_check_caps(self.server.getcapabilities())
209	n/a	# This re-emits the command
210	n/a	resp, caps = self.server.capabilities()
211	n/a	_check_caps(caps)
212	n/a
213	n/a	def test_zlogin(self):
214	n/a	# This test must be the penultimate because further commands will be
215	n/a	# refused.
216	n/a	baduser = "notarealuser"
217	n/a	badpw = "notarealpassword"
218	n/a	# Check that bogus credentials cause failure
219	n/a	self.assertRaises(nntplib.NNTPError, self.server.login,
220	n/a	user=baduser, password=badpw, usenetrc=False)
221	n/a	# FIXME: We should check that correct credentials succeed, but that
222	n/a	# would require valid details for some server somewhere to be in the
223	n/a	# test suite, I think. Gmane is anonymous, at least as used for the
224	n/a	# other tests.
225	n/a
226	n/a	def test_zzquit(self):
227	n/a	# This test must be called last, hence the name
228	n/a	cls = type(self)
229	n/a	try:
230	n/a	self.server.quit()
231	n/a	finally:
232	n/a	cls.server = None
233	n/a
234	n/a	@classmethod
235	n/a	def wrap_methods(cls):
236	n/a	# Wrap all methods in a transient_internet() exception catcher
237	n/a	# XXX put a generic version in test.support?
238	n/a	def wrap_meth(meth):
239	n/a	@functools.wraps(meth)
240	n/a	def wrapped(self):
241	n/a	with support.transient_internet(self.NNTP_HOST):
242	n/a	meth(self)
243	n/a	return wrapped
244	n/a	for name in dir(cls):
245	n/a	if not name.startswith('test_'):
246	n/a	continue
247	n/a	meth = getattr(cls, name)
248	n/a	if not callable(meth):
249	n/a	continue
250	n/a	# Need to use a closure so that meth remains bound to its current
251	n/a	# value
252	n/a	setattr(cls, name, wrap_meth(meth))
253	n/a
254	n/a	def test_with_statement(self):
255	n/a	def is_connected():
256	n/a	if not hasattr(server, 'file'):
257	n/a	return False
258	n/a	try:
259	n/a	server.help()
260	n/a	except (OSError, EOFError):
261	n/a	return False
262	n/a	return True
263	n/a
264	n/a	with self.NNTP_CLASS(self.NNTP_HOST, timeout=TIMEOUT, usenetrc=False) as server:
265	n/a	self.assertTrue(is_connected())
266	n/a	self.assertTrue(server.help())
267	n/a	self.assertFalse(is_connected())
268	n/a
269	n/a	with self.NNTP_CLASS(self.NNTP_HOST, timeout=TIMEOUT, usenetrc=False) as server:
270	n/a	server.quit()
271	n/a	self.assertFalse(is_connected())
272	n/a
273	n/a
274	n/a	NetworkedNNTPTestsMixin.wrap_methods()
275	n/a
276	n/a
277	n/a	class NetworkedNNTPTests(NetworkedNNTPTestsMixin, unittest.TestCase):
278	n/a	# This server supports STARTTLS (gmane doesn't)
279	n/a	NNTP_HOST = 'news.trigofacile.com'
280	n/a	GROUP_NAME = 'fr.comp.lang.python'
281	n/a	GROUP_PAT = 'fr.comp.lang.*'
282	n/a
283	n/a	NNTP_CLASS = NNTP
284	n/a
285	n/a	@classmethod
286	n/a	def setUpClass(cls):
287	n/a	support.requires("network")
288	n/a	with support.transient_internet(cls.NNTP_HOST):
289	n/a	cls.server = cls.NNTP_CLASS(cls.NNTP_HOST, timeout=TIMEOUT, usenetrc=False)
290	n/a
291	n/a	@classmethod
292	n/a	def tearDownClass(cls):
293	n/a	if cls.server is not None:
294	n/a	cls.server.quit()
295	n/a
296	n/a	@unittest.skipUnless(ssl, 'requires SSL support')
297	n/a	class NetworkedNNTP_SSLTests(NetworkedNNTPTests):
298	n/a
299	n/a	# Technical limits for this public NNTP server (see http://www.aioe.org):
300	n/a	# "Only two concurrent connections per IP address are allowed and
301	n/a	# 400 connections per day are accepted from each IP address."
302	n/a
303	n/a	NNTP_HOST = 'nntp.aioe.org'
304	n/a	GROUP_NAME = 'comp.lang.python'
305	n/a	GROUP_PAT = 'comp.lang.*'
306	n/a
307	n/a	NNTP_CLASS = getattr(nntplib, 'NNTP_SSL', None)
308	n/a
309	n/a	# Disabled as it produces too much data
310	n/a	test_list = None
311	n/a
312	n/a	# Disabled as the connection will already be encrypted.
313	n/a	test_starttls = None
314	n/a
315	n/a
316	n/a	#
317	n/a	# Non-networked tests using a local server (or something mocking it).
318	n/a	#
319	n/a
320	n/a	class _NNTPServerIO(io.RawIOBase):
321	n/a	"""A raw IO object allowing NNTP commands to be received and processed
322	n/a	by a handler. The handler can push responses which can then be read
323	n/a	from the IO object."""
324	n/a
325	n/a	def __init__(self, handler):
326	n/a	io.RawIOBase.__init__(self)
327	n/a	# The channel from the client
328	n/a	self.c2s = io.BytesIO()
329	n/a	# The channel to the client
330	n/a	self.s2c = io.BytesIO()
331	n/a	self.handler = handler
332	n/a	self.handler.start(self.c2s.readline, self.push_data)
333	n/a
334	n/a	def readable(self):
335	n/a	return True
336	n/a
337	n/a	def writable(self):
338	n/a	return True
339	n/a
340	n/a	def push_data(self, data):
341	n/a	"""Push (buffer) some data to send to the client."""
342	n/a	pos = self.s2c.tell()
343	n/a	self.s2c.seek(0, 2)
344	n/a	self.s2c.write(data)
345	n/a	self.s2c.seek(pos)
346	n/a
347	n/a	def write(self, b):
348	n/a	"""The client sends us some data"""
349	n/a	pos = self.c2s.tell()
350	n/a	self.c2s.write(b)
351	n/a	self.c2s.seek(pos)
352	n/a	self.handler.process_pending()
353	n/a	return len(b)
354	n/a
355	n/a	def readinto(self, buf):
356	n/a	"""The client wants to read a response"""
357	n/a	self.handler.process_pending()
358	n/a	b = self.s2c.read(len(buf))
359	n/a	n = len(b)
360	n/a	buf[:n] = b
361	n/a	return n
362	n/a
363	n/a
364	n/a	def make_mock_file(handler):
365	n/a	sio = _NNTPServerIO(handler)
366	n/a	# Using BufferedRWPair instead of BufferedRandom ensures the file
367	n/a	# isn't seekable.
368	n/a	file = io.BufferedRWPair(sio, sio)
369	n/a	return (sio, file)
370	n/a
371	n/a
372	n/a	class MockedNNTPTestsMixin:
373	n/a	# Override in derived classes
374	n/a	handler_class = None
375	n/a
376	n/a	def setUp(self):
377	n/a	super().setUp()
378	n/a	self.make_server()
379	n/a
380	n/a	def tearDown(self):
381	n/a	super().tearDown()
382	n/a	del self.server
383	n/a
384	n/a	def make_server(self, args, *kwargs):
385	n/a	self.handler = self.handler_class()
386	n/a	self.sio, file = make_mock_file(self.handler)
387	n/a	self.server = nntplib._NNTPBase(file, 'test.server', args, *kwargs)
388	n/a	return self.server
389	n/a
390	n/a
391	n/a	class MockedNNTPWithReaderModeMixin(MockedNNTPTestsMixin):
392	n/a	def setUp(self):
393	n/a	super().setUp()
394	n/a	self.make_server(readermode=True)
395	n/a
396	n/a
397	n/a	class NNTPv1Handler:
398	n/a	"""A handler for RFC 977"""
399	n/a
400	n/a	welcome = "200 NNTP mock server"
401	n/a
402	n/a	def start(self, readline, push_data):
403	n/a	self.in_body = False
404	n/a	self.allow_posting = True
405	n/a	self._readline = readline
406	n/a	self._push_data = push_data
407	n/a	self._logged_in = False
408	n/a	self._user_sent = False
409	n/a	# Our welcome
410	n/a	self.handle_welcome()
411	n/a
412	n/a	def _decode(self, data):
413	n/a	return str(data, "utf-8", "surrogateescape")
414	n/a
415	n/a	def process_pending(self):
416	n/a	if self.in_body:
417	n/a	while True:
418	n/a	line = self._readline()
419	n/a	if not line:
420	n/a	return
421	n/a	self.body.append(line)
422	n/a	if line == b".\r\n":
423	n/a	break
424	n/a	try:
425	n/a	meth, tokens = self.body_callback
426	n/a	meth(*tokens, body=self.body)
427	n/a	finally:
428	n/a	self.body_callback = None
429	n/a	self.body = None
430	n/a	self.in_body = False
431	n/a	while True:
432	n/a	line = self._decode(self._readline())
433	n/a	if not line:
434	n/a	return
435	n/a	if not line.endswith("\r\n"):
436	n/a	raise ValueError("line doesn't end with \\r\\n: {!r}".format(line))
437	n/a	line = line[:-2]
438	n/a	cmd, *tokens = line.split()
439	n/a	#meth = getattr(self.handler, "handle_" + cmd.upper(), None)
440	n/a	meth = getattr(self, "handle_" + cmd.upper(), None)
441	n/a	if meth is None:
442	n/a	self.handle_unknown()
443	n/a	else:
444	n/a	try:
445	n/a	meth(*tokens)
446	n/a	except Exception as e:
447	n/a	raise ValueError("command failed: {!r}".format(line)) from e
448	n/a	else:
449	n/a	if self.in_body:
450	n/a	self.body_callback = meth, tokens
451	n/a	self.body = []
452	n/a
453	n/a	def expect_body(self):
454	n/a	"""Flag that the client is expected to post a request body"""
455	n/a	self.in_body = True
456	n/a
457	n/a	def push_data(self, data):
458	n/a	"""Push some binary data"""
459	n/a	self._push_data(data)
460	n/a
461	n/a	def push_lit(self, lit):
462	n/a	"""Push a string literal"""
463	n/a	lit = textwrap.dedent(lit)
464	n/a	lit = "\r\n".join(lit.splitlines()) + "\r\n"
465	n/a	lit = lit.encode('utf-8')
466	n/a	self.push_data(lit)
467	n/a
468	n/a	def handle_unknown(self):
469	n/a	self.push_lit("500 What?")
470	n/a
471	n/a	def handle_welcome(self):
472	n/a	self.push_lit(self.welcome)
473	n/a
474	n/a	def handle_QUIT(self):
475	n/a	self.push_lit("205 Bye!")
476	n/a
477	n/a	def handle_DATE(self):
478	n/a	self.push_lit("111 20100914001155")
479	n/a
480	n/a	def handle_GROUP(self, group):
481	n/a	if group == "fr.comp.lang.python":
482	n/a	self.push_lit("211 486 761 1265 fr.comp.lang.python")
483	n/a	else:
484	n/a	self.push_lit("411 No such group {}".format(group))
485	n/a
486	n/a	def handle_HELP(self):
487	n/a	self.push_lit("""\
488	n/a	100 Legal commands
489	n/a	authinfo user Name\|pass Password\|generic <prog> <args>
490	n/a	date
491	n/a	help
492	n/a	Report problems to <root@example.org>
493	n/a	.""")
494	n/a
495	n/a	def handle_STAT(self, message_spec=None):
496	n/a	if message_spec is None:
497	n/a	self.push_lit("412 No newsgroup selected")
498	n/a	elif message_spec == "3000234":
499	n/a	self.push_lit("223 3000234 <45223423@example.com>")
500	n/a	elif message_spec == "<45223423@example.com>":
501	n/a	self.push_lit("223 0 <45223423@example.com>")
502	n/a	else:
503	n/a	self.push_lit("430 No Such Article Found")
504	n/a
505	n/a	def handle_NEXT(self):
506	n/a	self.push_lit("223 3000237 <668929@example.org> retrieved")
507	n/a
508	n/a	def handle_LAST(self):
509	n/a	self.push_lit("223 3000234 <45223423@example.com> retrieved")
510	n/a
511	n/a	def handle_LIST(self, action=None, param=None):
512	n/a	if action is None:
513	n/a	self.push_lit("""\
514	n/a	215 Newsgroups in form "group high low flags".
515	n/a	comp.lang.python 0000052340 0000002828 y
516	n/a	comp.lang.python.announce 0000001153 0000000993 m
517	n/a	free.it.comp.lang.python 0000000002 0000000002 y
518	n/a	fr.comp.lang.python 0000001254 0000000760 y
519	n/a	free.it.comp.lang.python.learner 0000000000 0000000001 y
520	n/a	tw.bbs.comp.lang.python 0000000304 0000000304 y
521	n/a	.""")
522	n/a	elif action == "ACTIVE":
523	n/a	if param == "distutils":
524	n/a	self.push_lit("""\
525	n/a	215 Newsgroups in form "group high low flags"
526	n/a	gmane.comp.python.distutils.devel 0000014104 0000000001 m
527	n/a	gmane.comp.python.distutils.cvs 0000000000 0000000001 m
528	n/a	.""")
529	n/a	else:
530	n/a	self.push_lit("""\
531	n/a	215 Newsgroups in form "group high low flags"
532	n/a	.""")
533	n/a	elif action == "OVERVIEW.FMT":
534	n/a	self.push_lit("""\
535	n/a	215 Order of fields in overview database.
536	n/a	Subject:
537	n/a	From:
538	n/a	Date:
539	n/a	Message-ID:
540	n/a	References:
541	n/a	Bytes:
542	n/a	Lines:
543	n/a	Xref:full
544	n/a	.""")
545	n/a	elif action == "NEWSGROUPS":
546	n/a	assert param is not None
547	n/a	if param == "comp.lang.python":
548	n/a	self.push_lit("""\
549	n/a	215 Descriptions in form "group description".
550	n/a	comp.lang.python\tThe Python computer language.
551	n/a	.""")
552	n/a	elif param == "comp.lang.python*":
553	n/a	self.push_lit("""\
554	n/a	215 Descriptions in form "group description".
555	n/a	comp.lang.python.announce\tAnnouncements about the Python language. (Moderated)
556	n/a	comp.lang.python\tThe Python computer language.
557	n/a	.""")
558	n/a	else:
559	n/a	self.push_lit("""\
560	n/a	215 Descriptions in form "group description".
561	n/a	.""")
562	n/a	else:
563	n/a	self.push_lit('501 Unknown LIST keyword')
564	n/a
565	n/a	def handle_NEWNEWS(self, group, date_str, time_str):
566	n/a	# We hard code different return messages depending on passed
567	n/a	# argument and date syntax.
568	n/a	if (group == "comp.lang.python" and date_str == "20100913"
569	n/a	and time_str == "082004"):
570	n/a	# Date was passed in RFC 3977 format (NNTP "v2")
571	n/a	self.push_lit("""\
572	n/a	230 list of newsarticles (NNTP v2) created after Mon Sep 13 08:20:04 2010 follows
573	n/a	<a4929a40-6328-491a-aaaf-cb79ed7309a2@q2g2000vbk.googlegroups.com>
574	n/a	<f30c0419-f549-4218-848f-d7d0131da931@y3g2000vbm.googlegroups.com>
575	n/a	.""")
576	n/a	elif (group == "comp.lang.python" and date_str == "100913"
577	n/a	and time_str == "082004"):
578	n/a	# Date was passed in RFC 977 format (NNTP "v1")
579	n/a	self.push_lit("""\
580	n/a	230 list of newsarticles (NNTP v1) created after Mon Sep 13 08:20:04 2010 follows
581	n/a	<a4929a40-6328-491a-aaaf-cb79ed7309a2@q2g2000vbk.googlegroups.com>
582	n/a	<f30c0419-f549-4218-848f-d7d0131da931@y3g2000vbm.googlegroups.com>
583	n/a	.""")
584	n/a	elif (group == 'comp.lang.python' and
585	n/a	date_str in ('20100101', '100101') and
586	n/a	time_str == '090000'):
587	n/a	self.push_lit('too long line' * 3000 +
588	n/a	'\n.')
589	n/a	else:
590	n/a	self.push_lit("""\
591	n/a	230 An empty list of newsarticles follows
592	n/a	.""")
593	n/a	# (Note for experiments: many servers disable NEWNEWS.
594	n/a	# As of this writing, sicinfo3.epfl.ch doesn't.)
595	n/a
596	n/a	def handle_XOVER(self, message_spec):
597	n/a	if message_spec == "57-59":
598	n/a	self.push_lit(
599	n/a	"224 Overview information for 57-58 follows\n"
600	n/a	"57\tRe: ANN: New Plone book with strong Python (and Zope) themes throughout"
601	n/a	"\tDoug Hellmann <doug.hellmann-Re5JQEeQqe8AvxtiuMwx3w@public.gmane.org>"
602	n/a	"\tSat, 19 Jun 2010 18:04:08 -0400"
603	n/a	"\t<4FD05F05-F98B-44DC-8111-C6009C925F0C@gmail.com>"
604	n/a	"\t<hvalf7$ort$1@dough.gmane.org>\t7103\t16"
605	n/a	"\tXref: news.gmane.org gmane.comp.python.authors:57"
606	n/a	"\n"
607	n/a	"58\tLooking for a few good bloggers"
608	n/a	"\tDoug Hellmann <doug.hellmann-Re5JQEeQqe8AvxtiuMwx3w@public.gmane.org>"
609	n/a	"\tThu, 22 Jul 2010 09:14:14 -0400"
610	n/a	"\t<A29863FA-F388-40C3-AA25-0FD06B09B5BF@gmail.com>"
611	n/a	"\t\t6683\t16"
612	n/a	"\t"
613	n/a	"\n"
614	n/a	# A UTF-8 overview line from fr.comp.lang.python
615	n/a	"59\tRe: Message d'erreur incomprÃ©hensible (par moi)"
616	n/a	"\tEric Brunel <eric.brunel@pragmadev.nospam.com>"
617	n/a	"\tWed, 15 Sep 2010 18:09:15 +0200"
618	n/a	"\t<eric.brunel-2B8B56.18091515092010@news.wanadoo.fr>"
619	n/a	"\t<4c90ec87$0$32425$ba4acef3@reader.news.orange.fr>\t1641\t27"
620	n/a	"\tXref: saria.nerim.net fr.comp.lang.python:1265"
621	n/a	"\n"
622	n/a	".\n")
623	n/a	else:
624	n/a	self.push_lit("""\
625	n/a	224 No articles
626	n/a	.""")
627	n/a
628	n/a	def handle_POST(self, *, body=None):
629	n/a	if body is None:
630	n/a	if self.allow_posting:
631	n/a	self.push_lit("340 Input article; end with <CR-LF>.<CR-LF>")
632	n/a	self.expect_body()
633	n/a	else:
634	n/a	self.push_lit("440 Posting not permitted")
635	n/a	else:
636	n/a	assert self.allow_posting
637	n/a	self.push_lit("240 Article received OK")
638	n/a	self.posted_body = body
639	n/a
640	n/a	def handle_IHAVE(self, message_id, *, body=None):
641	n/a	if body is None:
642	n/a	if (self.allow_posting and
643	n/a	message_id == "<i.am.an.article.you.will.want@example.com>"):
644	n/a	self.push_lit("335 Send it; end with <CR-LF>.<CR-LF>")
645	n/a	self.expect_body()
646	n/a	else:
647	n/a	self.push_lit("435 Article not wanted")
648	n/a	else:
649	n/a	assert self.allow_posting
650	n/a	self.push_lit("235 Article transferred OK")
651	n/a	self.posted_body = body
652	n/a
653	n/a	sample_head = """\
654	n/a	From: "Demo User" <nobody@example.net>
655	n/a	Subject: I am just a test article
656	n/a	Content-Type: text/plain; charset=UTF-8; format=flowed
657	n/a	Message-ID: <i.am.an.article.you.will.want@example.com>"""
658	n/a
659	n/a	sample_body = """\
660	n/a	This is just a test article.
661	n/a	..Here is a dot-starting line.
662	n/a
663	n/a	-- Signed by Andr\xe9."""
664	n/a
665	n/a	sample_article = sample_head + "\n\n" + sample_body
666	n/a
667	n/a	def handle_ARTICLE(self, message_spec=None):
668	n/a	if message_spec is None:
669	n/a	self.push_lit("220 3000237 <45223423@example.com>")
670	n/a	elif message_spec == "<45223423@example.com>":
671	n/a	self.push_lit("220 0 <45223423@example.com>")
672	n/a	elif message_spec == "3000234":
673	n/a	self.push_lit("220 3000234 <45223423@example.com>")
674	n/a	else:
675	n/a	self.push_lit("430 No Such Article Found")
676	n/a	return
677	n/a	self.push_lit(self.sample_article)
678	n/a	self.push_lit(".")
679	n/a
680	n/a	def handle_HEAD(self, message_spec=None):
681	n/a	if message_spec is None:
682	n/a	self.push_lit("221 3000237 <45223423@example.com>")
683	n/a	elif message_spec == "<45223423@example.com>":
684	n/a	self.push_lit("221 0 <45223423@example.com>")
685	n/a	elif message_spec == "3000234":
686	n/a	self.push_lit("221 3000234 <45223423@example.com>")
687	n/a	else:
688	n/a	self.push_lit("430 No Such Article Found")
689	n/a	return
690	n/a	self.push_lit(self.sample_head)
691	n/a	self.push_lit(".")
692	n/a
693	n/a	def handle_BODY(self, message_spec=None):
694	n/a	if message_spec is None:
695	n/a	self.push_lit("222 3000237 <45223423@example.com>")
696	n/a	elif message_spec == "<45223423@example.com>":
697	n/a	self.push_lit("222 0 <45223423@example.com>")
698	n/a	elif message_spec == "3000234":
699	n/a	self.push_lit("222 3000234 <45223423@example.com>")
700	n/a	else:
701	n/a	self.push_lit("430 No Such Article Found")
702	n/a	return
703	n/a	self.push_lit(self.sample_body)
704	n/a	self.push_lit(".")
705	n/a
706	n/a	def handle_AUTHINFO(self, cred_type, data):
707	n/a	if self._logged_in:
708	n/a	self.push_lit('502 Already Logged In')
709	n/a	elif cred_type == 'user':
710	n/a	if self._user_sent:
711	n/a	self.push_lit('482 User Credential Already Sent')
712	n/a	else:
713	n/a	self.push_lit('381 Password Required')
714	n/a	self._user_sent = True
715	n/a	elif cred_type == 'pass':
716	n/a	self.push_lit('281 Login Successful')
717	n/a	self._logged_in = True
718	n/a	else:
719	n/a	raise Exception('Unknown cred type {}'.format(cred_type))
720	n/a
721	n/a
722	n/a	class NNTPv2Handler(NNTPv1Handler):
723	n/a	"""A handler for RFC 3977 (NNTP "v2")"""
724	n/a
725	n/a	def handle_CAPABILITIES(self):
726	n/a	fmt = """\
727	n/a	101 Capability list:
728	n/a	VERSION 2 3
729	n/a	IMPLEMENTATION INN 2.5.1{}
730	n/a	HDR
731	n/a	LIST ACTIVE ACTIVE.TIMES DISTRIB.PATS HEADERS NEWSGROUPS OVERVIEW.FMT
732	n/a	OVER
733	n/a	POST
734	n/a	READER
735	n/a	."""
736	n/a
737	n/a	if not self._logged_in:
738	n/a	self.push_lit(fmt.format('\n AUTHINFO USER'))
739	n/a	else:
740	n/a	self.push_lit(fmt.format(''))
741	n/a
742	n/a	def handle_MODE(self, _):
743	n/a	raise Exception('MODE READER sent despite READER has been advertised')
744	n/a
745	n/a	def handle_OVER(self, message_spec=None):
746	n/a	return self.handle_XOVER(message_spec)
747	n/a
748	n/a
749	n/a	class CapsAfterLoginNNTPv2Handler(NNTPv2Handler):
750	n/a	"""A handler that allows CAPABILITIES only after login"""
751	n/a
752	n/a	def handle_CAPABILITIES(self):
753	n/a	if not self._logged_in:
754	n/a	self.push_lit('480 You must log in.')
755	n/a	else:
756	n/a	super().handle_CAPABILITIES()
757	n/a
758	n/a
759	n/a	class ModeSwitchingNNTPv2Handler(NNTPv2Handler):
760	n/a	"""A server that starts in transit mode"""
761	n/a
762	n/a	def __init__(self):
763	n/a	self._switched = False
764	n/a
765	n/a	def handle_CAPABILITIES(self):
766	n/a	fmt = """\
767	n/a	101 Capability list:
768	n/a	VERSION 2 3
769	n/a	IMPLEMENTATION INN 2.5.1
770	n/a	HDR
771	n/a	LIST ACTIVE ACTIVE.TIMES DISTRIB.PATS HEADERS NEWSGROUPS OVERVIEW.FMT
772	n/a	OVER
773	n/a	POST
774	n/a	{}READER
775	n/a	."""
776	n/a	if self._switched:
777	n/a	self.push_lit(fmt.format(''))
778	n/a	else:
779	n/a	self.push_lit(fmt.format('MODE-'))
780	n/a
781	n/a	def handle_MODE(self, what):
782	n/a	assert not self._switched and what == 'reader'
783	n/a	self._switched = True
784	n/a	self.push_lit('200 Posting allowed')
785	n/a
786	n/a
787	n/a	class NNTPv1v2TestsMixin:
788	n/a
789	n/a	def setUp(self):
790	n/a	super().setUp()
791	n/a
792	n/a	def test_welcome(self):
793	n/a	self.assertEqual(self.server.welcome, self.handler.welcome)
794	n/a
795	n/a	def test_authinfo(self):
796	n/a	if self.nntp_version == 2:
797	n/a	self.assertIn('AUTHINFO', self.server._caps)
798	n/a	self.server.login('testuser', 'testpw')
799	n/a	# if AUTHINFO is gone from _caps we also know that getcapabilities()
800	n/a	# has been called after login as it should
801	n/a	self.assertNotIn('AUTHINFO', self.server._caps)
802	n/a
803	n/a	def test_date(self):
804	n/a	resp, date = self.server.date()
805	n/a	self.assertEqual(resp, "111 20100914001155")
806	n/a	self.assertEqual(date, datetime.datetime(2010, 9, 14, 0, 11, 55))
807	n/a
808	n/a	def test_quit(self):
809	n/a	self.assertFalse(self.sio.closed)
810	n/a	resp = self.server.quit()
811	n/a	self.assertEqual(resp, "205 Bye!")
812	n/a	self.assertTrue(self.sio.closed)
813	n/a
814	n/a	def test_help(self):
815	n/a	resp, help = self.server.help()
816	n/a	self.assertEqual(resp, "100 Legal commands")
817	n/a	self.assertEqual(help, [
818	n/a	' authinfo user Name\|pass Password\|generic <prog> <args>',
819	n/a	' date',
820	n/a	' help',
821	n/a	'Report problems to <root@example.org>',
822	n/a	])
823	n/a
824	n/a	def test_list(self):
825	n/a	resp, groups = self.server.list()
826	n/a	self.assertEqual(len(groups), 6)
827	n/a	g = groups[1]
828	n/a	self.assertEqual(g,
829	n/a	GroupInfo("comp.lang.python.announce", "0000001153",
830	n/a	"0000000993", "m"))
831	n/a	resp, groups = self.server.list("distutils")
832	n/a	self.assertEqual(len(groups), 2)
833	n/a	g = groups[0]
834	n/a	self.assertEqual(g,
835	n/a	GroupInfo("gmane.comp.python.distutils.devel", "0000014104",
836	n/a	"0000000001", "m"))
837	n/a
838	n/a	def test_stat(self):
839	n/a	resp, art_num, message_id = self.server.stat(3000234)
840	n/a	self.assertEqual(resp, "223 3000234 <45223423@example.com>")
841	n/a	self.assertEqual(art_num, 3000234)
842	n/a	self.assertEqual(message_id, "<45223423@example.com>")
843	n/a	resp, art_num, message_id = self.server.stat("<45223423@example.com>")
844	n/a	self.assertEqual(resp, "223 0 <45223423@example.com>")
845	n/a	self.assertEqual(art_num, 0)
846	n/a	self.assertEqual(message_id, "<45223423@example.com>")
847	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
848	n/a	self.server.stat("<non.existent.id>")
849	n/a	self.assertEqual(cm.exception.response, "430 No Such Article Found")
850	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
851	n/a	self.server.stat()
852	n/a	self.assertEqual(cm.exception.response, "412 No newsgroup selected")
853	n/a
854	n/a	def test_next(self):
855	n/a	resp, art_num, message_id = self.server.next()
856	n/a	self.assertEqual(resp, "223 3000237 <668929@example.org> retrieved")
857	n/a	self.assertEqual(art_num, 3000237)
858	n/a	self.assertEqual(message_id, "<668929@example.org>")
859	n/a
860	n/a	def test_last(self):
861	n/a	resp, art_num, message_id = self.server.last()
862	n/a	self.assertEqual(resp, "223 3000234 <45223423@example.com> retrieved")
863	n/a	self.assertEqual(art_num, 3000234)
864	n/a	self.assertEqual(message_id, "<45223423@example.com>")
865	n/a
866	n/a	def test_description(self):
867	n/a	desc = self.server.description("comp.lang.python")
868	n/a	self.assertEqual(desc, "The Python computer language.")
869	n/a	desc = self.server.description("comp.lang.pythonx")
870	n/a	self.assertEqual(desc, "")
871	n/a
872	n/a	def test_descriptions(self):
873	n/a	resp, groups = self.server.descriptions("comp.lang.python")
874	n/a	self.assertEqual(resp, '215 Descriptions in form "group description".')
875	n/a	self.assertEqual(groups, {
876	n/a	"comp.lang.python": "The Python computer language.",
877	n/a	})
878	n/a	resp, groups = self.server.descriptions("comp.lang.python*")
879	n/a	self.assertEqual(groups, {
880	n/a	"comp.lang.python": "The Python computer language.",
881	n/a	"comp.lang.python.announce": "Announcements about the Python language. (Moderated)",
882	n/a	})
883	n/a	resp, groups = self.server.descriptions("comp.lang.pythonx")
884	n/a	self.assertEqual(groups, {})
885	n/a
886	n/a	def test_group(self):
887	n/a	resp, count, first, last, group = self.server.group("fr.comp.lang.python")
888	n/a	self.assertTrue(resp.startswith("211 "), resp)
889	n/a	self.assertEqual(first, 761)
890	n/a	self.assertEqual(last, 1265)
891	n/a	self.assertEqual(count, 486)
892	n/a	self.assertEqual(group, "fr.comp.lang.python")
893	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
894	n/a	self.server.group("comp.lang.python.devel")
895	n/a	exc = cm.exception
896	n/a	self.assertTrue(exc.response.startswith("411 No such group"),
897	n/a	exc.response)
898	n/a
899	n/a	def test_newnews(self):
900	n/a	# NEWNEWS comp.lang.python [20]100913 082004
901	n/a	dt = datetime.datetime(2010, 9, 13, 8, 20, 4)
902	n/a	resp, ids = self.server.newnews("comp.lang.python", dt)
903	n/a	expected = (
904	n/a	"230 list of newsarticles (NNTP v{0}) "
905	n/a	"created after Mon Sep 13 08:20:04 2010 follows"
906	n/a	).format(self.nntp_version)
907	n/a	self.assertEqual(resp, expected)
908	n/a	self.assertEqual(ids, [
909	n/a	"<a4929a40-6328-491a-aaaf-cb79ed7309a2@q2g2000vbk.googlegroups.com>",
910	n/a	"<f30c0419-f549-4218-848f-d7d0131da931@y3g2000vbm.googlegroups.com>",
911	n/a	])
912	n/a	# NEWNEWS fr.comp.lang.python [20]100913 082004
913	n/a	dt = datetime.datetime(2010, 9, 13, 8, 20, 4)
914	n/a	resp, ids = self.server.newnews("fr.comp.lang.python", dt)
915	n/a	self.assertEqual(resp, "230 An empty list of newsarticles follows")
916	n/a	self.assertEqual(ids, [])
917	n/a
918	n/a	def _check_article_body(self, lines):
919	n/a	self.assertEqual(len(lines), 4)
920	n/a	self.assertEqual(lines[-1].decode('utf-8'), "-- Signed by AndrÃ©.")
921	n/a	self.assertEqual(lines[-2], b"")
922	n/a	self.assertEqual(lines[-3], b".Here is a dot-starting line.")
923	n/a	self.assertEqual(lines[-4], b"This is just a test article.")
924	n/a
925	n/a	def _check_article_head(self, lines):
926	n/a	self.assertEqual(len(lines), 4)
927	n/a	self.assertEqual(lines[0], b'From: "Demo User" <nobody@example.net>')
928	n/a	self.assertEqual(lines[3], b"Message-ID: <i.am.an.article.you.will.want@example.com>")
929	n/a
930	n/a	def _check_article_data(self, lines):
931	n/a	self.assertEqual(len(lines), 9)
932	n/a	self._check_article_head(lines[:4])
933	n/a	self._check_article_body(lines[-4:])
934	n/a	self.assertEqual(lines[4], b"")
935	n/a
936	n/a	def test_article(self):
937	n/a	# ARTICLE
938	n/a	resp, info = self.server.article()
939	n/a	self.assertEqual(resp, "220 3000237 <45223423@example.com>")
940	n/a	art_num, message_id, lines = info
941	n/a	self.assertEqual(art_num, 3000237)
942	n/a	self.assertEqual(message_id, "<45223423@example.com>")
943	n/a	self._check_article_data(lines)
944	n/a	# ARTICLE num
945	n/a	resp, info = self.server.article(3000234)
946	n/a	self.assertEqual(resp, "220 3000234 <45223423@example.com>")
947	n/a	art_num, message_id, lines = info
948	n/a	self.assertEqual(art_num, 3000234)
949	n/a	self.assertEqual(message_id, "<45223423@example.com>")
950	n/a	self._check_article_data(lines)
951	n/a	# ARTICLE id
952	n/a	resp, info = self.server.article("<45223423@example.com>")
953	n/a	self.assertEqual(resp, "220 0 <45223423@example.com>")
954	n/a	art_num, message_id, lines = info
955	n/a	self.assertEqual(art_num, 0)
956	n/a	self.assertEqual(message_id, "<45223423@example.com>")
957	n/a	self._check_article_data(lines)
958	n/a	# Non-existent id
959	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
960	n/a	self.server.article("<non-existent@example.com>")
961	n/a	self.assertEqual(cm.exception.response, "430 No Such Article Found")
962	n/a
963	n/a	def test_article_file(self):
964	n/a	# With a "file" argument
965	n/a	f = io.BytesIO()
966	n/a	resp, info = self.server.article(file=f)
967	n/a	self.assertEqual(resp, "220 3000237 <45223423@example.com>")
968	n/a	art_num, message_id, lines = info
969	n/a	self.assertEqual(art_num, 3000237)
970	n/a	self.assertEqual(message_id, "<45223423@example.com>")
971	n/a	self.assertEqual(lines, [])
972	n/a	data = f.getvalue()
973	n/a	self.assertTrue(data.startswith(
974	n/a	b'From: "Demo User" <nobody@example.net>\r\n'
975	n/a	b'Subject: I am just a test article\r\n'
976	n/a	), ascii(data))
977	n/a	self.assertTrue(data.endswith(
978	n/a	b'This is just a test article.\r\n'
979	n/a	b'.Here is a dot-starting line.\r\n'
980	n/a	b'\r\n'
981	n/a	b'-- Signed by Andr\xc3\xa9.\r\n'
982	n/a	), ascii(data))
983	n/a
984	n/a	def test_head(self):
985	n/a	# HEAD
986	n/a	resp, info = self.server.head()
987	n/a	self.assertEqual(resp, "221 3000237 <45223423@example.com>")
988	n/a	art_num, message_id, lines = info
989	n/a	self.assertEqual(art_num, 3000237)
990	n/a	self.assertEqual(message_id, "<45223423@example.com>")
991	n/a	self._check_article_head(lines)
992	n/a	# HEAD num
993	n/a	resp, info = self.server.head(3000234)
994	n/a	self.assertEqual(resp, "221 3000234 <45223423@example.com>")
995	n/a	art_num, message_id, lines = info
996	n/a	self.assertEqual(art_num, 3000234)
997	n/a	self.assertEqual(message_id, "<45223423@example.com>")
998	n/a	self._check_article_head(lines)
999	n/a	# HEAD id
1000	n/a	resp, info = self.server.head("<45223423@example.com>")
1001	n/a	self.assertEqual(resp, "221 0 <45223423@example.com>")
1002	n/a	art_num, message_id, lines = info
1003	n/a	self.assertEqual(art_num, 0)
1004	n/a	self.assertEqual(message_id, "<45223423@example.com>")
1005	n/a	self._check_article_head(lines)
1006	n/a	# Non-existent id
1007	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1008	n/a	self.server.head("<non-existent@example.com>")
1009	n/a	self.assertEqual(cm.exception.response, "430 No Such Article Found")
1010	n/a
1011	n/a	def test_head_file(self):
1012	n/a	f = io.BytesIO()
1013	n/a	resp, info = self.server.head(file=f)
1014	n/a	self.assertEqual(resp, "221 3000237 <45223423@example.com>")
1015	n/a	art_num, message_id, lines = info
1016	n/a	self.assertEqual(art_num, 3000237)
1017	n/a	self.assertEqual(message_id, "<45223423@example.com>")
1018	n/a	self.assertEqual(lines, [])
1019	n/a	data = f.getvalue()
1020	n/a	self.assertTrue(data.startswith(
1021	n/a	b'From: "Demo User" <nobody@example.net>\r\n'
1022	n/a	b'Subject: I am just a test article\r\n'
1023	n/a	), ascii(data))
1024	n/a	self.assertFalse(data.endswith(
1025	n/a	b'This is just a test article.\r\n'
1026	n/a	b'.Here is a dot-starting line.\r\n'
1027	n/a	b'\r\n'
1028	n/a	b'-- Signed by Andr\xc3\xa9.\r\n'
1029	n/a	), ascii(data))
1030	n/a
1031	n/a	def test_body(self):
1032	n/a	# BODY
1033	n/a	resp, info = self.server.body()
1034	n/a	self.assertEqual(resp, "222 3000237 <45223423@example.com>")
1035	n/a	art_num, message_id, lines = info
1036	n/a	self.assertEqual(art_num, 3000237)
1037	n/a	self.assertEqual(message_id, "<45223423@example.com>")
1038	n/a	self._check_article_body(lines)
1039	n/a	# BODY num
1040	n/a	resp, info = self.server.body(3000234)
1041	n/a	self.assertEqual(resp, "222 3000234 <45223423@example.com>")
1042	n/a	art_num, message_id, lines = info
1043	n/a	self.assertEqual(art_num, 3000234)
1044	n/a	self.assertEqual(message_id, "<45223423@example.com>")
1045	n/a	self._check_article_body(lines)
1046	n/a	# BODY id
1047	n/a	resp, info = self.server.body("<45223423@example.com>")
1048	n/a	self.assertEqual(resp, "222 0 <45223423@example.com>")
1049	n/a	art_num, message_id, lines = info
1050	n/a	self.assertEqual(art_num, 0)
1051	n/a	self.assertEqual(message_id, "<45223423@example.com>")
1052	n/a	self._check_article_body(lines)
1053	n/a	# Non-existent id
1054	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1055	n/a	self.server.body("<non-existent@example.com>")
1056	n/a	self.assertEqual(cm.exception.response, "430 No Such Article Found")
1057	n/a
1058	n/a	def test_body_file(self):
1059	n/a	f = io.BytesIO()
1060	n/a	resp, info = self.server.body(file=f)
1061	n/a	self.assertEqual(resp, "222 3000237 <45223423@example.com>")
1062	n/a	art_num, message_id, lines = info
1063	n/a	self.assertEqual(art_num, 3000237)
1064	n/a	self.assertEqual(message_id, "<45223423@example.com>")
1065	n/a	self.assertEqual(lines, [])
1066	n/a	data = f.getvalue()
1067	n/a	self.assertFalse(data.startswith(
1068	n/a	b'From: "Demo User" <nobody@example.net>\r\n'
1069	n/a	b'Subject: I am just a test article\r\n'
1070	n/a	), ascii(data))
1071	n/a	self.assertTrue(data.endswith(
1072	n/a	b'This is just a test article.\r\n'
1073	n/a	b'.Here is a dot-starting line.\r\n'
1074	n/a	b'\r\n'
1075	n/a	b'-- Signed by Andr\xc3\xa9.\r\n'
1076	n/a	), ascii(data))
1077	n/a
1078	n/a	def check_over_xover_resp(self, resp, overviews):
1079	n/a	self.assertTrue(resp.startswith("224 "), resp)
1080	n/a	self.assertEqual(len(overviews), 3)
1081	n/a	art_num, over = overviews[0]
1082	n/a	self.assertEqual(art_num, 57)
1083	n/a	self.assertEqual(over, {
1084	n/a	"from": "Doug Hellmann <doug.hellmann-Re5JQEeQqe8AvxtiuMwx3w@public.gmane.org>",
1085	n/a	"subject": "Re: ANN: New Plone book with strong Python (and Zope) themes throughout",
1086	n/a	"date": "Sat, 19 Jun 2010 18:04:08 -0400",
1087	n/a	"message-id": "<4FD05F05-F98B-44DC-8111-C6009C925F0C@gmail.com>",
1088	n/a	"references": "<hvalf7$ort$1@dough.gmane.org>",
1089	n/a	":bytes": "7103",
1090	n/a	":lines": "16",
1091	n/a	"xref": "news.gmane.org gmane.comp.python.authors:57"
1092	n/a	})
1093	n/a	art_num, over = overviews[1]
1094	n/a	self.assertEqual(over["xref"], None)
1095	n/a	art_num, over = overviews[2]
1096	n/a	self.assertEqual(over["subject"],
1097	n/a	"Re: Message d'erreur incomprÃ©hensible (par moi)")
1098	n/a
1099	n/a	def test_xover(self):
1100	n/a	resp, overviews = self.server.xover(57, 59)
1101	n/a	self.check_over_xover_resp(resp, overviews)
1102	n/a
1103	n/a	def test_over(self):
1104	n/a	# In NNTP "v1", this will fallback on XOVER
1105	n/a	resp, overviews = self.server.over((57, 59))
1106	n/a	self.check_over_xover_resp(resp, overviews)
1107	n/a
1108	n/a	sample_post = (
1109	n/a	b'From: "Demo User" <nobody@example.net>\r\n'
1110	n/a	b'Subject: I am just a test article\r\n'
1111	n/a	b'Content-Type: text/plain; charset=UTF-8; format=flowed\r\n'
1112	n/a	b'Message-ID: <i.am.an.article.you.will.want@example.com>\r\n'
1113	n/a	b'\r\n'
1114	n/a	b'This is just a test article.\r\n'
1115	n/a	b'.Here is a dot-starting line.\r\n'
1116	n/a	b'\r\n'
1117	n/a	b'-- Signed by Andr\xc3\xa9.\r\n'
1118	n/a	)
1119	n/a
1120	n/a	def _check_posted_body(self):
1121	n/a	# Check the raw body as received by the server
1122	n/a	lines = self.handler.posted_body
1123	n/a	# One additional line for the "." terminator
1124	n/a	self.assertEqual(len(lines), 10)
1125	n/a	self.assertEqual(lines[-1], b'.\r\n')
1126	n/a	self.assertEqual(lines[-2], b'-- Signed by Andr\xc3\xa9.\r\n')
1127	n/a	self.assertEqual(lines[-3], b'\r\n')
1128	n/a	self.assertEqual(lines[-4], b'..Here is a dot-starting line.\r\n')
1129	n/a	self.assertEqual(lines[0], b'From: "Demo User" <nobody@example.net>\r\n')
1130	n/a
1131	n/a	def _check_post_ihave_sub(self, func, *args, file_factory):
1132	n/a	# First the prepared post with CRLF endings
1133	n/a	post = self.sample_post
1134	n/a	func_args = args + (file_factory(post),)
1135	n/a	self.handler.posted_body = None
1136	n/a	resp = func(*func_args)
1137	n/a	self._check_posted_body()
1138	n/a	# Then the same post with "normal" line endings - they should be
1139	n/a	# converted by NNTP.post and NNTP.ihave.
1140	n/a	post = self.sample_post.replace(b"\r\n", b"\n")
1141	n/a	func_args = args + (file_factory(post),)
1142	n/a	self.handler.posted_body = None
1143	n/a	resp = func(*func_args)
1144	n/a	self._check_posted_body()
1145	n/a	return resp
1146	n/a
1147	n/a	def check_post_ihave(self, func, success_resp, *args):
1148	n/a	# With a bytes object
1149	n/a	resp = self._check_post_ihave_sub(func, *args, file_factory=bytes)
1150	n/a	self.assertEqual(resp, success_resp)
1151	n/a	# With a bytearray object
1152	n/a	resp = self._check_post_ihave_sub(func, *args, file_factory=bytearray)
1153	n/a	self.assertEqual(resp, success_resp)
1154	n/a	# With a file object
1155	n/a	resp = self._check_post_ihave_sub(func, *args, file_factory=io.BytesIO)
1156	n/a	self.assertEqual(resp, success_resp)
1157	n/a	# With an iterable of terminated lines
1158	n/a	def iterlines(b):
1159	n/a	return iter(b.splitlines(keepends=True))
1160	n/a	resp = self._check_post_ihave_sub(func, *args, file_factory=iterlines)
1161	n/a	self.assertEqual(resp, success_resp)
1162	n/a	# With an iterable of non-terminated lines
1163	n/a	def iterlines(b):
1164	n/a	return iter(b.splitlines(keepends=False))
1165	n/a	resp = self._check_post_ihave_sub(func, *args, file_factory=iterlines)
1166	n/a	self.assertEqual(resp, success_resp)
1167	n/a
1168	n/a	def test_post(self):
1169	n/a	self.check_post_ihave(self.server.post, "240 Article received OK")
1170	n/a	self.handler.allow_posting = False
1171	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1172	n/a	self.server.post(self.sample_post)
1173	n/a	self.assertEqual(cm.exception.response,
1174	n/a	"440 Posting not permitted")
1175	n/a
1176	n/a	def test_ihave(self):
1177	n/a	self.check_post_ihave(self.server.ihave, "235 Article transferred OK",
1178	n/a	"<i.am.an.article.you.will.want@example.com>")
1179	n/a	with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1180	n/a	self.server.ihave("<another.message.id>", self.sample_post)
1181	n/a	self.assertEqual(cm.exception.response,
1182	n/a	"435 Article not wanted")
1183	n/a
1184	n/a	def test_too_long_lines(self):
1185	n/a	dt = datetime.datetime(2010, 1, 1, 9, 0, 0)
1186	n/a	self.assertRaises(nntplib.NNTPDataError,
1187	n/a	self.server.newnews, "comp.lang.python", dt)
1188	n/a
1189	n/a
1190	n/a	class NNTPv1Tests(NNTPv1v2TestsMixin, MockedNNTPTestsMixin, unittest.TestCase):
1191	n/a	"""Tests an NNTP v1 server (no capabilities)."""
1192	n/a
1193	n/a	nntp_version = 1
1194	n/a	handler_class = NNTPv1Handler
1195	n/a
1196	n/a	def test_caps(self):
1197	n/a	caps = self.server.getcapabilities()
1198	n/a	self.assertEqual(caps, {})
1199	n/a	self.assertEqual(self.server.nntp_version, 1)
1200	n/a	self.assertEqual(self.server.nntp_implementation, None)
1201	n/a
1202	n/a
1203	n/a	class NNTPv2Tests(NNTPv1v2TestsMixin, MockedNNTPTestsMixin, unittest.TestCase):
1204	n/a	"""Tests an NNTP v2 server (with capabilities)."""
1205	n/a
1206	n/a	nntp_version = 2
1207	n/a	handler_class = NNTPv2Handler
1208	n/a
1209	n/a	def test_caps(self):
1210	n/a	caps = self.server.getcapabilities()
1211	n/a	self.assertEqual(caps, {
1212	n/a	'VERSION': ['2', '3'],
1213	n/a	'IMPLEMENTATION': ['INN', '2.5.1'],
1214	n/a	'AUTHINFO': ['USER'],
1215	n/a	'HDR': [],
1216	n/a	'LIST': ['ACTIVE', 'ACTIVE.TIMES', 'DISTRIB.PATS',
1217	n/a	'HEADERS', 'NEWSGROUPS', 'OVERVIEW.FMT'],
1218	n/a	'OVER': [],
1219	n/a	'POST': [],
1220	n/a	'READER': [],
1221	n/a	})
1222	n/a	self.assertEqual(self.server.nntp_version, 3)
1223	n/a	self.assertEqual(self.server.nntp_implementation, 'INN 2.5.1')
1224	n/a
1225	n/a
1226	n/a	class CapsAfterLoginNNTPv2Tests(MockedNNTPTestsMixin, unittest.TestCase):
1227	n/a	"""Tests a probably NNTP v2 server with capabilities only after login."""
1228	n/a
1229	n/a	nntp_version = 2
1230	n/a	handler_class = CapsAfterLoginNNTPv2Handler
1231	n/a
1232	n/a	def test_caps_only_after_login(self):
1233	n/a	self.assertEqual(self.server._caps, {})
1234	n/a	self.server.login('testuser', 'testpw')
1235	n/a	self.assertIn('VERSION', self.server._caps)
1236	n/a
1237	n/a
1238	n/a	class SendReaderNNTPv2Tests(MockedNNTPWithReaderModeMixin,
1239	n/a	unittest.TestCase):
1240	n/a	"""Same tests as for v2 but we tell NTTP to send MODE READER to a server
1241	n/a	that isn't in READER mode by default."""
1242	n/a
1243	n/a	nntp_version = 2
1244	n/a	handler_class = ModeSwitchingNNTPv2Handler
1245	n/a
1246	n/a	def test_we_are_in_reader_mode_after_connect(self):
1247	n/a	self.assertIn('READER', self.server._caps)
1248	n/a
1249	n/a
1250	n/a	class MiscTests(unittest.TestCase):
1251	n/a
1252	n/a	def test_decode_header(self):
1253	n/a	def gives(a, b):
1254	n/a	self.assertEqual(nntplib.decode_header(a), b)
1255	n/a	gives("" , "")
1256	n/a	gives("a plain header", "a plain header")
1257	n/a	gives(" with extra spaces ", " with extra spaces ")
1258	n/a	gives("=?ISO-8859-15?Q?D=E9buter_en_Python?=", "DÃ©buter en Python")
1259	n/a	gives("=?utf-8?q?Re=3A_=5Bsqlite=5D_probl=C3=A8me_avec_ORDER_BY_sur_des_cha?="
1260	n/a	" =?utf-8?q?=C3=AEnes_de_caract=C3=A8res_accentu=C3=A9es?=",
1261	n/a	"Re: [sqlite] problÃ¨me avec ORDER BY sur des chaÃ®nes de caractÃ¨res accentuÃ©es")
1262	n/a	gives("Re: =?UTF-8?B?cHJvYmzDqG1lIGRlIG1hdHJpY2U=?=",
1263	n/a	"Re: problÃ¨me de matrice")
1264	n/a	# A natively utf-8 header (found in the real world!)
1265	n/a	gives("Re: Message d'erreur incomprÃ©hensible (par moi)",
1266	n/a	"Re: Message d'erreur incomprÃ©hensible (par moi)")
1267	n/a
1268	n/a	def test_parse_overview_fmt(self):
1269	n/a	# The minimal (default) response
1270	n/a	lines = ["Subject:", "From:", "Date:", "Message-ID:",
1271	n/a	"References:", ":bytes", ":lines"]
1272	n/a	self.assertEqual(nntplib._parse_overview_fmt(lines),
1273	n/a	["subject", "from", "date", "message-id", "references",
1274	n/a	":bytes", ":lines"])
1275	n/a	# The minimal response using alternative names
1276	n/a	lines = ["Subject:", "From:", "Date:", "Message-ID:",
1277	n/a	"References:", "Bytes:", "Lines:"]
1278	n/a	self.assertEqual(nntplib._parse_overview_fmt(lines),
1279	n/a	["subject", "from", "date", "message-id", "references",
1280	n/a	":bytes", ":lines"])
1281	n/a	# Variations in casing
1282	n/a	lines = ["subject:", "FROM:", "DaTe:", "message-ID:",
1283	n/a	"References:", "BYTES:", "Lines:"]
1284	n/a	self.assertEqual(nntplib._parse_overview_fmt(lines),
1285	n/a	["subject", "from", "date", "message-id", "references",
1286	n/a	":bytes", ":lines"])
1287	n/a	# First example from RFC 3977
1288	n/a	lines = ["Subject:", "From:", "Date:", "Message-ID:",
1289	n/a	"References:", ":bytes", ":lines", "Xref:full",
1290	n/a	"Distribution:full"]
1291	n/a	self.assertEqual(nntplib._parse_overview_fmt(lines),
1292	n/a	["subject", "from", "date", "message-id", "references",
1293	n/a	":bytes", ":lines", "xref", "distribution"])
1294	n/a	# Second example from RFC 3977
1295	n/a	lines = ["Subject:", "From:", "Date:", "Message-ID:",
1296	n/a	"References:", "Bytes:", "Lines:", "Xref:FULL",
1297	n/a	"Distribution:FULL"]
1298	n/a	self.assertEqual(nntplib._parse_overview_fmt(lines),
1299	n/a	["subject", "from", "date", "message-id", "references",
1300	n/a	":bytes", ":lines", "xref", "distribution"])
1301	n/a	# A classic response from INN
1302	n/a	lines = ["Subject:", "From:", "Date:", "Message-ID:",
1303	n/a	"References:", "Bytes:", "Lines:", "Xref:full"]
1304	n/a	self.assertEqual(nntplib._parse_overview_fmt(lines),
1305	n/a	["subject", "from", "date", "message-id", "references",
1306	n/a	":bytes", ":lines", "xref"])
1307	n/a
1308	n/a	def test_parse_overview(self):
1309	n/a	fmt = nntplib._DEFAULT_OVERVIEW_FMT + ["xref"]
1310	n/a	# First example from RFC 3977
1311	n/a	lines = [
1312	n/a	'3000234\tI am just a test article\t"Demo User" '
1313	n/a	'<nobody@example.com>\t6 Oct 1998 04:38:40 -0500\t'
1314	n/a	'<45223423@example.com>\t<45454@example.net>\t1234\t'
1315	n/a	'17\tXref: news.example.com misc.test:3000363',
1316	n/a	]
1317	n/a	overview = nntplib._parse_overview(lines, fmt)
1318	n/a	(art_num, fields), = overview
1319	n/a	self.assertEqual(art_num, 3000234)
1320	n/a	self.assertEqual(fields, {
1321	n/a	'subject': 'I am just a test article',
1322	n/a	'from': '"Demo User" <nobody@example.com>',
1323	n/a	'date': '6 Oct 1998 04:38:40 -0500',
1324	n/a	'message-id': '<45223423@example.com>',
1325	n/a	'references': '<45454@example.net>',
1326	n/a	':bytes': '1234',
1327	n/a	':lines': '17',
1328	n/a	'xref': 'news.example.com misc.test:3000363',
1329	n/a	})
1330	n/a	# Second example; here the "Xref" field is totally absent (including
1331	n/a	# the header name) and comes out as None
1332	n/a	lines = [
1333	n/a	'3000234\tI am just a test article\t"Demo User" '
1334	n/a	'<nobody@example.com>\t6 Oct 1998 04:38:40 -0500\t'
1335	n/a	'<45223423@example.com>\t<45454@example.net>\t1234\t'
1336	n/a	'17\t\t',
1337	n/a	]
1338	n/a	overview = nntplib._parse_overview(lines, fmt)
1339	n/a	(art_num, fields), = overview
1340	n/a	self.assertEqual(fields['xref'], None)
1341	n/a	# Third example; the "Xref" is an empty string, while "references"
1342	n/a	# is a single space.
1343	n/a	lines = [
1344	n/a	'3000234\tI am just a test article\t"Demo User" '
1345	n/a	'<nobody@example.com>\t6 Oct 1998 04:38:40 -0500\t'
1346	n/a	'<45223423@example.com>\t \t1234\t'
1347	n/a	'17\tXref: \t',
1348	n/a	]
1349	n/a	overview = nntplib._parse_overview(lines, fmt)
1350	n/a	(art_num, fields), = overview
1351	n/a	self.assertEqual(fields['references'], ' ')
1352	n/a	self.assertEqual(fields['xref'], '')
1353	n/a
1354	n/a	def test_parse_datetime(self):
1355	n/a	def gives(a, b, *c):
1356	n/a	self.assertEqual(nntplib._parse_datetime(a, b),
1357	n/a	datetime.datetime(*c))
1358	n/a	# Output of DATE command
1359	n/a	gives("19990623135624", None, 1999, 6, 23, 13, 56, 24)
1360	n/a	# Variations
1361	n/a	gives("19990623", "135624", 1999, 6, 23, 13, 56, 24)
1362	n/a	gives("990623", "135624", 1999, 6, 23, 13, 56, 24)
1363	n/a	gives("090623", "135624", 2009, 6, 23, 13, 56, 24)
1364	n/a
1365	n/a	def test_unparse_datetime(self):
1366	n/a	# Test non-legacy mode
1367	n/a	# 1) with a datetime
1368	n/a	def gives(y, M, d, h, m, s, date_str, time_str):
1369	n/a	dt = datetime.datetime(y, M, d, h, m, s)
1370	n/a	self.assertEqual(nntplib._unparse_datetime(dt),
1371	n/a	(date_str, time_str))
1372	n/a	self.assertEqual(nntplib._unparse_datetime(dt, False),
1373	n/a	(date_str, time_str))
1374	n/a	gives(1999, 6, 23, 13, 56, 24, "19990623", "135624")
1375	n/a	gives(2000, 6, 23, 13, 56, 24, "20000623", "135624")
1376	n/a	gives(2010, 6, 5, 1, 2, 3, "20100605", "010203")
1377	n/a	# 2) with a date
1378	n/a	def gives(y, M, d, date_str, time_str):
1379	n/a	dt = datetime.date(y, M, d)
1380	n/a	self.assertEqual(nntplib._unparse_datetime(dt),
1381	n/a	(date_str, time_str))
1382	n/a	self.assertEqual(nntplib._unparse_datetime(dt, False),
1383	n/a	(date_str, time_str))
1384	n/a	gives(1999, 6, 23, "19990623", "000000")
1385	n/a	gives(2000, 6, 23, "20000623", "000000")
1386	n/a	gives(2010, 6, 5, "20100605", "000000")
1387	n/a
1388	n/a	def test_unparse_datetime_legacy(self):
1389	n/a	# Test legacy mode (RFC 977)
1390	n/a	# 1) with a datetime
1391	n/a	def gives(y, M, d, h, m, s, date_str, time_str):
1392	n/a	dt = datetime.datetime(y, M, d, h, m, s)
1393	n/a	self.assertEqual(nntplib._unparse_datetime(dt, True),
1394	n/a	(date_str, time_str))
1395	n/a	gives(1999, 6, 23, 13, 56, 24, "990623", "135624")
1396	n/a	gives(2000, 6, 23, 13, 56, 24, "000623", "135624")
1397	n/a	gives(2010, 6, 5, 1, 2, 3, "100605", "010203")
1398	n/a	# 2) with a date
1399	n/a	def gives(y, M, d, date_str, time_str):
1400	n/a	dt = datetime.date(y, M, d)
1401	n/a	self.assertEqual(nntplib._unparse_datetime(dt, True),
1402	n/a	(date_str, time_str))
1403	n/a	gives(1999, 6, 23, "990623", "000000")
1404	n/a	gives(2000, 6, 23, "000623", "000000")
1405	n/a	gives(2010, 6, 5, "100605", "000000")
1406	n/a
1407	n/a	@unittest.skipUnless(ssl, 'requires SSL support')
1408	n/a	def test_ssl_support(self):
1409	n/a	self.assertTrue(hasattr(nntplib, 'NNTP_SSL'))
1410	n/a
1411	n/a
1412	n/a	class PublicAPITests(unittest.TestCase):
1413	n/a	"""Ensures that the correct values are exposed in the public API."""
1414	n/a
1415	n/a	def test_module_all_attribute(self):
1416	n/a	self.assertTrue(hasattr(nntplib, '__all__'))
1417	n/a	target_api = ['NNTP', 'NNTPError', 'NNTPReplyError',
1418	n/a	'NNTPTemporaryError', 'NNTPPermanentError',
1419	n/a	'NNTPProtocolError', 'NNTPDataError', 'decode_header']
1420	n/a	if ssl is not None:
1421	n/a	target_api.append('NNTP_SSL')
1422	n/a	self.assertEqual(set(nntplib.__all__), set(target_api))
1423	n/a
1424	n/a	class MockSocketTests(unittest.TestCase):
1425	n/a	"""Tests involving a mock socket object
1426	n/a
1427	n/a	Used where the _NNTPServerIO file object is not enough."""
1428	n/a
1429	n/a	nntp_class = nntplib.NNTP
1430	n/a
1431	n/a	def check_constructor_error_conditions(
1432	n/a	self, handler_class,
1433	n/a	expected_error_type, expected_error_msg,
1434	n/a	login=None, password=None):
1435	n/a
1436	n/a	class mock_socket_module:
1437	n/a	def create_connection(address, timeout):
1438	n/a	return MockSocket()
1439	n/a
1440	n/a	class MockSocket:
1441	n/a	def close(self):
1442	n/a	nonlocal socket_closed
1443	n/a	socket_closed = True
1444	n/a
1445	n/a	def makefile(socket, mode):
1446	n/a	handler = handler_class()
1447	n/a	_, file = make_mock_file(handler)
1448	n/a	files.append(file)
1449	n/a	return file
1450	n/a
1451	n/a	socket_closed = False
1452	n/a	files = []
1453	n/a	with patch('nntplib.socket', mock_socket_module), \
1454	n/a	self.assertRaisesRegex(expected_error_type, expected_error_msg):
1455	n/a	self.nntp_class('dummy', user=login, password=password)
1456	n/a	self.assertTrue(socket_closed)
1457	n/a	for f in files:
1458	n/a	self.assertTrue(f.closed)
1459	n/a
1460	n/a	def test_bad_welcome(self):
1461	n/a	#Test a bad welcome message
1462	n/a	class Handler(NNTPv1Handler):
1463	n/a	welcome = 'Bad Welcome'
1464	n/a	self.check_constructor_error_conditions(
1465	n/a	Handler, nntplib.NNTPProtocolError, Handler.welcome)
1466	n/a
1467	n/a	def test_service_temporarily_unavailable(self):
1468	n/a	#Test service temporarily unavailable
1469	n/a	class Handler(NNTPv1Handler):
1470	n/a	welcome = '400 Service temporarily unavailable'
1471	n/a	self.check_constructor_error_conditions(
1472	n/a	Handler, nntplib.NNTPTemporaryError, Handler.welcome)
1473	n/a
1474	n/a	def test_service_permanently_unavailable(self):
1475	n/a	#Test service permanently unavailable
1476	n/a	class Handler(NNTPv1Handler):
1477	n/a	welcome = '502 Service permanently unavailable'
1478	n/a	self.check_constructor_error_conditions(
1479	n/a	Handler, nntplib.NNTPPermanentError, Handler.welcome)
1480	n/a
1481	n/a	def test_bad_capabilities(self):
1482	n/a	#Test a bad capabilities response
1483	n/a	class Handler(NNTPv1Handler):
1484	n/a	def handle_CAPABILITIES(self):
1485	n/a	self.push_lit(capabilities_response)
1486	n/a	capabilities_response = '201 bad capability'
1487	n/a	self.check_constructor_error_conditions(
1488	n/a	Handler, nntplib.NNTPReplyError, capabilities_response)
1489	n/a
1490	n/a	def test_login_aborted(self):
1491	n/a	#Test a bad authinfo response
1492	n/a	login = 't@e.com'
1493	n/a	password = 'python'
1494	n/a	class Handler(NNTPv1Handler):
1495	n/a	def handle_AUTHINFO(self, *args):
1496	n/a	self.push_lit(authinfo_response)
1497	n/a	authinfo_response = '503 Mechanism not recognized'
1498	n/a	self.check_constructor_error_conditions(
1499	n/a	Handler, nntplib.NNTPPermanentError, authinfo_response,
1500	n/a	login, password)
1501	n/a
1502	n/a	class bypass_context:
1503	n/a	"""Bypass encryption and actual SSL module"""
1504	n/a	def wrap_socket(sock, **args):
1505	n/a	return sock
1506	n/a
1507	n/a	@unittest.skipUnless(ssl, 'requires SSL support')
1508	n/a	class MockSslTests(MockSocketTests):
1509	n/a	@staticmethod
1510	n/a	def nntp_class(pos, *kw):
1511	n/a	return nntplib.NNTP_SSL(pos, ssl_context=bypass_context, *kw)
1512	n/a
1513	n/a	@unittest.skipUnless(threading, 'requires multithreading')
1514	n/a	class LocalServerTests(unittest.TestCase):
1515	n/a	def setUp(self):
1516	n/a	sock = socket.socket()
1517	n/a	port = support.bind_port(sock)
1518	n/a	sock.listen()
1519	n/a	self.background = threading.Thread(
1520	n/a	target=self.run_server, args=(sock,))
1521	n/a	self.background.start()
1522	n/a	self.addCleanup(self.background.join)
1523	n/a
1524	n/a	self.nntp = NNTP(support.HOST, port, usenetrc=False).__enter__()
1525	n/a	self.addCleanup(self.nntp.__exit__, None, None, None)
1526	n/a
1527	n/a	def run_server(self, sock):
1528	n/a	# Could be generalized to handle more commands in separate methods
1529	n/a	with sock:
1530	n/a	[client, _] = sock.accept()
1531	n/a	with contextlib.ExitStack() as cleanup:
1532	n/a	cleanup.enter_context(client)
1533	n/a	reader = cleanup.enter_context(client.makefile('rb'))
1534	n/a	client.sendall(b'200 Server ready\r\n')
1535	n/a	while True:
1536	n/a	cmd = reader.readline()
1537	n/a	if cmd == b'CAPABILITIES\r\n':
1538	n/a	client.sendall(
1539	n/a	b'101 Capability list:\r\n'
1540	n/a	b'VERSION 2\r\n'
1541	n/a	b'STARTTLS\r\n'
1542	n/a	b'.\r\n'
1543	n/a	)
1544	n/a	elif cmd == b'STARTTLS\r\n':
1545	n/a	reader.close()
1546	n/a	client.sendall(b'382 Begin TLS negotiation now\r\n')
1547	n/a	context = ssl.SSLContext()
1548	n/a	context.load_cert_chain(certfile)
1549	n/a	client = context.wrap_socket(
1550	n/a	client, server_side=True)
1551	n/a	cleanup.enter_context(client)
1552	n/a	reader = cleanup.enter_context(client.makefile('rb'))
1553	n/a	elif cmd == b'QUIT\r\n':
1554	n/a	client.sendall(b'205 Bye!\r\n')
1555	n/a	break
1556	n/a	else:
1557	n/a	raise ValueError('Unexpected command {!r}'.format(cmd))
1558	n/a
1559	n/a	@unittest.skipUnless(ssl, 'requires SSL support')
1560	n/a	def test_starttls(self):
1561	n/a	file = self.nntp.file
1562	n/a	sock = self.nntp.sock
1563	n/a	self.nntp.starttls()
1564	n/a	# Check that the socket and internal pseudo-file really were
1565	n/a	# changed.
1566	n/a	self.assertNotEqual(file, self.nntp.file)
1567	n/a	self.assertNotEqual(sock, self.nntp.sock)
1568	n/a	# Check that the new socket really is an SSL one
1569	n/a	self.assertIsInstance(self.nntp.sock, ssl.SSLSocket)
1570	n/a	# Check that trying starttls when it's already active fails.
1571	n/a	self.assertRaises(ValueError, self.nntp.starttls)
1572	n/a
1573	n/a
1574	n/a	if __name__ == "__main__":
1575	n/a	unittest.main()