~didrocks/unity/altf10

« back to all changes in this revision

Viewing changes to grackle/tests/test_client.py

  • Committer: Curtis Hovey
  • Date: 2012-03-16 16:10:54 UTC
  • Revision ID: curtis.hovey@canonical.com-20120316161054-jk68i83kh81qm2vi
Store the body text only.

Show diffs side-by-side

added added

removed removed

Lines of Context:
32
32
def make_message(message_id, body='body', headers=None, hidden=False):
33
33
    if headers is None:
34
34
        headers = {}
35
 
    message_headers = {
36
 
        'Message-Id': message_id,
37
 
        'date': '2005-01-01',
38
 
        'subject': 'subject',
39
 
        'from': 'author',
40
 
        'replies': '',
 
35
    headers['Message-Id'] = message_id
 
36
    message = {
 
37
        'message_id': message_id,
 
38
        'headers': headers,
 
39
        'thread_id': message_id,
 
40
        'date': headers.get('date', '2005-01-01'),
 
41
        'subject': headers.get('subject', 'subject'),
 
42
        'author': headers.get('author', 'author'),
 
43
        'hidden': hidden,
 
44
        'attachments': [],
 
45
        'replies': headers.get('in-reply-to', None),
 
46
        'body': body,
41
47
        }
42
 
    message_headers.update(headers.items())
43
 
    message = Message()
44
 
    message.set_payload(body)
45
 
    for key, value in message_headers.items():
46
 
        message[key] = value
47
 
    return make_json_message(message_id, message.as_string(), hidden)
 
48
    return message
48
49
 
49
50
 
50
51
def make_mime_message(message_id, body='body', headers=None, hidden=False,
51
52
                      attachment_type=None):
 
53
    if headers is None:
 
54
        headers = {}
52
55
    parts = MIMEMultipart()
53
56
    parts.attach(MIMEText(body))
54
57
    if attachment_type is not None:
57
60
        attachment['Content-Type'] = attachment_type
58
61
        attachment['Content-Disposition'] = 'attachment; filename="file.ext"'
59
62
        parts.attach(attachment)
60
 
    return make_message(message_id, parts.as_string(), headers, hidden)
 
63
    message = Message()
 
64
    message.set_payload(parts.as_string())
 
65
    for key, value in headers.items():
 
66
        message[key] = value
 
67
    return make_json_message(message_id, message.as_string())
61
68
 
62
69
 
63
70
class ForkedFakeService:
135
142
 
136
143
    def do_POST(self):
137
144
        """Create a message on POST."""
 
145
        message = self.rfile.read(int(self.headers['content-length']))
138
146
        scheme, netloc, path, params, query_string, fragments = (
139
147
            urlparse(self.path))
140
148
        parts = path.split('/')
141
 
        if parts[1] != 'archive':
142
 
            # This is an unknonwn operation?
143
 
            return
144
 
        if 'content-length' in self.headers:
145
 
            operation = 'put_message'
146
 
        else:
147
 
            operation = 'hide_message'
148
 
        if operation == 'put_message':
149
 
            message = self.rfile.read(int(self.headers['content-length']))
 
149
        if parts[1] == 'archive' and len(parts) == 4:
150
150
            try:
151
151
                # This expected path is /archive/archive_id/message_id.
152
152
                self.server.store.put_message(parts[2], parts[3], message)
155
155
                self.wfile.close()
156
156
            except:
157
157
                self.send_error(httplib.BAD_REQUEST)
158
 
        elif operation == 'hide_message':
159
 
            try:
160
 
                # This expected path is /archive/archive_id/message_id.
161
 
                response = self.server.store.hide_message(
162
 
                    parts[2], parts[3], query_string)
163
 
                self.send_response(httplib.OK)
164
 
                self.end_headers()
165
 
                self.wfile.write(simplejson.dumps(response))
166
 
            except:
167
 
                self.send_error(httplib.BAD_REQUEST)
168
158
 
169
159
    def do_GET(self):
170
160
        """Retrieve a list of messages on GET."""
254
244
 
255
245
    def get_messages_member_order_test(self, key):
256
246
        client = GrackleClient('localhost', 8439)
257
 
        if key == 'author':
258
 
            header_name = 'from'
259
 
        else:
260
 
            header_name = key
261
247
        archive = {
262
248
            'baz': [
263
 
                make_message('foo', headers={header_name: '2011-03-25'}),
264
 
                make_message('bar', headers={header_name: '2011-03-24'}),
 
249
                make_message('foo', headers={key: '2011-03-25'}),
 
250
                make_message('bar', headers={key: '2011-03-24'}),
265
251
             ]}
266
252
        with ForkedFakeService.from_client(client, archive):
267
253
            response = client.get_messages('baz')
414
400
        first_message = response['messages'][0]
415
401
        self.assertEqual('foo', first_message['message_id'])
416
402
        self.assertEqual(
417
 
            archive['baz'][0]['headers'], first_message['headers'])
 
403
            {'From': 'me', 'Message-Id': 'foo', 'To': 'you'},
 
404
            first_message['headers'])
418
405
        self.assertNotIn('body', first_message)
419
406
 
420
407
    def test_display_type_text_only(self):
490
477
        with ForkedFakeService.from_client(client, archive):
491
478
            with ExpectedException(UnparsableDateRange, ''):
492
479
                client.get_messages('baz', date_range='2012-01..12-02..12-03')
493
 
 
494
 
 
495
 
class TestHideMessages(TestCase):
496
 
 
497
 
    def test_hide_message_true(self):
498
 
        client = GrackleClient('localhost', 8470)
499
 
        archive = {
500
 
            'baz': [
501
 
                make_message('foo', hidden=False),
502
 
            ]}
503
 
        with ForkedFakeService.from_client(client, archive):
504
 
            response = client.hide_message('baz', 'foo', hidden=True)
505
 
        self.assertEqual('foo', response['message_id'])
506
 
        self.assertIs(True, response['hidden'])
507
 
 
508
 
    def test_hide_message_false(self):
509
 
        client = GrackleClient('localhost', 8470)
510
 
        archive = {
511
 
            'baz': [
512
 
                make_message('foo', hidden=True),
513
 
            ]}
514
 
        with ForkedFakeService.from_client(client, archive):
515
 
            response = client.hide_message('baz', 'foo', hidden=False)
516
 
        self.assertEqual('foo', response['message_id'])
517
 
        self.assertIs(False, response['hidden'])