forked from bugy/script-server
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathalerts_service_test.py
More file actions
168 lines (123 loc) · 5.7 KB
/
Copy pathalerts_service_test.py
File metadata and controls
168 lines (123 loc) · 5.7 KB
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
import importlib
import json
import sys
import unittest
from collections import OrderedDict
from unittest import mock
from communications import alerts_service
from communications.alerts_service import AlertsService
from communications.communication_model import File
from communications.communicaton_service import CommunicationsService
from communications.destination_base import Destination
from tests.communications.communication_test_utils import MockDestination, mock_communicators
class TestCommunicationService(unittest.TestCase):
def test_send_single_destination(self):
destination = self.create_destination()
service = self.create_service(destination)
service.send('title', 'body')
service._wait()
self.assertEqual(1, len(destination.messages))
self.validate_message(destination.messages[0], 'title', 'body')
def test_no_destinations(self):
service = self.create_service()
service.send('title', 'body')
service._wait()
def test_3_sequential_alerts(self):
destination = self.create_destination()
service = self.create_service(destination)
service.send('title 1', 'body 1')
service.send('title 2', 'body 2')
service.send('title 3', 'body 3')
service._wait()
self.assertEqual(3, len(destination.messages))
self.validate_message(destination.messages[0], 'title 1', 'body 1')
self.validate_message(destination.messages[1], 'title 2', 'body 2')
self.validate_message(destination.messages[2], 'title 3', 'body 3')
def test_2_destinations(self):
destination1 = self.create_destination()
destination2 = self.create_destination()
service = self.create_service(destination1, destination2)
service.send('title', 'body')
service._wait()
self.assertEqual(1, len(destination1.messages))
self.assertEqual(1, len(destination2.messages))
def test_2_destinations_when_one_failing(self):
destination1 = self.create_failing_destination()
destination2 = self.create_destination()
service = self.create_service(destination1, destination2)
service.send('title', 'body')
service._wait()
self.assertEqual(1, len(destination2.messages))
def create_destination(self):
return MockDestination('mockDestination')
@staticmethod
def create_failing_destination():
class FailingDestination(Destination):
def send(self, title, body, files=None):
raise Exception('Send failed')
return FailingDestination()
def create_service(self, *destinations):
return CommunicationsService(destinations)
def validate_message(self, message_tuple, title, body, files=None):
message_title, message_body, message_logs = message_tuple
self.assertEqual(title, message_title)
self.assertEqual(body, message_body)
self.assertEqual(files, message_logs)
@mock_communicators
class TestAlertsService(unittest.TestCase):
def test_create_single_http_destination(self):
config = self.create_config(['http'])
AlertsService(config)
self.assert_created_destinations(['http1'])
def test_create_single_email_destination(self):
config = self.create_config(['email'])
AlertsService(config)
self.assert_created_destinations(['email1'])
def test_create_for_missing_config(self):
AlertsService(None)
self.assert_created_destinations([])
def test_create_mixed_destinations(self):
config = self.create_config(['email', 'http', 'http', 'email'])
AlertsService(config)
self.assert_created_destinations(['email1', 'http1', 'http2', 'email2'])
def test_create_unknown_destination(self):
config = self.create_config(['socket'])
self.assertRaisesRegex(Exception, 'Unknown alert destination type: socket', AlertsService, config)
self.assert_created_destinations([])
def test_send_email_alert(self):
config = self.create_config(['email'])
alerts_service = AlertsService(config)
title = 'My test alert'
body = 'Test message body'
files = [File(filename='log.txt', content='doing X')]
self.send_alert(alerts_service, body, files, title)
self.assertEqual([(title, body, files)], self.get_communicators()[0].messages)
def test_send_http_alert(self):
config = self.create_config(['http'])
alerts_service = AlertsService(config)
title = 'My test alert'
body = 'Test message body'
files = [File(filename='log.txt', content='doing X')]
self.send_alert(alerts_service, body, files, title)
expected_body = json.dumps(OrderedDict([
('title', title),
('message', body),
('files', {
'log.txt': 'doing X'
})]))
self.assertEqual([(None, expected_body, None)], self.get_communicators()[0].messages)
def test_import_alerts_service_with_missing_dependencies(self):
with mock.patch.dict(sys.modules, {'requests': None}):
with mock.patch.dict(sys.modules, {'smtplib': None}):
importlib.reload(alerts_service)
def send_alert(self, alerts_service, body, files, title):
alerts_service.send_alert(title, body, files=files)
alerts_service._wait()
def create_config(self, types):
destinations = [{'type': t} for t in types]
config = {'destinations': destinations}
return config
def assert_created_destinations(self, expected_names):
communicators = self.get_communicators()
actual_names = [d.name for d in communicators]
self.assertEqual(expected_names, actual_names)