Issue #9360: Cleanup and improvements to the nntplib module. The API

now conforms to the philosophy of bytes and unicode separation in Python 3.
A test suite has also been added.

1 files changed, 1091 insertions, 0 deletions

diff --git a/Lib/test/test_nntplib.py b/Lib/test/test_nntplib.py
new file mode 100644
index 0000000..512bcd5
--- /dev/null
+++ b/Lib/test/test_nntplib.py
@@ -0,0 +1,1091 @@
+import io
+import datetime
+import textwrap
+import unittest
+import contextlib
+from test import support
+from nntplib import NNTP, GroupInfo
+import nntplib
+
+TIMEOUT = 30
+
+# TODO:
+# - test the `file` arg to more commands
+# - test error conditions
+
+
+class NetworkedNNTPTestsMixin:
+
+    def test_welcome(self):
+        welcome = self.server.getwelcome()
+        self.assertEqual(str, type(welcome))
+
+    def test_help(self):
+        resp, list = self.server.help()
+        self.assertTrue(resp.startswith("100 "), resp)
+        for line in list:
+            self.assertEqual(str, type(line))
+
+    def test_list(self):
+        resp, list = self.server.list()
+        if len(list) > 0:
+            self.assertEqual(GroupInfo, type(list[0]))
+            self.assertEqual(str, type(list[0].group))
+
+    def test_unknown_command(self):
+        with self.assertRaises(nntplib.NNTPPermanentError) as cm:
+            self.server._shortcmd("XYZZY")
+        resp = cm.exception.response
+        self.assertTrue(resp.startswith("500 "), resp)
+
+    def test_newgroups(self):
+        # gmane gets a constant influx of new groups.  In order not to stress
+        # the server too much, we choose a recent date in the past.
+        dt = datetime.date.today() - datetime.timedelta(days=7)
+        resp, groups = self.server.newgroups(dt)
+        if len(groups) > 0:
+            self.assertIsInstance(groups[0], GroupInfo)
+            self.assertIsInstance(groups[0].group, str)
+
+    def test_description(self):
+        def _check_desc(desc):
+            # Sanity checks
+            self.assertIsInstance(desc, str)
+            self.assertNotIn(self.GROUP_NAME, desc)
+        desc = self.server.description(self.GROUP_NAME)
+        _check_desc(desc)
+        # Another sanity check
+        self.assertIn("Python", desc)
+        # With a pattern
+        desc = self.server.description(self.GROUP_PAT)
+        _check_desc(desc)
+        # Shouldn't exist
+        desc = self.server.description("zk.brrtt.baz")
+        self.assertEqual(desc, '')
+
+    def test_descriptions(self):
+        resp, descs = self.server.descriptions(self.GROUP_PAT)
+        # 215 for LIST NEWSGROUPS, 282 for XGTITLE
+        self.assertTrue(
+            resp.startswith("215 ") or resp.startswith("282 "), resp)
+        self.assertIsInstance(descs, dict)
+        desc = descs[self.GROUP_NAME]
+        self.assertEqual(desc, self.server.description(self.GROUP_NAME))
+
+    def test_group(self):
+        result = self.server.group(self.GROUP_NAME)
+        self.assertEqual(5, len(result))
+        resp, count, first, last, group = result
+        self.assertEqual(group, self.GROUP_NAME)
+        self.assertIsInstance(count, int)
+        self.assertIsInstance(first, int)
+        self.assertIsInstance(last, int)
+        self.assertLessEqual(first, last)
+        self.assertTrue(resp.startswith("211 "), resp)
+
+    def test_date(self):
+        resp, date = self.server.date()
+        self.assertIsInstance(date, datetime.datetime)
+        # Sanity check
+        self.assertGreaterEqual(date.year, 1995)
+        self.assertLessEqual(date.year, 2030)
+
+    def _check_art_dict(self, art_dict):
+        # Some sanity checks for a field dictionary returned by OVER / XOVER
+        self.assertIsInstance(art_dict, dict)
+        # NNTP has 7 mandatory fields
+        self.assertGreaterEqual(art_dict.keys(),
+            {"subject", "from", "date", "message-id",
+             "references", ":bytes", ":lines"}
+            )
+        for v in art_dict.values():
+            self.assertIsInstance(v, str)
+
+    def test_xover(self):
+        resp, count, first, last, name = self.server.group(self.GROUP_NAME)
+        resp, lines = self.server.xover(last, last)
+        art_num, art_dict = lines[0]
+        self.assertEqual(art_num, last)
+        self._check_art_dict(art_dict)
+
+    def test_over(self):
+        resp, count, first, last, name = self.server.group(self.GROUP_NAME)
+        start = last - 10
+        # The "start-" article range form
+        resp, lines = self.server.over((start, None))
+        art_num, art_dict = lines[0]
+        self._check_art_dict(art_dict)
+        # The "start-end" article range form
+        resp, lines = self.server.over((start, last))
+        art_num, art_dict = lines[-1]
+        self.assertEqual(art_num, last)
+        self._check_art_dict(art_dict)
+        # XXX The "message_id" form is unsupported by gmane
+        # 503 Overview by message-ID unsupported
+
+    def test_xhdr(self):
+        resp, count, first, last, name = self.server.group(self.GROUP_NAME)
+        resp, lines = self.server.xhdr('subject', last)
+        for line in lines:
+            self.assertEqual(str, type(line[1]))
+
+    def check_article_resp(self, resp, article, art_num=None):
+        self.assertIsInstance(article, nntplib.ArticleInfo)
+        if art_num is not None:
+            self.assertEqual(article.number, art_num)
+        for line in article.lines:
+            self.assertIsInstance(line, bytes)
+        # XXX this could exceptionally happen...
+        self.assertNotIn(article.lines[-1], (b".", b".\n", b".\r\n"))
+
+    def test_article_head_body(self):
+        resp, count, first, last, name = self.server.group(self.GROUP_NAME)
+        resp, head = self.server.head(last)
+        self.assertTrue(resp.startswith("221 "), resp)
+        self.check_article_resp(resp, head, last)
+        resp, body = self.server.body(last)
+        self.assertTrue(resp.startswith("222 "), resp)
+        self.check_article_resp(resp, body, last)
+        resp, article = self.server.article(last)
+        self.assertTrue(resp.startswith("220 "), resp)
+        self.check_article_resp(resp, article, last)
+        self.assertEqual(article.lines, head.lines + [b''] + body.lines)
+
+    def test_quit(self):
+        self.server.quit()
+        self.server = None
+
+
+class NetworkedNNTPTests(NetworkedNNTPTestsMixin, unittest.TestCase):
+    NNTP_HOST = 'news.gmane.org'
+    GROUP_NAME = 'gmane.comp.python.devel'
+    GROUP_PAT = 'gmane.comp.python.d*'
+
+    def setUp(self):
+        support.requires("network")
+        with support.transient_internet(self.NNTP_HOST):
+            self.server = NNTP(self.NNTP_HOST, timeout=TIMEOUT)
+
+    def tearDown(self):
+        if self.server is not None:
+            self.server.quit()
+
+    # Disabled with gmane as it produces too much data
+    test_list = None
+
+    def test_capabilities(self):
+        # As of this writing, gmane implements NNTP version 2 and has a
+        # couple of well-known capabilities. Just sanity check that we
+        # got them.
+        def _check_caps(caps):
+            caps_list = caps['LIST']
+            self.assertIsInstance(caps_list, (list, tuple))
+            self.assertIn('OVERVIEW.FMT', caps_list)
+        self.assertGreaterEqual(self.server.nntp_version, 2)
+        _check_caps(self.server.getcapabilities())
+        # This re-emits the command
+        resp, caps = self.server.capabilities()
+        _check_caps(caps)
+
+
+#
+# Non-networked tests using a local server (or something mocking it).
+#
+
+class _NNTPServerIO(io.RawIOBase):
+    """A raw IO object allowing NNTP commands to be received and processed
+    by a handler.  The handler can push responses which can then be read
+    from the IO object."""
+
+    def __init__(self, handler):
+        io.RawIOBase.__init__(self)
+        # The channel from the client
+        self.c2s = io.BytesIO()
+        # The channel to the client
+        self.s2c = io.BytesIO()
+        self.handler = handler
+        self.handler.start(self.c2s.readline, self.push_data)
+
+    def readable(self):
+        return True
+
+    def writable(self):
+        return True
+
+    def push_data(self, data):
+        """Push (buffer) some data to send to the client."""
+        pos = self.s2c.tell()
+        self.s2c.seek(0, 2)
+        self.s2c.write(data)
+        self.s2c.seek(pos)
+
+    def write(self, b):
+        """The client sends us some data"""
+        pos = self.c2s.tell()
+        self.c2s.write(b)
+        self.c2s.seek(pos)
+        self.handler.process_pending()
+        return len(b)
+
+    def readinto(self, buf):
+        """The client wants to read a response"""
+        self.handler.process_pending()
+        b = self.s2c.read(len(buf))
+        n = len(b)
+        buf[:n] = b
+        return n
+
+
+class MockedNNTPTestsMixin:
+    # Override in derived classes
+    handler_class = None
+
+    def setUp(self):
+        super().setUp()
+        self.make_server()
+
+    def tearDown(self):
+        super().tearDown()
+        del self.server
+
+    def make_server(self, *args, **kwargs):
+        self.handler = self.handler_class()
+        self.sio = _NNTPServerIO(self.handler)
+        # Using BufferedRWPair instead of BufferedRandom ensures the file
+        # isn't seekable.
+        file = io.BufferedRWPair(self.sio, self.sio)
+        self.server = nntplib._NNTPBase(file, *args, **kwargs)
+        return self.server
+
+
+class NNTPv1Handler:
+    """A handler for RFC 977"""
+
+    welcome = "200 NNTP mock server"
+
+    def start(self, readline, push_data):
+        self.in_body = False
+        self.allow_posting = True
+        self._readline = readline
+        self._push_data = push_data
+        # Our welcome
+        self.handle_welcome()
+
+    def _decode(self, data):
+        return str(data, "utf-8", "surrogateescape")
+
+    def process_pending(self):
+        if self.in_body:
+            while True:
+                line = self._readline()
+                if not line:
+                    return
+                self.body.append(line)
+                if line == b".\r\n":
+                    break
+            try:
+                meth, tokens = self.body_callback
+                meth(*tokens, body=self.body)
+            finally:
+                self.body_callback = None
+                self.body = None
+                self.in_body = False
+        while True:
+            line = self._decode(self._readline())
+            if not line:
+                return
+            if not line.endswith("\r\n"):
+                raise ValueError("line doesn't end with \\r\\n: {!r}".format(line))
+            line = line[:-2]
+            cmd, *tokens = line.split()
+            #meth = getattr(self.handler, "handle_" + cmd.upper(), None)
+            meth = getattr(self, "handle_" + cmd.upper(), None)
+            if meth is None:
+                self.handle_unknown()
+            else:
+                try:
+                    meth(*tokens)
+                except Exception as e:
+                    raise ValueError("command failed: {!r}".format(line)) from e
+                else:
+                    if self.in_body:
+                        self.body_callback = meth, tokens
+                        self.body = []
+
+    def expect_body(self):
+        """Flag that the client is expected to post a request body"""
+        self.in_body = True
+
+    def push_data(self, data):
+        """Push some binary data"""
+        self._push_data(data)
+
+    def push_lit(self, lit):
+        """Push a string literal"""
+        lit = textwrap.dedent(lit)
+        lit = "\r\n".join(lit.splitlines()) + "\r\n"
+        lit = lit.encode('utf-8')
+        self.push_data(lit)
+
+    def handle_unknown(self):
+        self.push_lit("500 What?")
+
+    def handle_welcome(self):
+        self.push_lit(self.welcome)
+
+    def handle_QUIT(self):
+        self.push_lit("205 Bye!")
+
+    def handle_DATE(self):
+        self.push_lit("111 20100914001155")
+
+    def handle_GROUP(self, group):
+        if group == "fr.comp.lang.python":
+            self.push_lit("211 486 761 1265 fr.comp.lang.python")
+        else:
+            self.push_lit("411 No such group {}".format(group))
+
+    def handle_HELP(self):
+        self.push_lit("""\
+            100 Legal commands
+              authinfo user Name|pass Password|generic <prog> <args>
+              date
+              help
+            Report problems to <root@example.org>
+            .""")
+
+    def handle_STAT(self, message_spec=None):
+        if message_spec is None:
+            self.push_lit("412 No newsgroup selected")
+        elif message_spec == "3000234":
+            self.push_lit("223 3000234 <45223423@example.com>")
+        elif message_spec == "<45223423@example.com>":
+            self.push_lit("223 0 <45223423@example.com>")
+        else:
+            self.push_lit("430 No Such Article Found")
+
+    def handle_NEXT(self):
+        self.push_lit("223 3000237 <668929@example.org> retrieved")
+
+    def handle_LAST(self):
+        self.push_lit("223 3000234 <45223423@example.com> retrieved")
+
+    def handle_LIST(self, action=None, param=None):
+        if action is None:
+            self.push_lit("""\
+                215 Newsgroups in form "group high low flags".
+                comp.lang.python 0000052340 0000002828 y
+                comp.lang.python.announce 0000001153 0000000993 m
+                free.it.comp.lang.python 0000000002 0000000002 y
+                fr.comp.lang.python 0000001254 0000000760 y
+                free.it.comp.lang.python.learner 0000000000 0000000001 y
+                tw.bbs.comp.lang.python 0000000304 0000000304 y
+                .""")
+        elif action == "OVERVIEW.FMT":
+            self.push_lit("""\
+                215 Order of fields in overview database.
+                Subject:
+                From:
+                Date:
+                Message-ID:
+                References:
+                Bytes:
+                Lines:
+                Xref:full
+                .""")
+        elif action == "NEWSGROUPS":
+            assert param is not None
+            if param == "comp.lang.python":
+                self.push_lit("""\
+                    215 Descriptions in form "group description".
+                    comp.lang.python\tThe Python computer language.
+                    .""")
+            elif param == "comp.lang.python*":
+                self.push_lit("""\
+                    215 Descriptions in form "group description".
+                    comp.lang.python.announce\tAnnouncements about the Python language. (Moderated)
+                    comp.lang.python\tThe Python computer language.
+                    .""")
+            else:
+                self.push_lit("""\
+                    215 Descriptions in form "group description".
+                    .""")
+        else:
+            self.push_lit('501 Unknown LIST keyword')
+
+    def handle_NEWNEWS(self, group, date_str, time_str):
+        # We hard code different return messages depending on passed
+        # argument and date syntax.
+        if (group == "comp.lang.python" and date_str == "20100913"
+            and time_str == "082004"):
+            # Date was passed in RFC 3977 format (NNTP "v2")
+            self.push_lit("""\
+                230 list of newsarticles (NNTP v2) created after Mon Sep 13 08:20:04 2010 follows
+                <a4929a40-6328-491a-aaaf-cb79ed7309a2@q2g2000vbk.googlegroups.com>
+                <f30c0419-f549-4218-848f-d7d0131da931@y3g2000vbm.googlegroups.com>
+                .""")
+        elif (group == "comp.lang.python" and date_str == "100913"
+            and time_str == "082004"):
+            # Date was passed in RFC 977 format (NNTP "v1")
+            self.push_lit("""\
+                230 list of newsarticles (NNTP v1) created after Mon Sep 13 08:20:04 2010 follows
+                <a4929a40-6328-491a-aaaf-cb79ed7309a2@q2g2000vbk.googlegroups.com>
+                <f30c0419-f549-4218-848f-d7d0131da931@y3g2000vbm.googlegroups.com>
+                .""")
+        else:
+            self.push_lit("""\
+                230 An empty list of newsarticles follows
+                .""")
+        # (Note for experiments: many servers disable NEWNEWS.
+        #  As of this writing, sicinfo3.epfl.ch doesn't.)
+
+    def handle_XOVER(self, message_spec):
+        if message_spec == "57-59":
+            self.push_lit(
+                "224 Overview information for 57-58 follows\n"
+                "57\tRe: ANN: New Plone book with strong Python (and Zope) themes throughout"
+                    "\tDoug Hellmann <doug.hellmann-Re5JQEeQqe8AvxtiuMwx3w@public.gmane.org>"
+                    "\tSat, 19 Jun 2010 18:04:08 -0400"
+                    "\t<4FD05F05-F98B-44DC-8111-C6009C925F0C@gmail.com>"
+                    "\t<hvalf7$ort$1@dough.gmane.org>\t7103\t16"
+                    "\tXref: news.gmane.org gmane.comp.python.authors:57"
+                    "\n"
+                "58\tLooking for a few good bloggers"
+                    "\tDoug Hellmann <doug.hellmann-Re5JQEeQqe8AvxtiuMwx3w@public.gmane.org>"
+                    "\tThu, 22 Jul 2010 09:14:14 -0400"
+                    "\t<A29863FA-F388-40C3-AA25-0FD06B09B5BF@gmail.com>"
+                    "\t\t6683\t16"
+                    "\tXref: news.gmane.org gmane.comp.python.authors:58"
+                    "\n"
+                # An UTF-8 overview line from fr.comp.lang.python
+                "59\tRe: Message d'erreur incompréhensible (par moi)"
+                    "\tEric Brunel <eric.brunel@pragmadev.nospam.com>"
+                    "\tWed, 15 Sep 2010 18:09:15 +0200"
+                    "\t<eric.brunel-2B8B56.18091515092010@news.wanadoo.fr>"
+                    "\t<4c90ec87$0$32425$ba4acef3@reader.news.orange.fr>\t1641\t27"
+                    "\tXref: saria.nerim.net fr.comp.lang.python:1265"
+                    "\n"
+                ".\n")
+        else:
+            self.push_lit("""\
+                224 No articles
+                .""")
+
+    def handle_POST(self, *, body=None):
+        if body is None:
+            if self.allow_posting:
+                self.push_lit("340 Input article; end with <CR-LF>.<CR-LF>")
+                self.expect_body()
+            else:
+                self.push_lit("440 Posting not permitted")
+        else:
+            assert self.allow_posting
+            self.push_lit("240 Article received OK")
+            self.posted_body = body
+
+    def handle_IHAVE(self, message_id, *, body=None):
+        if body is None:
+            if (self.allow_posting and
+                message_id == "<i.am.an.article.you.will.want@example.com>"):
+                self.push_lit("335 Send it; end with <CR-LF>.<CR-LF>")
+                self.expect_body()
+            else:
+                self.push_lit("435 Article not wanted")
+        else:
+            assert self.allow_posting
+            self.push_lit("235 Article transferred OK")
+            self.posted_body = body
+
+    sample_head = """\
+        From: "Demo User" <nobody@example.net>
+        Subject: I am just a test article
+        Content-Type: text/plain; charset=UTF-8; format=flowed
+        Message-ID: <i.am.an.article.you.will.want@example.com>"""
+
+    sample_body = """\
+        This is just a test article.
+        ..Here is a dot-starting line.
+
+        -- Signed by Andr\xe9."""
+
+    sample_article = sample_head + "\n\n" + sample_body
+
+    def handle_ARTICLE(self, message_spec=None):
+        if message_spec is None:
+            self.push_lit("220 3000237 <45223423@example.com>")
+        elif message_spec == "<45223423@example.com>":
+            self.push_lit("220 0 <45223423@example.com>")
+        elif message_spec == "3000234":
+            self.push_lit("220 3000234 <45223423@example.com>")
+        else:
+            self.push_lit("430 No Such Article Found")
+            return
+        self.push_lit(self.sample_article)
+        self.push_lit(".")
+
+    def handle_HEAD(self, message_spec=None):
+        if message_spec is None:
+            self.push_lit("221 3000237 <45223423@example.com>")
+        elif message_spec == "<45223423@example.com>":
+            self.push_lit("221 0 <45223423@example.com>")
+        elif message_spec == "3000234":
+            self.push_lit("221 3000234 <45223423@example.com>")
+        else:
+            self.push_lit("430 No Such Article Found")
+            return
+        self.push_lit(self.sample_head)
+        self.push_lit(".")
+
+    def handle_BODY(self, message_spec=None):
+        if message_spec is None:
+            self.push_lit("222 3000237 <45223423@example.com>")
+        elif message_spec == "<45223423@example.com>":
+            self.push_lit("222 0 <45223423@example.com>")
+        elif message_spec == "3000234":
+            self.push_lit("222 3000234 <45223423@example.com>")
+        else:
+            self.push_lit("430 No Such Article Found")
+            return
+        self.push_lit(self.sample_body)
+        self.push_lit(".")
+
+
+class NNTPv2Handler(NNTPv1Handler):
+    """A handler for RFC 3977 (NNTP "v2")"""
+
+    def handle_CAPABILITIES(self):
+        self.push_lit("""\
+            101 Capability list:
+            VERSION 2
+            IMPLEMENTATION INN 2.5.1
+            AUTHINFO USER
+            HDR
+            LIST ACTIVE ACTIVE.TIMES DISTRIB.PATS HEADERS NEWSGROUPS OVERVIEW.FMT
+            OVER
+            POST
+            READER
+            .""")
+
+    def handle_OVER(self, message_spec=None):
+        return self.handle_XOVER(message_spec)
+
+
+class NNTPv1v2TestsMixin:
+
+    def setUp(self):
+        super().setUp()
+
+    def test_welcome(self):
+        self.assertEqual(self.server.welcome, self.handler.welcome)
+
+    def test_date(self):
+        resp, date = self.server.date()
+        self.assertEqual(resp, "111 20100914001155")
+        self.assertEqual(date, datetime.datetime(2010, 9, 14, 0, 11, 55))
+
+    def test_quit(self):
+        self.assertFalse(self.sio.closed)
+        resp = self.server.quit()
+        self.assertEqual(resp, "205 Bye!")
+        self.assertTrue(self.sio.closed)
+
+    def test_help(self):
+        resp, help = self.server.help()
+        self.assertEqual(resp, "100 Legal commands")
+        self.assertEqual(help, [
+            '  authinfo user Name|pass Password|generic <prog> <args>',
+            '  date',
+            '  help',
+            'Report problems to <root@example.org>',
+        ])
+
+    def test_list(self):
+        resp, groups = self.server.list()
+        self.assertEqual(len(groups), 6)
+        g = groups[1]
+        self.assertEqual(g,
+            GroupInfo("comp.lang.python.announce", "0000001153",
+                      "0000000993", "m"))
+
+    def test_stat(self):
+        resp, art_num, message_id = self.server.stat(3000234)
+        self.assertEqual(resp, "223 3000234 <45223423@example.com>")
+        self.assertEqual(art_num, 3000234)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        resp, art_num, message_id = self.server.stat("<45223423@example.com>")
+        self.assertEqual(resp, "223 0 <45223423@example.com>")
+        self.assertEqual(art_num, 0)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.stat("<non.existent.id>")
+        self.assertEqual(cm.exception.response, "430 No Such Article Found")
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.stat()
+        self.assertEqual(cm.exception.response, "412 No newsgroup selected")
+
+    def test_next(self):
+        resp, art_num, message_id = self.server.next()
+        self.assertEqual(resp, "223 3000237 <668929@example.org> retrieved")
+        self.assertEqual(art_num, 3000237)
+        self.assertEqual(message_id, "<668929@example.org>")
+
+    def test_last(self):
+        resp, art_num, message_id = self.server.last()
+        self.assertEqual(resp, "223 3000234 <45223423@example.com> retrieved")
+        self.assertEqual(art_num, 3000234)
+        self.assertEqual(message_id, "<45223423@example.com>")
+
+    def test_description(self):
+        desc = self.server.description("comp.lang.python")
+        self.assertEqual(desc, "The Python computer language.")
+        desc = self.server.description("comp.lang.pythonx")
+        self.assertEqual(desc, "")
+
+    def test_descriptions(self):
+        resp, groups = self.server.descriptions("comp.lang.python")
+        self.assertEqual(resp, '215 Descriptions in form "group description".')
+        self.assertEqual(groups, {
+            "comp.lang.python": "The Python computer language.",
+            })
+        resp, groups = self.server.descriptions("comp.lang.python*")
+        self.assertEqual(groups, {
+            "comp.lang.python": "The Python computer language.",
+            "comp.lang.python.announce": "Announcements about the Python language. (Moderated)",
+            })
+        resp, groups = self.server.descriptions("comp.lang.pythonx")
+        self.assertEqual(groups, {})
+
+    def test_group(self):
+        resp, count, first, last, group = self.server.group("fr.comp.lang.python")
+        self.assertTrue(resp.startswith("211 "), resp)
+        self.assertEqual(first, 761)
+        self.assertEqual(last, 1265)
+        self.assertEqual(count, 486)
+        self.assertEqual(group, "fr.comp.lang.python")
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.group("comp.lang.python.devel")
+        exc = cm.exception
+        self.assertTrue(exc.response.startswith("411 No such group"),
+                        exc.response)
+
+    def test_newnews(self):
+        # NEWNEWS comp.lang.python [20]100913 082004
+        dt = datetime.datetime(2010, 9, 13, 8, 20, 4)
+        resp, ids = self.server.newnews("comp.lang.python", dt)
+        expected = (
+            "230 list of newsarticles (NNTP v{0}) "
+            "created after Mon Sep 13 08:20:04 2010 follows"
+            ).format(self.nntp_version)
+        self.assertEqual(resp, expected)
+        self.assertEqual(ids, [
+            "<a4929a40-6328-491a-aaaf-cb79ed7309a2@q2g2000vbk.googlegroups.com>",
+            "<f30c0419-f549-4218-848f-d7d0131da931@y3g2000vbm.googlegroups.com>",
+            ])
+        # NEWNEWS fr.comp.lang.python [20]100913 082004
+        dt = datetime.datetime(2010, 9, 13, 8, 20, 4)
+        resp, ids = self.server.newnews("fr.comp.lang.python", dt)
+        self.assertEqual(resp, "230 An empty list of newsarticles follows")
+        self.assertEqual(ids, [])
+
+    def _check_article_body(self, lines):
+        self.assertEqual(len(lines), 4)
+        self.assertEqual(lines[-1].decode('utf8'), "-- Signed by André.")
+        self.assertEqual(lines[-2], b"")
+        self.assertEqual(lines[-3], b".Here is a dot-starting line.")
+        self.assertEqual(lines[-4], b"This is just a test article.")
+
+    def _check_article_head(self, lines):
+        self.assertEqual(len(lines), 4)
+        self.assertEqual(lines[0], b'From: "Demo User" <nobody@example.net>')
+        self.assertEqual(lines[3], b"Message-ID: <i.am.an.article.you.will.want@example.com>")
+
+    def _check_article_data(self, lines):
+        self.assertEqual(len(lines), 9)
+        self._check_article_head(lines[:4])
+        self._check_article_body(lines[-4:])
+        self.assertEqual(lines[4], b"")
+
+    def test_article(self):
+        # ARTICLE
+        resp, info = self.server.article()
+        self.assertEqual(resp, "220 3000237 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000237)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_data(lines)
+        # ARTICLE num
+        resp, info = self.server.article(3000234)
+        self.assertEqual(resp, "220 3000234 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000234)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_data(lines)
+        # ARTICLE id
+        resp, info = self.server.article("<45223423@example.com>")
+        self.assertEqual(resp, "220 0 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 0)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_data(lines)
+        # Non-existent id
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.article("<non-existent@example.com>")
+        self.assertEqual(cm.exception.response, "430 No Such Article Found")
+
+    def test_article_file(self):
+        # With a "file" argument
+        f = io.BytesIO()
+        resp, info = self.server.article(file=f)
+        self.assertEqual(resp, "220 3000237 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000237)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self.assertEqual(lines, [])
+        data = f.getvalue()
+        self.assertTrue(data.startswith(
+            b'From: "Demo User" <nobody@example.net>\r\n'
+            b'Subject: I am just a test article\r\n'
+            ), ascii(data))
+        self.assertTrue(data.endswith(
+            b'This is just a test article.\r\n'
+            b'.Here is a dot-starting line.\r\n'
+            b'\r\n'
+            b'-- Signed by Andr\xc3\xa9.\r\n'
+            ), ascii(data))
+
+    def test_head(self):
+        # HEAD
+        resp, info = self.server.head()
+        self.assertEqual(resp, "221 3000237 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000237)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_head(lines)
+        # HEAD num
+        resp, info = self.server.head(3000234)
+        self.assertEqual(resp, "221 3000234 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000234)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_head(lines)
+        # HEAD id
+        resp, info = self.server.head("<45223423@example.com>")
+        self.assertEqual(resp, "221 0 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 0)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_head(lines)
+        # Non-existent id
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.head("<non-existent@example.com>")
+        self.assertEqual(cm.exception.response, "430 No Such Article Found")
+
+    def test_body(self):
+        # BODY
+        resp, info = self.server.body()
+        self.assertEqual(resp, "222 3000237 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000237)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_body(lines)
+        # BODY num
+        resp, info = self.server.body(3000234)
+        self.assertEqual(resp, "222 3000234 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 3000234)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_body(lines)
+        # BODY id
+        resp, info = self.server.body("<45223423@example.com>")
+        self.assertEqual(resp, "222 0 <45223423@example.com>")
+        art_num, message_id, lines = info
+        self.assertEqual(art_num, 0)
+        self.assertEqual(message_id, "<45223423@example.com>")
+        self._check_article_body(lines)
+        # Non-existent id
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.body("<non-existent@example.com>")
+        self.assertEqual(cm.exception.response, "430 No Such Article Found")
+
+    def check_over_xover_resp(self, resp, overviews):
+        self.assertTrue(resp.startswith("224 "), resp)
+        self.assertEqual(len(overviews), 3)
+        art_num, over = overviews[0]
+        self.assertEqual(art_num, 57)
+        self.assertEqual(over, {
+            "from": "Doug Hellmann <doug.hellmann-Re5JQEeQqe8AvxtiuMwx3w@public.gmane.org>",
+            "subject": "Re: ANN: New Plone book with strong Python (and Zope) themes throughout",
+            "date": "Sat, 19 Jun 2010 18:04:08 -0400",
+            "message-id": "<4FD05F05-F98B-44DC-8111-C6009C925F0C@gmail.com>",
+            "references": "<hvalf7$ort$1@dough.gmane.org>",
+            ":bytes": "7103",
+            ":lines": "16",
+            "xref": "news.gmane.org gmane.comp.python.authors:57"
+            })
+        art_num, over = overviews[2]
+        self.assertEqual(over["subject"],
+                         "Re: Message d'erreur incompréhensible (par moi)")
+
+    def test_xover(self):
+        resp, overviews = self.server.xover(57, 59)
+        self.check_over_xover_resp(resp, overviews)
+
+    def test_over(self):
+        # In NNTP "v1", this will fallback on XOVER
+        resp, overviews = self.server.over((57, 59))
+        self.check_over_xover_resp(resp, overviews)
+
+    sample_post = (
+        b'From: "Demo User" <nobody@example.net>\r\n'
+        b'Subject: I am just a test article\r\n'
+        b'Content-Type: text/plain; charset=UTF-8; format=flowed\r\n'
+        b'Message-ID: <i.am.an.article.you.will.want@example.com>\r\n'
+        b'\r\n'
+        b'This is just a test article.\r\n'
+        b'.Here is a dot-starting line.\r\n'
+        b'\r\n'
+        b'-- Signed by Andr\xc3\xa9.\r\n'
+    )
+
+    def _check_posted_body(self):
+        # Check the raw body as received by the server
+        lines = self.handler.posted_body
+        # One additional line for the "." terminator
+        self.assertEqual(len(lines), 10)
+        self.assertEqual(lines[-1], b'.\r\n')
+        self.assertEqual(lines[-2], b'-- Signed by Andr\xc3\xa9.\r\n')
+        self.assertEqual(lines[-3], b'\r\n')
+        self.assertEqual(lines[-4], b'..Here is a dot-starting line.\r\n')
+        self.assertEqual(lines[0], b'From: "Demo User" <nobody@example.net>\r\n')
+
+    def _check_post_ihave_sub(self, func, *args, file_factory):
+        # First the prepared post with CRLF endings
+        post = self.sample_post
+        func_args = args + (file_factory(post),)
+        self.handler.posted_body = None
+        resp = func(*func_args)
+        self._check_posted_body()
+        # Then the same post with "normal" line endings - they should be
+        # converted by NNTP.post and NNTP.ihave.
+        post = self.sample_post.replace(b"\r\n", b"\n")
+        func_args = args + (file_factory(post),)
+        self.handler.posted_body = None
+        resp = func(*func_args)
+        self._check_posted_body()
+        return resp
+
+    def check_post_ihave(self, func, success_resp, *args):
+        # With a bytes object
+        resp = self._check_post_ihave_sub(func, *args, file_factory=bytes)
+        self.assertEqual(resp, success_resp)
+        # With a bytearray object
+        resp = self._check_post_ihave_sub(func, *args, file_factory=bytearray)
+        self.assertEqual(resp, success_resp)
+        # With a file object
+        resp = self._check_post_ihave_sub(func, *args, file_factory=io.BytesIO)
+        self.assertEqual(resp, success_resp)
+        # With an iterable of terminated lines
+        def iterlines(b):
+            return iter(b.splitlines(True))
+        resp = self._check_post_ihave_sub(func, *args, file_factory=iterlines)
+        self.assertEqual(resp, success_resp)
+        # With an iterable of non-terminated lines
+        def iterlines(b):
+            return iter(b.splitlines(False))
+        resp = self._check_post_ihave_sub(func, *args, file_factory=iterlines)
+        self.assertEqual(resp, success_resp)
+
+    def test_post(self):
+        self.check_post_ihave(self.server.post, "240 Article received OK")
+        self.handler.allow_posting = False
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.post(self.sample_post)
+        self.assertEqual(cm.exception.response,
+                         "440 Posting not permitted")
+
+    def test_ihave(self):
+        self.check_post_ihave(self.server.ihave, "235 Article transferred OK",
+                              "<i.am.an.article.you.will.want@example.com>")
+        with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
+            self.server.ihave("<another.message.id>", self.sample_post)
+        self.assertEqual(cm.exception.response,
+                         "435 Article not wanted")
+
+
+class NNTPv1Tests(NNTPv1v2TestsMixin, MockedNNTPTestsMixin, unittest.TestCase):
+    """Tests an NNTP v1 server (no capabilities)."""
+
+    nntp_version = 1
+    handler_class = NNTPv1Handler
+
+    def test_caps(self):
+        caps = self.server.getcapabilities()
+        self.assertEqual(caps, {})
+        self.assertEqual(self.server.nntp_version, 1)
+
+
+class NNTPv2Tests(NNTPv1v2TestsMixin, MockedNNTPTestsMixin, unittest.TestCase):
+    """Tests an NNTP v2 server (with capabilities)."""
+
+    nntp_version = 2
+    handler_class = NNTPv2Handler
+
+    def test_caps(self):
+        caps = self.server.getcapabilities()
+        self.assertEqual(caps, {
+            'VERSION': ['2'],
+            'IMPLEMENTATION': ['INN', '2.5.1'],
+            'AUTHINFO': ['USER'],
+            'HDR': [],
+            'LIST': ['ACTIVE', 'ACTIVE.TIMES', 'DISTRIB.PATS',
+                     'HEADERS', 'NEWSGROUPS', 'OVERVIEW.FMT'],
+            'OVER': [],
+            'POST': [],
+            'READER': [],
+            })
+        self.assertEqual(self.server.nntp_version, 2)
+
+
+class MiscTests(unittest.TestCase):
+
+    def test_decode_header(self):
+        def gives(a, b):
+            self.assertEqual(nntplib.decode_header(a), b)
+        gives("" , "")
+        gives("a plain header", "a plain header")
+        gives(" with extra  spaces ", " with extra  spaces ")
+        gives("=?ISO-8859-15?Q?D=E9buter_en_Python?=", "Débuter en Python")
+        gives("=?utf-8?q?Re=3A_=5Bsqlite=5D_probl=C3=A8me_avec_ORDER_BY_sur_des_cha?="
+              " =?utf-8?q?=C3=AEnes_de_caract=C3=A8res_accentu=C3=A9es?=",
+              "Re: [sqlite] problème avec ORDER BY sur des chaînes de caractères accentuées")
+        gives("Re: =?UTF-8?B?cHJvYmzDqG1lIGRlIG1hdHJpY2U=?=",
+              "Re: problème de matrice")
+        # A natively utf-8 header (found in the real world!)
+        gives("Re: Message d'erreur incompréhensible (par moi)",
+              "Re: Message d'erreur incompréhensible (par moi)")
+
+    def test_parse_overview_fmt(self):
+        # The minimal (default) response
+        lines = ["Subject:", "From:", "Date:", "Message-ID:",
+                 "References:", ":bytes", ":lines"]
+        self.assertEqual(nntplib._parse_overview_fmt(lines),
+            ["subject", "from", "date", "message-id", "references",
+             ":bytes", ":lines"])
+        # The minimal response using alternative names
+        lines = ["Subject:", "From:", "Date:", "Message-ID:",
+                 "References:", "Bytes:", "Lines:"]
+        self.assertEqual(nntplib._parse_overview_fmt(lines),
+            ["subject", "from", "date", "message-id", "references",
+             ":bytes", ":lines"])
+        # Variations in casing
+        lines = ["subject:", "FROM:", "DaTe:", "message-ID:",
+                 "References:", "BYTES:", "Lines:"]
+        self.assertEqual(nntplib._parse_overview_fmt(lines),
+            ["subject", "from", "date", "message-id", "references",
+             ":bytes", ":lines"])
+        # First example from RFC 3977
+        lines = ["Subject:", "From:", "Date:", "Message-ID:",
+                 "References:", ":bytes", ":lines", "Xref:full",
+                 "Distribution:full"]
+        self.assertEqual(nntplib._parse_overview_fmt(lines),
+            ["subject", "from", "date", "message-id", "references",
+             ":bytes", ":lines", "xref", "distribution"])
+        # Second example from RFC 3977
+        lines = ["Subject:", "From:", "Date:", "Message-ID:",
+                 "References:", "Bytes:", "Lines:", "Xref:FULL",
+                 "Distribution:FULL"]
+        self.assertEqual(nntplib._parse_overview_fmt(lines),
+            ["subject", "from", "date", "message-id", "references",
+             ":bytes", ":lines", "xref", "distribution"])
+        # A classic response from INN
+        lines = ["Subject:", "From:", "Date:", "Message-ID:",
+                 "References:", "Bytes:", "Lines:", "Xref:full"]
+        self.assertEqual(nntplib._parse_overview_fmt(lines),
+            ["subject", "from", "date", "message-id", "references",
+             ":bytes", ":lines", "xref"])
+
+    def test_parse_overview(self):
+        fmt = nntplib._DEFAULT_OVERVIEW_FMT + ["xref"]
+        # First example from RFC 3977
+        lines = [
+            '3000234\tI am just a test article\t"Demo User" '
+            '<nobody@example.com>\t6 Oct 1998 04:38:40 -0500\t'
+            '<45223423@example.com>\t<45454@example.net>\t1234\t'
+            '17\tXref: news.example.com misc.test:3000363',
+        ]
+        overview = nntplib._parse_overview(lines, fmt)
+        (art_num, fields), = overview
+        self.assertEqual(art_num, 3000234)
+        self.assertEqual(fields, {
+            'subject': 'I am just a test article',
+            'from': '"Demo User" <nobody@example.com>',
+            'date': '6 Oct 1998 04:38:40 -0500',
+            'message-id': '<45223423@example.com>',
+            'references': '<45454@example.net>',
+            ':bytes': '1234',
+            ':lines': '17',
+            'xref': 'news.example.com misc.test:3000363',
+        })
+
+    def test_parse_datetime(self):
+        def gives(a, b, *c):
+            self.assertEqual(nntplib._parse_datetime(a, b),
+                             datetime.datetime(*c))
+        # Output of DATE command
+        gives("19990623135624", None, 1999, 6, 23, 13, 56, 24)
+        # Variations
+        gives("19990623", "135624", 1999, 6, 23, 13, 56, 24)
+        gives("990623", "135624", 1999, 6, 23, 13, 56, 24)
+        gives("090623", "135624", 2009, 6, 23, 13, 56, 24)
+
+    def test_unparse_datetime(self):
+        # Test non-legacy mode
+        # 1) with a datetime
+        def gives(y, M, d, h, m, s, date_str, time_str):
+            dt = datetime.datetime(y, M, d, h, m, s)
+            self.assertEqual(nntplib._unparse_datetime(dt),
+                             (date_str, time_str))
+            self.assertEqual(nntplib._unparse_datetime(dt, False),
+                             (date_str, time_str))
+        gives(1999, 6, 23, 13, 56, 24, "19990623", "135624")
+        gives(2000, 6, 23, 13, 56, 24, "20000623", "135624")
+        gives(2010, 6, 5, 1, 2, 3, "20100605", "010203")
+        # 2) with a date
+        def gives(y, M, d, date_str, time_str):
+            dt = datetime.date(y, M, d)
+            self.assertEqual(nntplib._unparse_datetime(dt),
+                             (date_str, time_str))
+            self.assertEqual(nntplib._unparse_datetime(dt, False),
+                             (date_str, time_str))
+        gives(1999, 6, 23, "19990623", "000000")
+        gives(2000, 6, 23, "20000623", "000000")
+        gives(2010, 6, 5, "20100605", "000000")
+
+    def test_unparse_datetime_legacy(self):
+        # Test legacy mode (RFC 977)
+        # 1) with a datetime
+        def gives(y, M, d, h, m, s, date_str, time_str):
+            dt = datetime.datetime(y, M, d, h, m, s)
+            self.assertEqual(nntplib._unparse_datetime(dt, True),
+                             (date_str, time_str))
+        gives(1999, 6, 23, 13, 56, 24, "990623", "135624")
+        gives(2000, 6, 23, 13, 56, 24, "000623", "135624")
+        gives(2010, 6, 5, 1, 2, 3, "100605", "010203")
+        # 2) with a date
+        def gives(y, M, d, date_str, time_str):
+            dt = datetime.date(y, M, d)
+            self.assertEqual(nntplib._unparse_datetime(dt, True),
+                             (date_str, time_str))
+        gives(1999, 6, 23, "990623", "000000")
+        gives(2000, 6, 23, "000623", "000000")
+        gives(2010, 6, 5, "100605", "000000")
+
+
+def test_main():
+    support.run_unittest(MiscTests, NNTPv1Tests, NNTPv2Tests,
+                         NetworkedNNTPTests
+                         )
+
+
+if __name__ == "__main__":
+    test_main()