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
|
# -*- coding: utf-8 -*-
# pylint: disable=R0904,C0103
# - Too many public methods
# - Invalid name XXX (should match YYY)
import unittest
import datetime
from urllib2 import HTTPError
from mock import Mock
from mailman.email.message import Message
from mailman.interfaces.archiver import ArchivePolicy
from kittystore import get_store
from kittystore.caching import mailman_user
from kittystore.test import get_test_file, FakeList, SettingsModule
class ListCacheTestCase(unittest.TestCase):
def setUp(self):
self.store = get_store(SettingsModule(), auto_create=True)
def tearDown(self):
self.store.close()
def test_properties_on_new_message(self):
ml = FakeList("example-list")
ml.display_name = u"name 1"
ml.subject_prefix = u"[prefix 1]"
ml.description = u"desc 1"
msg = Message()
msg["From"] = "dummy@example.com"
msg["Message-ID"] = "<dummy>"
msg.set_payload("Dummy message")
self.store.add_to_list(ml, msg)
ml_db = self.store.get_lists()[0]
self.assertEqual(ml_db.display_name, "name 1")
self.assertEqual(ml_db.subject_prefix, "[prefix 1]")
ml.display_name = u"name 2"
ml.subject_prefix = u"[prefix 2]"
ml.description = u"desc 2"
ml.archive_policy = ArchivePolicy.private
msg.replace_header("Message-ID", "<dummy2>")
self.store.add_to_list(ml, msg)
ml_db = self.store.get_lists()[0]
#ml_db = self.store.db.find(List).one()
self.assertEqual(ml_db.display_name, "name 2")
self.assertEqual(ml_db.subject_prefix, "[prefix 2]")
self.assertEqual(ml_db.description, "desc 2")
self.assertEqual(ml_db.archive_policy, ArchivePolicy.private)
def test_on_old_message(self):
olddate = datetime.datetime.utcnow() - datetime.timedelta(days=40)
ml = FakeList("example-list")
msg = Message()
msg["From"] = "dummy@example.com"
msg["Message-ID"] = "<dummy>"
msg["Date"] = olddate.isoformat()
msg.set_payload("Dummy message")
self.store.add_to_list(ml, msg)
ml_db = self.store.get_lists()[0]
self.assertEqual(ml_db.recent_participants_count, 0)
self.assertEqual(ml_db.recent_threads_count, 0)
class FakeMMUser(object):
user_id = None
class UserIdCacheTestCase(unittest.TestCase):
def setUp(self):
self.store = get_store(SettingsModule(), auto_create=True)
self.mm_client = Mock()
mailman_user._MAILMAN_CLIENT = self.mm_client
self.mm_client.get_user.side_effect = HTTPError(
None, 404, "dummy", {}, None)
def tearDown(self):
self.store.close()
mailman_user._MAILMAN_CLIENT = None
def test_on_new_message_userid(self):
# Check that the user_id is set on a new message
msg = Message()
msg["From"] = "dummy@example.com"
msg["Message-ID"] = "<dummy>"
msg.set_payload("Dummy message")
# setup Mailman's reply
new_user_id = FakeMMUser()
new_user_id.user_id = "DUMMY-USER-ID"
self.mm_client.get_user.side_effect = lambda addr: new_user_id
# check the User does not exist yet
self.assertEqual(0,
self.store.get_message_count_by_user_id("DUMMY-USER-ID"))
# do the test and check
self.store.add_to_list(FakeList("example-list"), msg)
dbmsg = self.store.get_message_by_id_from_list(
"example-list", "dummy")
self.assertEqual(dbmsg.sender.user_id, "DUMMY-USER-ID")
self.assertTrue(dbmsg.sender.user is not None,
"A 'User' instance was not created")
self.assertEqual(dbmsg.sender.user.id, "DUMMY-USER-ID")
self.assertEqual(1,
self.store.get_message_count_by_user_id("DUMMY-USER-ID"))
# XXX: Storm-specific
from kittystore.storm.model import User
self.assertEqual(self.store.db.find(User).count(), 1)
def test_on_new_message_no_reply_from_mailman(self):
# Check that the user_id is set on a new message
msg = Message()
msg["From"] = "dummy@example.com"
msg["Message-ID"] = "<dummy>"
msg.set_payload("Dummy message")
self.store.add_to_list(FakeList("example-list"), msg)
dbmsg = self.store.get_message_by_id_from_list(
"example-list", "dummy")
self.assertEqual(dbmsg.sender.user_id, None)
def test_sync_mailman_user(self):
# Check that the user_id is set when sync_mailman_user is run
msg = Message()
msg["From"] = "dummy@example.com"
msg["Message-ID"] = "<dummy>"
msg.set_payload("Dummy message")
self.store.add_to_list(FakeList("example-list"), msg)
dbmsg = self.store.get_message_by_id_from_list(
"example-list", "dummy")
self.assertEqual(dbmsg.sender.user_id, None)
# setup Mailman's reply
new_user_id = FakeMMUser()
new_user_id.user_id = "DUMMY-USER-ID"
self.mm_client.get_user.side_effect = lambda addr: new_user_id
# do the test and check
mailman_user.sync_mailman_user(self.store)
#dbmsg = self.store.get_message_by_id_from_list(
# "example-list", "dummy")
self.assertEqual(dbmsg.sender.user_id, "DUMMY-USER-ID")
self.assertTrue(dbmsg.sender.user is not None,
"A 'User' instance was not created")
self.assertEqual(dbmsg.sender.user.id, "DUMMY-USER-ID")
self.assertEqual(1,
self.store.get_message_count_by_user_id("DUMMY-USER-ID"))
|
