/usr/lib/python2.7/dist-packages/twisted/news/test/test_database.py is in python-twisted-news 13.2.0-1ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 | # Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for L{twisted.news.database}.
"""
__metaclass__ = type
from email.Parser import Parser
from socket import gethostname
from twisted.trial.unittest import TestCase
from twisted.internet.defer import succeed
from twisted.mail.smtp import messageid
from twisted.news.database import Article, PickleStorage, NewsShelf
class ModerationTestsMixin:
"""
Tests for the moderation features of L{INewsStorage} implementations.
"""
def setUp(self):
self._email = []
def sendmail(self, smtphost, from_addr, to_addrs, msg,
senderDomainName=None, port=25):
"""
Fake of L{twisted.mail.smtp.sendmail} which records attempts to send
email and immediately pretends success.
Subclasses should arrange for their storage implementation to call this
instead of the real C{sendmail} function.
"""
self._email.append((
smtphost, from_addr, to_addrs, msg, senderDomainName, port))
return succeed(None)
_messageTemplate = """\
From: some dude
To: another person
Subject: activities etc
Message-ID: %(articleID)s
Newsgroups: %(newsgroup)s
%(approved)s
Body of the message is such.
""".replace('\n', '\r\n')
def getApprovedMessage(self, articleID, group):
"""
Return a C{str} containing an RFC 2822 formatted message including an
I{Approved} header indicating it has passed through moderation.
"""
return self._messageTemplate % {
'articleID': articleID,
'newsgroup': group,
'approved': 'Approved: yup\r\n'}
def getUnapprovedMessage(self, articleID, group):
"""
Return a C{str} containing an RFC 2822 formatted message with no
I{Approved} header indicating it may require moderation.
"""
return self._messageTemplate % {
'articleID': articleID,
'newsgroup': group,
'approved': '\r\n'}
def getStorage(self, groups, moderators, mailhost, sender):
"""
Override in a subclass to return a L{INewsStorage} provider to test for
correct moderation behavior.
@param groups: A C{list} of C{str} naming the groups which should exist
in the resulting storage object.
@param moderators: A C{dict} mapping C{str} each group name to a C{list}
of C{str} giving moderator email (RFC 2821) addresses.
"""
raise NotImplementedError()
def test_postApproved(self):
"""
L{INewsStorage.postRequest} posts the message if it includes an
I{Approved} header.
"""
group = "example.group"
moderator = "alice@example.com"
mailhost = "127.0.0.1"
sender = "bob@example.org"
articleID = messageid()
storage = self.getStorage(
[group], {group: [moderator]}, mailhost, sender)
message = self.getApprovedMessage(articleID, group)
result = storage.postRequest(message)
def cbPosted(ignored):
self.assertEqual(self._email, [])
exists = storage.articleExistsRequest(articleID)
exists.addCallback(self.assertTrue)
return exists
result.addCallback(cbPosted)
return result
def test_postModerated(self):
"""
L{INewsStorage.postRequest} forwards a message to the moderator if it
does not include an I{Approved} header.
"""
group = "example.group"
moderator = "alice@example.com"
mailhost = "127.0.0.1"
sender = "bob@example.org"
articleID = messageid()
storage = self.getStorage(
[group], {group: [moderator]}, mailhost, sender)
message = self.getUnapprovedMessage(articleID, group)
result = storage.postRequest(message)
def cbModerated(ignored):
self.assertEqual(len(self._email), 1)
self.assertEqual(self._email[0][0], mailhost)
self.assertEqual(self._email[0][1], sender)
self.assertEqual(self._email[0][2], [moderator])
self._checkModeratorMessage(
self._email[0][3], sender, moderator, group, message)
self.assertEqual(self._email[0][4], None)
self.assertEqual(self._email[0][5], 25)
exists = storage.articleExistsRequest(articleID)
exists.addCallback(self.assertFalse)
return exists
result.addCallback(cbModerated)
return result
def _checkModeratorMessage(self, messageText, sender, moderator, group, postingText):
p = Parser()
msg = p.parsestr(messageText)
headers = dict(msg.items())
del headers['Message-ID']
self.assertEqual(
headers,
{'From': sender,
'To': moderator,
'Subject': 'Moderate new %s message: activities etc' % (group,),
'Content-Type': 'message/rfc822'})
posting = p.parsestr(postingText)
attachment = msg.get_payload()[0]
for header in ['from', 'to', 'subject', 'message-id', 'newsgroups']:
self.assertEqual(posting[header], attachment[header])
self.assertEqual(posting.get_payload(), attachment.get_payload())
class PickleStorageTests(ModerationTestsMixin, TestCase):
"""
Tests for L{PickleStorage}.
"""
def getStorage(self, groups, moderators, mailhost, sender):
"""
Create and return a L{PickleStorage} instance configured to require
moderation.
"""
storageFilename = self.mktemp()
storage = PickleStorage(
storageFilename, groups, moderators, mailhost, sender)
storage.sendmail = self.sendmail
self.addCleanup(PickleStorage.sharedDBs.pop, storageFilename)
return storage
class NewsShelfTests(ModerationTestsMixin, TestCase):
"""
Tests for L{NewsShelf}.
"""
def getStorage(self, groups, moderators, mailhost, sender):
"""
Create and return a L{NewsShelf} instance configured to require
moderation.
"""
storageFilename = self.mktemp()
shelf = NewsShelf(mailhost, storageFilename, sender)
for name in groups:
shelf.addGroup(name, 'm') # Dial 'm' for moderator
for address in moderators.get(name, []):
shelf.addModerator(name, address)
shelf.sendmail = self.sendmail
return shelf
def test_notifyModerator(self):
"""
L{NewsShelf.notifyModerator} sends a moderation email to a single
moderator.
"""
shelf = NewsShelf('example.com', self.mktemp(), 'alice@example.com')
shelf.sendmail = self.sendmail
shelf.notifyModerator('bob@example.org', Article('Foo: bar', 'Some text'))
self.assertEqual(len(self._email), 1)
def test_defaultSender(self):
"""
If no sender is specified to L{NewsShelf.notifyModerators}, a default
address based on the system hostname is used for both the envelope and
RFC 2822 sender addresses.
"""
shelf = NewsShelf('example.com', self.mktemp())
shelf.sendmail = self.sendmail
shelf.notifyModerators(['bob@example.org'], Article('Foo: bar', 'Some text'))
self.assertEqual(self._email[0][1], 'twisted-news@' + gethostname())
self.assertIn('From: twisted-news@' + gethostname(), self._email[0][3])
|