OLD | NEW |
| (Empty) |
1 # Copyright (c) 2001-2004 Twisted Matrix Laboratories. | |
2 # See LICENSE for details. | |
3 | |
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 | |
9 ALL_GROUPS = ('alt.test.nntp', 0, 1, 'y'), | |
10 GROUP = ('0', '1', '0', 'alt.test.nntp', 'group', 'selected') | |
11 SUBSCRIPTIONS = ['alt.test.nntp', 'news.testgroup'] | |
12 | |
13 POST_STRING = """Path: not-for-mail | |
14 From: <exarkun@somehost.domain.com> | |
15 Subject: a test | |
16 Newsgroups: alt.test.nntp | |
17 Organization: | |
18 Summary: | |
19 Keywords: | |
20 User-Agent: tin/1.4.5-20010409 ("One More Nightmare") (UNIX) (Linux/2.4.17 (i686
)) | |
21 | |
22 this is a test | |
23 . | |
24 .. | |
25 ... | |
26 lala | |
27 moo | |
28 -- | |
29 "One World, one Web, one Program." - Microsoft(R) promotional ad | |
30 "Ein Volk, ein Reich, ein Fuhrer." - Adolf Hitler | |
31 -- | |
32 10:56pm up 4 days, 4:42, 1 user, load average: 0.08, 0.08, 0.12 | |
33 """ | |
34 | |
35 class TestNNTPClient(nntp.NNTPClient): | |
36 def __init__(self): | |
37 nntp.NNTPClient.__init__(self) | |
38 | |
39 def assertEquals(self, foo, bar): | |
40 if foo != bar: raise AssertionError("%r != %r!" % (foo, bar)) | |
41 | |
42 def connectionMade(self): | |
43 nntp.NNTPClient.connectionMade(self) | |
44 self.fetchSubscriptions() | |
45 | |
46 | |
47 def gotSubscriptions(self, subscriptions): | |
48 self.assertEquals(len(subscriptions), len(SUBSCRIPTIONS)) | |
49 for s in subscriptions: | |
50 assert s in SUBSCRIPTIONS | |
51 | |
52 self.fetchGroups() | |
53 | |
54 def gotAllGroups(self, info): | |
55 self.assertEquals(len(info), len(ALL_GROUPS)) | |
56 self.assertEquals(info[0], ALL_GROUPS[0]) | |
57 | |
58 self.fetchGroup('alt.test.nntp') | |
59 | |
60 | |
61 def getAllGroupsFailed(self, error): | |
62 raise AssertionError("fetchGroups() failed: %s" % (error,)) | |
63 | |
64 | |
65 def gotGroup(self, info): | |
66 self.assertEquals(len(info), 6) | |
67 self.assertEquals(info, GROUP) | |
68 | |
69 self.postArticle(POST_STRING) | |
70 | |
71 | |
72 def getSubscriptionsFailed(self, error): | |
73 raise AssertionError("fetchSubscriptions() failed: %s" % (error,)) | |
74 | |
75 | |
76 def getGroupFailed(self, error): | |
77 raise AssertionError("fetchGroup() failed: %s" % (error,)) | |
78 | |
79 | |
80 def postFailed(self, error): | |
81 raise AssertionError("postArticle() failed: %s" % (error,)) | |
82 | |
83 | |
84 def postedOk(self): | |
85 self.fetchArticle(1) | |
86 | |
87 | |
88 def gotArticle(self, info): | |
89 origBody = POST_STRING.split('\n\n')[1] | |
90 newBody = info.split('\n\n', 1)[1] | |
91 | |
92 self.assertEquals(origBody, newBody) | |
93 | |
94 # We're done | |
95 self.transport.loseConnection() | |
96 | |
97 | |
98 def getArticleFailed(self, error): | |
99 raise AssertionError("fetchArticle() failed: %s" % (error,)) | |
100 | |
101 | |
102 class NNTPTestCase(unittest.TestCase): | |
103 def setUp(self): | |
104 self.server = nntp.NNTPServer() | |
105 self.server.factory = self | |
106 self.backend = database.NewsShelf(None, 'news.db') | |
107 self.backend.addGroup('alt.test.nntp', 'y') | |
108 | |
109 for s in SUBSCRIPTIONS: | |
110 self.backend.addSubscription(s) | |
111 | |
112 self.client = TestNNTPClient() | |
113 | |
114 def testLoopback(self): | |
115 return loopback.loopbackAsync(self.server, self.client) | |
116 | |
117 # XXX This test is woefully incomplete. It tests the single | |
118 # most common code path and nothing else. Expand it and the | |
119 # test fairy will leave you a surprise. | |
120 | |
121 # reactor.iterate(1) # fetchGroups() | |
122 # reactor.iterate(1) # fetchGroup() | |
123 # reactor.iterate(1) # postArticle() | |
124 | |
OLD | NEW |