← Back to branch summary

~unity-2d-team/unity-2d/Shell-MultiMonitor

« back to all changes in this revision

Viewing changes to grackle/tests/test_model.py

  • Committer: William Grant
  • Date: 2012-04-20 02:20:58 UTC
  • mfrom: (6.1.70 trunk)
  • Revision ID: william.grant@canonical.com-20120420022058-3nkracsmlg7akydu
Merge trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
grackle/tests/test_model.py
16
16
 
17
17
__metaclass__ = type
18
18
 
 
19
import calendar
19
20
import datetime
20
21
from email.utils import formatdate
21
22
import os
22
23
from tempfile import _RandomNameSequence
23
 
import time
24
 
import unittest
25
24
 
26
25
from dateutil.tz import (
27
26
    tzoffset,
28
27
    tzutc,
29
28
    )
30
29
from pycassa.util import convert_uuid_to_time
 
30
import testtools
31
31
 
32
32
from grackle.model import (
33
33
    CassandraConnection,
 
34
    _format_message,
34
35
    _parse_message,
35
36
    )
36
37
 
50
51
    id='<aaaaaaaaaaaaa@example.com>')
51
52
 
52
53
 
53
 
class TestParseMessage(unittest.TestCase):
 
54
class TestParseMessage(testtools.TestCase):
54
55
 
55
56
    def test_works(self):
56
57
        # _parse_message extracts interesting fields. It also parses the
67
68
            date)
68
69
 
69
70
 
70
 
class TestAddMessage(unittest.TestCase):
 
71
class TestAddMessage(testtools.TestCase):
71
72
 
72
73
    def test_add_message(self):
73
74
        c = CassandraConnection(
86
87
 
87
88
        # The key in archive_message is a TimeUUID for the Date field in
88
89
        # the message.
89
 
        utctime = time.mktime(datetime.datetime(
 
90
        utctime = calendar.timegm(datetime.datetime(
90
91
            2000, 1, 1, 0, 2, 34, tzinfo=tzutc()).timetuple())
91
92
        self.assertEqual(
92
93
            utctime,
94
95
 
95
96
        # The stored message contains the full original text of the
96
97
        # message, as well as interesting fields parsed out.
97
 
        self.assertEqual(TEST_MESSAGE, cmsg['content'])
 
98
        self.assertEqual(TEST_MESSAGE, cmsg['raw'])
98
99
        parsed_message = _parse_message(TEST_MESSAGE)[1]
99
100
        for key, value in parsed_message.iteritems():
100
101
            self.assertEqual(value, cmsg[key])
101
102
 
102
103
 
103
 
class TestGetMessages(unittest.TestCase):
 
104
class TestGetMessages(testtools.TestCase):
104
105
 
105
106
    def assertMessages(self, expected_ids, messages):
106
107
        expected_msgids = [
107
108
            '<message%d@example.com>' % id for id in expected_ids]
108
 
        actual_msgids = [msg['message-id'] for msg in messages]
 
109
        actual_msgids = [msg['headers']['message-id'] for msg in messages]
109
110
        self.assertEqual(expected_msgids, actual_msgids)
110
111
 
111
112
    def makeMessages(self, conn, archive, count):
126
127
    def test_single_message(self):
127
128
        conn, archive = self.makeArchive()
128
129
        self.makeMessages(conn, archive, 1)
129
 
        self.assertMessages(
130
 
            [0], conn.get_messages(archive, 'date', 1, '')[1])
 
130
 
 
131
        # We get a single message when we ask for it.
 
132
        messages = conn.get_messages(archive, 'date', 1, '')[1]
 
133
        self.assertMessages([0], messages)
 
134
        cmsg = messages[0]
 
135
 
 
136
        # The raw message matches what we generated, and the result is
 
137
        # formatted according to the default settings.
 
138
        expected_raw = TEMPLATE_MESSAGE.format(
 
139
            date=formatdate(0), id='<message0@example.com>')
 
140
        self.assertEqual(expected_raw, messages[0]['raw'])
 
141
        pmsg = _parse_message(expected_raw)[1]
 
142
        pmsg['raw'] = expected_raw
 
143
        self.assertEqual(
 
144
            _format_message(
 
145
                pmsg,
 
146
                headers=['date', 'from', 'subject', 'message-id'],
 
147
                include_raw=True),
 
148
            cmsg)
131
149
 
132
150
    def test_limit(self):
133
151
        conn, archive = self.makeArchive()
170
188
        self.assertIs(None, prev)
171
189
        self.assertMessages([], messages)
172
190
        self.assertIs(None, next)
 
191
 
 
192
    def test_date_filter(self):
 
193
        conn, archive = self.makeArchive()
 
194
        self.makeMessages(conn, archive, 10)
 
195
        start = datetime.datetime.utcfromtimestamp(250).replace(
 
196
            tzinfo=tzutc())
 
197
        finish = datetime.datetime.utcfromtimestamp(500).replace(
 
198
            tzinfo=tzutc())
 
199
        prev, messages, next = conn.get_messages(
 
200
            archive, 'date', 2, '', start_date=start, finish_date=finish)
 
201
        self.assertMessages([3, 4], messages)
 
202
        prev, messages, next = conn.get_messages(
 
203
            archive, 'date', 2, next, start_date=start, finish_date=finish)
 
204
        self.assertMessages([5], messages)
 
205
 
 
206
 
 
207
class TestMessageFormatter(testtools.TestCase):
 
208
 
 
209
    def test_all(self):
 
210
        parsed = _parse_message(TEST_MESSAGE)[1]
 
211
        parsed['raw'] = TEST_MESSAGE
 
212
        formatted = _format_message(
 
213
            parsed,
 
214
            headers=['date', 'from', 'subject', 'message-id'],
 
215
            include_raw=True)
 
216
        expected = {
 
217
            'headers': {
 
218
                'date': '2000-01-01T11:02:34+11:00',
 
219
                'from': 'sysadmin@example.com',
 
220
                'message-id': '<aaaaaaaaaaaaa@example.com>',
 
221
                'subject': 'Everything is broken',
 
222
                },
 
223
            'raw': TEST_MESSAGE,
 
224
            }
 
225
        self.assertEqual(expected, formatted)