1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
4 from twisted.trial import unittest
5 from twisted.news import database
6 from twisted.news import nntp
7 from twisted.protocols import loopback
8 from twisted.test import proto_helpers
10 ALL_GROUPS = ('alt.test.nntp', 0, 1, 'y'),
11 GROUP = ('0', '1', '0', 'alt.test.nntp', 'group', 'selected')
12 SUBSCRIPTIONS = ['alt.test.nntp', 'news.testgroup']
14 POST_STRING = """Path: not-for-mail
15 From: <exarkun@somehost.domain.com>
17 Newsgroups: alt.test.nntp
21 User-Agent: tin/1.4.5-20010409 ("One More Nightmare") (UNIX) (Linux/2.4.17 (i686))
30 "One World, one Web, one Program." - Microsoft(R) promotional ad
31 "Ein Volk, ein Reich, ein Fuhrer." - Adolf Hitler
33 10:56pm up 4 days, 4:42, 1 user, load average: 0.08, 0.08, 0.12
36 class TestNNTPClient(nntp.NNTPClient):
38 nntp.NNTPClient.__init__(self)
40 def assertEqual(self, foo, bar):
41 if foo != bar: raise AssertionError("%r != %r!" % (foo, bar))
43 def connectionMade(self):
44 nntp.NNTPClient.connectionMade(self)
45 self.fetchSubscriptions()
48 def gotSubscriptions(self, subscriptions):
49 self.assertEqual(len(subscriptions), len(SUBSCRIPTIONS))
50 for s in subscriptions:
51 assert s in SUBSCRIPTIONS
55 def gotAllGroups(self, info):
56 self.assertEqual(len(info), len(ALL_GROUPS))
57 self.assertEqual(info[0], ALL_GROUPS[0])
59 self.fetchGroup('alt.test.nntp')
62 def getAllGroupsFailed(self, error):
63 raise AssertionError("fetchGroups() failed: %s" % (error,))
66 def gotGroup(self, info):
67 self.assertEqual(len(info), 6)
68 self.assertEqual(info, GROUP)
70 self.postArticle(POST_STRING)
73 def getSubscriptionsFailed(self, error):
74 raise AssertionError("fetchSubscriptions() failed: %s" % (error,))
77 def getGroupFailed(self, error):
78 raise AssertionError("fetchGroup() failed: %s" % (error,))
81 def postFailed(self, error):
82 raise AssertionError("postArticle() failed: %s" % (error,))
89 def gotArticle(self, info):
90 origBody = POST_STRING.split('\n\n')[1]
91 newBody = info.split('\n\n', 1)[1]
93 self.assertEqual(origBody, newBody)
96 self.transport.loseConnection()
99 def getArticleFailed(self, error):
100 raise AssertionError("fetchArticle() failed: %s" % (error,))
103 class NNTPTestCase(unittest.TestCase):
105 self.server = nntp.NNTPServer()
106 self.server.factory = self
107 self.backend = database.NewsShelf(None, 'news.db')
108 self.backend.addGroup('alt.test.nntp', 'y')
110 for s in SUBSCRIPTIONS:
111 self.backend.addSubscription(s)
113 self.transport = proto_helpers.StringTransport()
114 self.server.makeConnection(self.transport)
115 self.client = TestNNTPClient()
117 def testLoopback(self):
118 return loopback.loopbackAsync(self.server, self.client)
120 # XXX This test is woefully incomplete. It tests the single
121 # most common code path and nothing else. Expand it and the
122 # test fairy will leave you a surprise.
124 # reactor.iterate(1) # fetchGroups()
125 # reactor.iterate(1) # fetchGroup()
126 # reactor.iterate(1) # postArticle()
129 def test_connectionMade(self):
131 When L{NNTPServer} is connected, it sends a server greeting to the
135 self.transport.value().split('\r\n'), [
136 '200 server ready - posting allowed',
142 When L{NTTPServer} receives a I{LIST} command, it sends a list of news
143 groups to the client (RFC 3977, section 7.6.1.1).
145 self.transport.clear()
146 self.server.do_LIST()
148 self.transport.value().split('\r\n'), [
149 '215 newsgroups in form "group high low flags"',
150 'alt.test.nntp 0 1 y',
155 def test_GROUP(self):
157 When L{NNTPServer} receives a I{GROUP} command, it sends a line of
158 information about that group to the client (RFC 3977, section 6.1.1.1).
160 self.transport.clear()
161 self.server.do_GROUP('alt.test.nntp')
163 self.transport.value().split('\r\n'), [
164 '211 0 1 0 alt.test.nntp group selected',
168 def test_LISTGROUP(self):
170 When L{NNTPServer} receives a I{LISTGROUP} command, it sends a list of
171 message numbers for the messages in a particular group (RFC 3977,
174 self.transport.clear()
175 self.server.do_LISTGROUP('alt.test.nntp')
177 self.transport.value().split('\r\n'), [
178 '211 list of article numbers follow',
183 def test_XROVER(self):
185 When L{NTTPServer} receives a I{XROVER} command, it sends a list of
186 I{References} header values for the messages in a particular group (RFC
189 self.server.do_GROUP('alt.test.nntp')
190 self.transport.clear()
192 self.server.do_XROVER()
194 self.transport.value().split('\r\n'), [
195 '221 Header follows',