datatracker/ietf/message/tests.py at patch-2 · codecontr/datatracker · GitHub

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
# Copyright The IETF Trust 2013-2020, All Rights Reserved
# -*- coding: utf-8 -*-
import datetime
import mock

from smtplib import SMTPException

from django.urls import reverse as urlreverse
from django.utils import timezone

import debug                            # pyflakes:ignore

from ietf.group.factories import GroupFactory
from ietf.message.factories import SendQueueFactory
from ietf.message.models import Message, SendQueue
from ietf.message.tasks import send_scheduled_mail_task
from ietf.message.utils import send_scheduled_message_from_send_queue
from ietf.person.models import Person
from ietf.utils.mail import outbox, send_mail_text, send_mail_message, get_payload_text
from ietf.utils.test_utils import TestCase
from ietf.utils.timezone import date_today


class MessageTests(TestCase):
    def test_message_view(self):
        nomcom = GroupFactory(name="nomcom%s" % date_today().year, type_id="nomcom")
        msg = Message.objects.create(
            by=Person.objects.get(name="(System)"),
            subject="This is a test",
            to="test@example.com",
            frm="nomcomchair@example.com",
            body="Hello World!",
            content_type="text/plain",
            )
        msg.related_groups.add(nomcom)

        r = self.client.get(urlreverse("ietf.message.views.message", kwargs=dict(message_id=msg.id)))
        self.assertEqual(r.status_code, 200)
        self.assertContains(r, msg.subject)
        self.assertContains(r, msg.to)
        self.assertContains(r, msg.frm)
        self.assertContains(r, "Hello World!")

    def test_capture_send_mail_text(self):
        def cmp(e1, e2):
            e1keys = set(e1.keys())
            e2keys = set(e2.keys())
            self.assertEqual(e1keys, e2keys)
            self.longMessage = True
            for k in e1keys:
                if k in ['Date', ]:
                    continue
                self.assertEqual(e1.get_all(k), e2.get_all(k), "Header field: %s" % k)
            self.longMessage = False
            self.assertEqual(get_payload_text(e1), get_payload_text(e2))

        #
        self.assertEqual(Message.objects.count(), 0)
        to = "<iesg-secretary@ietf.org>"
        subj = "Dummy subject"
        cc="cc.a@example.com, cc.b@example.com"
        body = "Dummy message text"
        bcc="bcc@example.com"
        msg1 = send_mail_text(None, to, None, subj, body, cc=cc, bcc=bcc)
        self.assertEqual(Message.objects.count(), 1)
        message = Message.objects.last()
        self.assertEqual(message.by.name, '(System)')
        self.assertEqual(message.to, to)
        self.assertEqual(message.subject, subj)
        self.assertEqual(message.body, body)
        self.assertEqual(message.cc, cc)
        self.assertEqual(message.bcc, bcc)
        self.assertEqual(message.content_type, 'text/plain')
        # Check round-trip msg --> message --> msg
        msg2 = send_mail_message(None, message)
        cmp(msg1, msg2)
        cmp(msg1, outbox[-1])


class SendScheduledAnnouncementsTests(TestCase):
    def test_send_plain_announcement(self):
        msg = Message.objects.create(
            by=Person.objects.get(name="(System)"),
            subject="This is a test",
            to="test@example.com",
            frm="testmonkey@example.com",
            cc="cc.a@example.com, cc.b@example.com",
            bcc="bcc@example.com",
            body="Hello World!",
            content_type="text/plain",
            )

        q = SendQueue.objects.create(
            by=Person.objects.get(name="(System)"),
            message=msg,
            send_at=timezone.now() + datetime.timedelta(hours=12)
            )

        mailbox_before = len(outbox)

        send_scheduled_message_from_send_queue(q)

        self.assertEqual(len(outbox), mailbox_before + 1)
        self.assertTrue("This is a test" in outbox[-1]["Subject"])
        self.assertTrue(SendQueue.objects.get(id=q.id).sent_at)

    def test_send_mime_announcement(self):
        msg = Message.objects.create(
            by=Person.objects.get(name="(System)"),
            subject="This is a test",
            to="test@example.com",
            frm="testmonkey@example.com",
            cc="cc.a@example.com, cc.b@example.com",
            bcc="bcc@example.com",
            body='--NextPart\r\n\r\nA New Internet-Draft is available from the on-line Internet-Drafts directories.\r\n--NextPart\r\nContent-Type: Message/External-body;\r\n\tname="draft-huang-behave-bih-01.txt";\r\n\tsite="ftp.ietf.org";\r\n\taccess-type="anon-ftp";\r\n\tdirectory="internet-drafts"\r\n\r\nContent-Type: text/plain\r\nContent-ID:     <2010-07-30001541.I-D@ietf.org>\r\n\r\n--NextPart--',
            content_type='Multipart/Mixed; Boundary="NextPart"',
            )

        q = SendQueue.objects.create(
            by=Person.objects.get(name="(System)"),
            message=msg,
            send_at=timezone.now() + datetime.timedelta(hours=12)
            )

        mailbox_before = len(outbox)

        send_scheduled_message_from_send_queue(q)

        self.assertEqual(len(outbox), mailbox_before + 1)
        self.assertTrue("This is a test" in outbox[-1]["Subject"])
        self.assertTrue("--NextPart" in outbox[-1].as_string())
        self.assertTrue(SendQueue.objects.get(id=q.id).sent_at)


class TaskTests(TestCase):
    @mock.patch("ietf.message.tasks.log_smtp_exception")
    @mock.patch("ietf.message.tasks.send_scheduled_message_from_send_queue")
    def test_send_scheduled_mail_task(self, mock_send_message, mock_log_smtp_exception):
        not_yet_sent = SendQueueFactory()
        SendQueueFactory(sent_at=timezone.now())  # already sent
        send_scheduled_mail_task()
        self.assertEqual(mock_send_message.call_count, 1)
        self.assertEqual(mock_send_message.call_args[0], (not_yet_sent,))
        self.assertFalse(mock_log_smtp_exception.called)

        mock_send_message.reset_mock()
        mock_send_message.side_effect = SMTPException
        send_scheduled_mail_task()
        self.assertEqual(mock_send_message.call_count, 1)
        self.assertEqual(mock_send_message.call_args[0], (not_yet_sent,))
        self.assertTrue(mock_log_smtp_exception.called)