Lines Matching refs:self
42 def test_welcome(self):
43 welcome = self.server.getwelcome()
44 self.assertEqual(str, type(welcome))
46 def test_help(self):
47 resp, lines = self.server.help()
48 self.assertTrue(resp.startswith("100 "), resp)
50 self.assertEqual(str, type(line))
52 def test_list(self):
53 resp, groups = self.server.list()
55 self.assertEqual(GroupInfo, type(groups[0]))
56 self.assertEqual(str, type(groups[0].group))
58 def test_list_active(self):
59 resp, groups = self.server.list(self.GROUP_PAT)
61 self.assertEqual(GroupInfo, type(groups[0]))
62 self.assertEqual(str, type(groups[0].group))
64 def test_unknown_command(self):
65 with self.assertRaises(nntplib.NNTPPermanentError) as cm:
66 self.server._shortcmd("XYZZY")
68 self.assertTrue(resp.startswith("500 "), resp)
70 def test_newgroups(self):
74 resp, groups = self.server.newgroups(dt)
76 self.assertIsInstance(groups[0], GroupInfo)
77 self.assertIsInstance(groups[0].group, str)
79 def test_description(self):
82 self.assertIsInstance(desc, str)
83 self.assertNotIn(self.GROUP_NAME, desc)
84 desc = self.server.description(self.GROUP_NAME)
87 self.assertIn(self.DESC, desc)
89 desc = self.server.description(self.GROUP_PAT)
92 desc = self.server.description("zk.brrtt.baz")
93 self.assertEqual(desc, '')
95 def test_descriptions(self):
96 resp, descs = self.server.descriptions(self.GROUP_PAT)
98 self.assertTrue(
100 self.assertIsInstance(descs, dict)
101 desc = descs[self.GROUP_NAME]
102 self.assertEqual(desc, self.server.description(self.GROUP_NAME))
104 def test_group(self):
105 result = self.server.group(self.GROUP_NAME)
106 self.assertEqual(5, len(result))
108 self.assertEqual(group, self.GROUP_NAME)
109 self.assertIsInstance(count, int)
110 self.assertIsInstance(first, int)
111 self.assertIsInstance(last, int)
112 self.assertLessEqual(first, last)
113 self.assertTrue(resp.startswith("211 "), resp)
115 def test_date(self):
116 resp, date = self.server.date()
117 self.assertIsInstance(date, datetime.datetime)
119 self.assertGreaterEqual(date.year, 1995)
120 self.assertLessEqual(date.year, 2030)
122 def _check_art_dict(self, art_dict):
124 self.assertIsInstance(art_dict, dict)
126 self.assertGreaterEqual(art_dict.keys(),
131 self.assertIsInstance(v, (str, type(None)))
133 def test_xover(self):
134 resp, count, first, last, name = self.server.group(self.GROUP_NAME)
135 resp, lines = self.server.xover(last - 5, last)
137 self.skipTest("no articles retrieved")
140 self.assertGreaterEqual(art_num, last - 5)
141 self.assertLessEqual(art_num, last)
142 self._check_art_dict(art_dict)
146 def test_over(self):
147 resp, count, first, last, name = self.server.group(self.GROUP_NAME)
150 resp, lines = self.server.over((start, None))
152 self._check_art_dict(art_dict)
154 resp, lines = self.server.over((start, last))
157 self.assertGreaterEqual(art_num, start)
158 self.assertLessEqual(art_num, last)
159 self._check_art_dict(art_dict)
163 def test_xhdr(self):
164 resp, count, first, last, name = self.server.group(self.GROUP_NAME)
165 resp, lines = self.server.xhdr('subject', last)
167 self.assertEqual(str, type(line[1]))
169 def check_article_resp(self, resp, article, art_num=None):
170 self.assertIsInstance(article, nntplib.ArticleInfo)
172 self.assertEqual(article.number, art_num)
174 self.assertIsInstance(line, bytes)
176 self.assertNotIn(article.lines[-1], (b".", b".\n", b".\r\n"))
179 def test_article_head_body(self):
180 resp, count, first, last, name = self.server.group(self.GROUP_NAME)
184 resp, head = self.server.head(art_num)
192 self.skipTest("could not find a suitable article number")
193 self.assertTrue(resp.startswith("221 "), resp)
194 self.check_article_resp(resp, head, art_num)
195 resp, body = self.server.body(art_num)
196 self.assertTrue(resp.startswith("222 "), resp)
197 self.check_article_resp(resp, body, art_num)
198 resp, article = self.server.article(art_num)
199 self.assertTrue(resp.startswith("220 "), resp)
200 self.check_article_resp(resp, article, art_num)
207 self.assertEqual(filtered_lines, filtered_head_lines + [b''] + body.lines)
209 def test_capabilities(self):
215 self.assertIsInstance(caps_list, (list, tuple))
216 self.assertIn('OVERVIEW.FMT', caps_list)
217 self.assertGreaterEqual(self.server.nntp_version, 2)
218 _check_caps(self.server.getcapabilities())
220 resp, caps = self.server.capabilities()
223 def test_zlogin(self):
229 self.assertRaises(nntplib.NNTPError, self.server.login,
236 def test_zzquit(self):
238 cls = type(self)
240 self.server.quit()
250 def wrapped(self):
251 with socket_helper.transient_internet(self.NNTP_HOST):
252 meth(self)
264 def test_timeout(self):
265 with self.assertRaises(ValueError):
266 self.NNTP_CLASS(self.NNTP_HOST, timeout=0, usenetrc=False)
268 def test_with_statement(self):
282 if self.ssl_context is not None:
283 kwargs["ssl_context"] = self.ssl_context
286 server = self.NNTP_CLASS(self.NNTP_HOST, **kwargs)
288 self.assertTrue(is_connected())
289 self.assertTrue(server.help())
290 self.assertFalse(is_connected())
292 server = self.NNTP_CLASS(self.NNTP_HOST, **kwargs)
295 self.assertFalse(is_connected())
300 f"to {self.NNTP_HOST!r}")
385 def __init__(self, handler):
386 io.RawIOBase.__init__(self)
388 self.c2s = io.BytesIO()
390 self.s2c = io.BytesIO()
391 self.handler = handler
392 self.handler.start(self.c2s.readline, self.push_data)
394 def readable(self):
397 def writable(self):
400 def push_data(self, data):
402 pos = self.s2c.tell()
403 self.s2c.seek(0, 2)
404 self.s2c.write(data)
405 self.s2c.seek(pos)
407 def write(self, b):
409 pos = self.c2s.tell()
410 self.c2s.write(b)
411 self.c2s.seek(pos)
412 self.handler.process_pending()
415 def readinto(self, buf):
417 self.handler.process_pending()
418 b = self.s2c.read(len(buf))
434 def __init__(self, f, host, readermode=None):
435 self.file = f
436 self.host = host
437 self._base_init(readermode)
439 def _close(self):
440 self.file.close()
441 del self.file
448 def setUp(self):
450 self.make_server()
452 def tearDown(self):
454 del self.server
456 def make_server(self, *args, **kwargs):
457 self.handler = self.handler_class()
458 self.sio, file = make_mock_file(self.handler)
459 self.server = NNTPServer(file, 'test.server', *args, **kwargs)
460 return self.server
464 def setUp(self):
466 self.make_server(readermode=True)
474 def start(self, readline, push_data):
475 self.in_body = False
476 self.allow_posting = True
477 self._readline = readline
478 self._push_data = push_data
479 self._logged_in = False
480 self._user_sent = False
482 self.handle_welcome()
484 def _decode(self, data):
487 def process_pending(self):
488 if self.in_body:
490 line = self._readline()
493 self.body.append(line)
497 meth, tokens = self.body_callback
498 meth(*tokens, body=self.body)
500 self.body_callback = None
501 self.body = None
502 self.in_body = False
504 line = self._decode(self._readline())
511 #meth = getattr(self.handler, "handle_" + cmd.upper(), None)
512 meth = getattr(self, "handle_" + cmd.upper(), None)
514 self.handle_unknown()
521 if self.in_body:
522 self.body_callback = meth, tokens
523 self.body = []
525 def expect_body(self):
527 self.in_body = True
529 def push_data(self, data):
531 self._push_data(data)
533 def push_lit(self, lit):
538 self.push_data(lit)
540 def handle_unknown(self):
541 self.push_lit("500 What?")
543 def handle_welcome(self):
544 self.push_lit(self.welcome)
546 def handle_QUIT(self):
547 self.push_lit("205 Bye!")
549 def handle_DATE(self):
550 self.push_lit("111 20100914001155")
552 def handle_GROUP(self, group):
554 self.push_lit("211 486 761 1265 fr.comp.lang.python")
556 self.push_lit("411 No such group {}".format(group))
558 def handle_HELP(self):
559 self.push_lit("""\
567 def handle_STAT(self, message_spec=None):
569 self.push_lit("412 No newsgroup selected")
571 self.push_lit("223 3000234 <45223423@example.com>")
573 self.push_lit("223 0 <45223423@example.com>")
575 self.push_lit("430 No Such Article Found")
577 def handle_NEXT(self):
578 self.push_lit("223 3000237 <668929@example.org> retrieved")
580 def handle_LAST(self):
581 self.push_lit("223 3000234 <45223423@example.com> retrieved")
583 def handle_LIST(self, action=None, param=None):
585 self.push_lit("""\
596 self.push_lit("""\
602 self.push_lit("""\
606 self.push_lit("""\
620 self.push_lit("""\
625 self.push_lit("""\
631 self.push_lit("""\
635 self.push_lit('501 Unknown LIST keyword')
637 def handle_NEWNEWS(self, group, date_str, time_str):
643 self.push_lit("""\
651 self.push_lit("""\
659 self.push_lit('too long line' * 3000 +
662 self.push_lit("""\
668 def handle_XOVER(self, message_spec):
670 self.push_lit(
696 self.push_lit("""\
700 def handle_POST(self, *, body=None):
702 if self.allow_posting:
703 self.push_lit("340 Input article; end with <CR-LF>.<CR-LF>")
704 self.expect_body()
706 self.push_lit("440 Posting not permitted")
708 assert self.allow_posting
709 self.push_lit("240 Article received OK")
710 self.posted_body = body
712 def handle_IHAVE(self, message_id, *, body=None):
714 if (self.allow_posting and
716 self.push_lit("335 Send it; end with <CR-LF>.<CR-LF>")
717 self.expect_body()
719 self.push_lit("435 Article not wanted")
721 assert self.allow_posting
722 self.push_lit("235 Article transferred OK")
723 self.posted_body = body
739 def handle_ARTICLE(self, message_spec=None):
741 self.push_lit("220 3000237 <45223423@example.com>")
743 self.push_lit("220 0 <45223423@example.com>")
745 self.push_lit("220 3000234 <45223423@example.com>")
747 self.push_lit("430 No Such Article Found")
749 self.push_lit(self.sample_article)
750 self.push_lit(".")
752 def handle_HEAD(self, message_spec=None):
754 self.push_lit("221 3000237 <45223423@example.com>")
756 self.push_lit("221 0 <45223423@example.com>")
758 self.push_lit("221 3000234 <45223423@example.com>")
760 self.push_lit("430 No Such Article Found")
762 self.push_lit(self.sample_head)
763 self.push_lit(".")
765 def handle_BODY(self, message_spec=None):
767 self.push_lit("222 3000237 <45223423@example.com>")
769 self.push_lit("222 0 <45223423@example.com>")
771 self.push_lit("222 3000234 <45223423@example.com>")
773 self.push_lit("430 No Such Article Found")
775 self.push_lit(self.sample_body)
776 self.push_lit(".")
778 def handle_AUTHINFO(self, cred_type, data):
779 if self._logged_in:
780 self.push_lit('502 Already Logged In')
782 if self._user_sent:
783 self.push_lit('482 User Credential Already Sent')
785 self.push_lit('381 Password Required')
786 self._user_sent = True
788 self.push_lit('281 Login Successful')
789 self._logged_in = True
797 def handle_CAPABILITIES(self):
809 if not self._logged_in:
810 self.push_lit(fmt.format('\n AUTHINFO USER'))
812 self.push_lit(fmt.format(''))
814 def handle_MODE(self, _):
817 def handle_OVER(self, message_spec=None):
818 return self.handle_XOVER(message_spec)
824 def handle_CAPABILITIES(self):
825 if not self._logged_in:
826 self.push_lit('480 You must log in.')
834 def __init__(self):
835 self._switched = False
837 def handle_CAPABILITIES(self):
848 if self._switched:
849 self.push_lit(fmt.format(''))
851 self.push_lit(fmt.format('MODE-'))
853 def handle_MODE(self, what):
854 assert not self._switched and what == 'reader'
855 self._switched = True
856 self.push_lit('200 Posting allowed')
861 def setUp(self):
864 def test_welcome(self):
865 self.assertEqual(self.server.welcome, self.handler.welcome)
867 def test_authinfo(self):
868 if self.nntp_version == 2:
869 self.assertIn('AUTHINFO', self.server._caps)
870 self.server.login('testuser', 'testpw')
873 self.assertNotIn('AUTHINFO', self.server._caps)
875 def test_date(self):
876 resp, date = self.server.date()
877 self.assertEqual(resp, "111 20100914001155")
878 self.assertEqual(date, datetime.datetime(2010, 9, 14, 0, 11, 55))
880 def test_quit(self):
881 self.assertFalse(self.sio.closed)
882 resp = self.server.quit()
883 self.assertEqual(resp, "205 Bye!")
884 self.assertTrue(self.sio.closed)
886 def test_help(self):
887 resp, help = self.server.help()
888 self.assertEqual(resp, "100 Legal commands")
889 self.assertEqual(help, [
896 def test_list(self):
897 resp, groups = self.server.list()
898 self.assertEqual(len(groups), 6)
900 self.assertEqual(g,
903 resp, groups = self.server.list("*distutils*")
904 self.assertEqual(len(groups), 2)
906 self.assertEqual(g,
910 def test_stat(self):
911 resp, art_num, message_id = self.server.stat(3000234)
912 self.assertEqual(resp, "223 3000234 <45223423@example.com>")
913 self.assertEqual(art_num, 3000234)
914 self.assertEqual(message_id, "<45223423@example.com>")
915 resp, art_num, message_id = self.server.stat("<45223423@example.com>")
916 self.assertEqual(resp, "223 0 <45223423@example.com>")
917 self.assertEqual(art_num, 0)
918 self.assertEqual(message_id, "<45223423@example.com>")
919 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
920 self.server.stat("<non.existent.id>")
921 self.assertEqual(cm.exception.response, "430 No Such Article Found")
922 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
923 self.server.stat()
924 self.assertEqual(cm.exception.response, "412 No newsgroup selected")
926 def test_next(self):
927 resp, art_num, message_id = self.server.next()
928 self.assertEqual(resp, "223 3000237 <668929@example.org> retrieved")
929 self.assertEqual(art_num, 3000237)
930 self.assertEqual(message_id, "<668929@example.org>")
932 def test_last(self):
933 resp, art_num, message_id = self.server.last()
934 self.assertEqual(resp, "223 3000234 <45223423@example.com> retrieved")
935 self.assertEqual(art_num, 3000234)
936 self.assertEqual(message_id, "<45223423@example.com>")
938 def test_description(self):
939 desc = self.server.description("comp.lang.python")
940 self.assertEqual(desc, "The Python computer language.")
941 desc = self.server.description("comp.lang.pythonx")
942 self.assertEqual(desc, "")
944 def test_descriptions(self):
945 resp, groups = self.server.descriptions("comp.lang.python")
946 self.assertEqual(resp, '215 Descriptions in form "group description".')
947 self.assertEqual(groups, {
950 resp, groups = self.server.descriptions("comp.lang.python*")
951 self.assertEqual(groups, {
955 resp, groups = self.server.descriptions("comp.lang.pythonx")
956 self.assertEqual(groups, {})
958 def test_group(self):
959 resp, count, first, last, group = self.server.group("fr.comp.lang.python")
960 self.assertTrue(resp.startswith("211 "), resp)
961 self.assertEqual(first, 761)
962 self.assertEqual(last, 1265)
963 self.assertEqual(count, 486)
964 self.assertEqual(group, "fr.comp.lang.python")
965 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
966 self.server.group("comp.lang.python.devel")
968 self.assertTrue(exc.response.startswith("411 No such group"),
971 def test_newnews(self):
974 resp, ids = self.server.newnews("comp.lang.python", dt)
978 ).format(self.nntp_version)
979 self.assertEqual(resp, expected)
980 self.assertEqual(ids, [
986 resp, ids = self.server.newnews("fr.comp.lang.python", dt)
987 self.assertEqual(resp, "230 An empty list of newsarticles follows")
988 self.assertEqual(ids, [])
990 def _check_article_body(self, lines):
991 self.assertEqual(len(lines), 4)
992 self.assertEqual(lines[-1].decode('utf-8'), "-- Signed by André.")
993 self.assertEqual(lines[-2], b"")
994 self.assertEqual(lines[-3], b".Here is a dot-starting line.")
995 self.assertEqual(lines[-4], b"This is just a test article.")
997 def _check_article_head(self, lines):
998 self.assertEqual(len(lines), 4)
999 self.assertEqual(lines[0], b'From: "Demo User" <nobody@example.net>')
1000 self.assertEqual(lines[3], b"Message-ID: <i.am.an.article.you.will.want@example.com>")
1002 def _check_article_data(self, lines):
1003 self.assertEqual(len(lines), 9)
1004 self._check_article_head(lines[:4])
1005 self._check_article_body(lines[-4:])
1006 self.assertEqual(lines[4], b"")
1008 def test_article(self):
1010 resp, info = self.server.article()
1011 self.assertEqual(resp, "220 3000237 <45223423@example.com>")
1013 self.assertEqual(art_num, 3000237)
1014 self.assertEqual(message_id, "<45223423@example.com>")
1015 self._check_article_data(lines)
1017 resp, info = self.server.article(3000234)
1018 self.assertEqual(resp, "220 3000234 <45223423@example.com>")
1020 self.assertEqual(art_num, 3000234)
1021 self.assertEqual(message_id, "<45223423@example.com>")
1022 self._check_article_data(lines)
1024 resp, info = self.server.article("<45223423@example.com>")
1025 self.assertEqual(resp, "220 0 <45223423@example.com>")
1027 self.assertEqual(art_num, 0)
1028 self.assertEqual(message_id, "<45223423@example.com>")
1029 self._check_article_data(lines)
1031 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1032 self.server.article("<non-existent@example.com>")
1033 self.assertEqual(cm.exception.response, "430 No Such Article Found")
1035 def test_article_file(self):
1038 resp, info = self.server.article(file=f)
1039 self.assertEqual(resp, "220 3000237 <45223423@example.com>")
1041 self.assertEqual(art_num, 3000237)
1042 self.assertEqual(message_id, "<45223423@example.com>")
1043 self.assertEqual(lines, [])
1045 self.assertTrue(data.startswith(
1049 self.assertTrue(data.endswith(
1056 def test_head(self):
1058 resp, info = self.server.head()
1059 self.assertEqual(resp, "221 3000237 <45223423@example.com>")
1061 self.assertEqual(art_num, 3000237)
1062 self.assertEqual(message_id, "<45223423@example.com>")
1063 self._check_article_head(lines)
1065 resp, info = self.server.head(3000234)
1066 self.assertEqual(resp, "221 3000234 <45223423@example.com>")
1068 self.assertEqual(art_num, 3000234)
1069 self.assertEqual(message_id, "<45223423@example.com>")
1070 self._check_article_head(lines)
1072 resp, info = self.server.head("<45223423@example.com>")
1073 self.assertEqual(resp, "221 0 <45223423@example.com>")
1075 self.assertEqual(art_num, 0)
1076 self.assertEqual(message_id, "<45223423@example.com>")
1077 self._check_article_head(lines)
1079 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1080 self.server.head("<non-existent@example.com>")
1081 self.assertEqual(cm.exception.response, "430 No Such Article Found")
1083 def test_head_file(self):
1085 resp, info = self.server.head(file=f)
1086 self.assertEqual(resp, "221 3000237 <45223423@example.com>")
1088 self.assertEqual(art_num, 3000237)
1089 self.assertEqual(message_id, "<45223423@example.com>")
1090 self.assertEqual(lines, [])
1092 self.assertTrue(data.startswith(
1096 self.assertFalse(data.endswith(
1103 def test_body(self):
1105 resp, info = self.server.body()
1106 self.assertEqual(resp, "222 3000237 <45223423@example.com>")
1108 self.assertEqual(art_num, 3000237)
1109 self.assertEqual(message_id, "<45223423@example.com>")
1110 self._check_article_body(lines)
1112 resp, info = self.server.body(3000234)
1113 self.assertEqual(resp, "222 3000234 <45223423@example.com>")
1115 self.assertEqual(art_num, 3000234)
1116 self.assertEqual(message_id, "<45223423@example.com>")
1117 self._check_article_body(lines)
1119 resp, info = self.server.body("<45223423@example.com>")
1120 self.assertEqual(resp, "222 0 <45223423@example.com>")
1122 self.assertEqual(art_num, 0)
1123 self.assertEqual(message_id, "<45223423@example.com>")
1124 self._check_article_body(lines)
1126 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1127 self.server.body("<non-existent@example.com>")
1128 self.assertEqual(cm.exception.response, "430 No Such Article Found")
1130 def test_body_file(self):
1132 resp, info = self.server.body(file=f)
1133 self.assertEqual(resp, "222 3000237 <45223423@example.com>")
1135 self.assertEqual(art_num, 3000237)
1136 self.assertEqual(message_id, "<45223423@example.com>")
1137 self.assertEqual(lines, [])
1139 self.assertFalse(data.startswith(
1143 self.assertTrue(data.endswith(
1150 def check_over_xover_resp(self, resp, overviews):
1151 self.assertTrue(resp.startswith("224 "), resp)
1152 self.assertEqual(len(overviews), 3)
1154 self.assertEqual(art_num, 57)
1155 self.assertEqual(over, {
1166 self.assertEqual(over["xref"], None)
1168 self.assertEqual(over["subject"],
1171 def test_xover(self):
1172 resp, overviews = self.server.xover(57, 59)
1173 self.check_over_xover_resp(resp, overviews)
1175 def test_over(self):
1177 resp, overviews = self.server.over((57, 59))
1178 self.check_over_xover_resp(resp, overviews)
1192 def _check_posted_body(self):
1194 lines = self.handler.posted_body
1196 self.assertEqual(len(lines), 10)
1197 self.assertEqual(lines[-1], b'.\r\n')
1198 self.assertEqual(lines[-2], b'-- Signed by Andr\xc3\xa9.\r\n')
1199 self.assertEqual(lines[-3], b'\r\n')
1200 self.assertEqual(lines[-4], b'..Here is a dot-starting line.\r\n')
1201 self.assertEqual(lines[0], b'From: "Demo User" <nobody@example.net>\r\n')
1203 def _check_post_ihave_sub(self, func, *args, file_factory):
1205 post = self.sample_post
1207 self.handler.posted_body = None
1209 self._check_posted_body()
1212 post = self.sample_post.replace(b"\r\n", b"\n")
1214 self.handler.posted_body = None
1216 self._check_posted_body()
1219 def check_post_ihave(self, func, success_resp, *args):
1221 resp = self._check_post_ihave_sub(func, *args, file_factory=bytes)
1222 self.assertEqual(resp, success_resp)
1224 resp = self._check_post_ihave_sub(func, *args, file_factory=bytearray)
1225 self.assertEqual(resp, success_resp)
1227 resp = self._check_post_ihave_sub(func, *args, file_factory=io.BytesIO)
1228 self.assertEqual(resp, success_resp)
1232 resp = self._check_post_ihave_sub(func, *args, file_factory=iterlines)
1233 self.assertEqual(resp, success_resp)
1237 resp = self._check_post_ihave_sub(func, *args, file_factory=iterlines)
1238 self.assertEqual(resp, success_resp)
1240 def test_post(self):
1241 self.check_post_ihave(self.server.post, "240 Article received OK")
1242 self.handler.allow_posting = False
1243 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1244 self.server.post(self.sample_post)
1245 self.assertEqual(cm.exception.response,
1248 def test_ihave(self):
1249 self.check_post_ihave(self.server.ihave, "235 Article transferred OK",
1251 with self.assertRaises(nntplib.NNTPTemporaryError) as cm:
1252 self.server.ihave("<another.message.id>", self.sample_post)
1253 self.assertEqual(cm.exception.response,
1256 def test_too_long_lines(self):
1258 self.assertRaises(nntplib.NNTPDataError,
1259 self.server.newnews, "comp.lang.python", dt)
1268 def test_caps(self):
1269 caps = self.server.getcapabilities()
1270 self.assertEqual(caps, {})
1271 self.assertEqual(self.server.nntp_version, 1)
1272 self.assertEqual(self.server.nntp_implementation, None)
1281 def test_caps(self):
1282 caps = self.server.getcapabilities()
1283 self.assertEqual(caps, {
1294 self.assertEqual(self.server.nntp_version, 3)
1295 self.assertEqual(self.server.nntp_implementation, 'INN 2.5.1')
1304 def test_caps_only_after_login(self):
1305 self.assertEqual(self.server._caps, {})
1306 self.server.login('testuser', 'testpw')
1307 self.assertIn('VERSION', self.server._caps)
1318 def test_we_are_in_reader_mode_after_connect(self):
1319 self.assertIn('READER', self.server._caps)
1324 def test_decode_header(self):
1326 self.assertEqual(nntplib.decode_header(a), b)
1340 def test_parse_overview_fmt(self):
1344 self.assertEqual(nntplib._parse_overview_fmt(lines),
1350 self.assertEqual(nntplib._parse_overview_fmt(lines),
1356 self.assertEqual(nntplib._parse_overview_fmt(lines),
1363 self.assertEqual(nntplib._parse_overview_fmt(lines),
1370 self.assertEqual(nntplib._parse_overview_fmt(lines),
1376 self.assertEqual(nntplib._parse_overview_fmt(lines),
1380 def test_parse_overview(self):
1391 self.assertEqual(art_num, 3000234)
1392 self.assertEqual(fields, {
1412 self.assertEqual(fields['xref'], None)
1423 self.assertEqual(fields['references'], ' ')
1424 self.assertEqual(fields['xref'], '')
1426 def test_parse_datetime(self):
1428 self.assertEqual(nntplib._parse_datetime(a, b),
1437 def test_unparse_datetime(self):
1442 self.assertEqual(nntplib._unparse_datetime(dt),
1444 self.assertEqual(nntplib._unparse_datetime(dt, False),
1452 self.assertEqual(nntplib._unparse_datetime(dt),
1454 self.assertEqual(nntplib._unparse_datetime(dt, False),
1460 def test_unparse_datetime_legacy(self):
1465 self.assertEqual(nntplib._unparse_datetime(dt, True),
1473 self.assertEqual(nntplib._unparse_datetime(dt, True),
1480 def test_ssl_support(self):
1481 self.assertTrue(hasattr(nntplib, 'NNTP_SSL'))
1487 def test_module_all_attribute(self):
1488 self.assertTrue(hasattr(nntplib, '__all__'))
1494 self.assertEqual(set(nntplib.__all__), set(target_api))
1504 self, handler_class,
1513 def close(self):
1526 self.assertRaisesRegex(expected_error_type, expected_error_msg):
1527 self.nntp_class('dummy', user=login, password=password)
1528 self.assertTrue(socket_closed)
1530 self.assertTrue(f.closed)
1532 def test_bad_welcome(self):
1536 self.check_constructor_error_conditions(
1539 def test_service_temporarily_unavailable(self):
1543 self.check_constructor_error_conditions(
1546 def test_service_permanently_unavailable(self):
1550 self.check_constructor_error_conditions(
1553 def test_bad_capabilities(self):
1556 def handle_CAPABILITIES(self):
1557 self.push_lit(capabilities_response)
1559 self.check_constructor_error_conditions(
1562 def test_login_aborted(self):
1567 def handle_AUTHINFO(self, *args):
1568 self.push_lit(authinfo_response)
1570 self.check_constructor_error_conditions(
1587 def setUp(self):
1591 self.background = threading.Thread(
1592 target=self.run_server, args=(sock,))
1593 self.background.start()
1594 self.addCleanup(self.background.join)
1596 self.nntp = self.enterContext(NNTP(socket_helper.HOST, port, usenetrc=False))
1598 def run_server(self, sock):
1631 def test_starttls(self):
1632 file = self.nntp.file
1633 sock = self.nntp.sock
1634 self.nntp.starttls()
1637 self.assertNotEqual(file, self.nntp.file)
1638 self.assertNotEqual(sock, self.nntp.sock)
1640 self.assertIsInstance(self.nntp.sock, ssl.SSLSocket)
1642 self.assertRaises(ValueError, self.nntp.starttls)