123 lines
5.7 KiB
Python
123 lines
5.7 KiB
Python
from units.compat import unittest
|
|
from ansible.modules.monitoring import pagerduty
|
|
|
|
import json
|
|
|
|
|
|
class PagerDutyTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.pd = pagerduty.PagerDutyRequest(module=pagerduty, name='name', user='user', token='token')
|
|
|
|
def _assert_ongoing_maintenance_windows(self, module, url, headers):
|
|
self.assertEqual('https://api.pagerduty.com/maintenance_windows?filter=ongoing', url)
|
|
return object(), {'status': 200}
|
|
|
|
def _assert_ongoing_window_with_v1_compatible_header(self, module, url, headers, data=None, method=None):
|
|
self.assertDictContainsSubset(
|
|
{'Accept': 'application/vnd.pagerduty+json;version=2'},
|
|
headers,
|
|
'Accept:application/vnd.pagerduty+json;version=2 HTTP header not found'
|
|
)
|
|
return object(), {'status': 200}
|
|
|
|
def _assert_create_a_maintenance_window_url(self, module, url, headers, data=None, method=None):
|
|
self.assertEqual('https://api.pagerduty.com/maintenance_windows', url)
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_create_a_maintenance_window_http_method(self, module, url, headers, data=None, method=None):
|
|
self.assertEqual('POST', method)
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_create_a_maintenance_window_from_header(self, module, url, headers, data=None, method=None):
|
|
self.assertDictContainsSubset(
|
|
{'From': 'requester_id'},
|
|
headers,
|
|
'From:requester_id HTTP header not found'
|
|
)
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_create_window_with_v1_compatible_header(self, module, url, headers, data=None, method=None):
|
|
self.assertDictContainsSubset(
|
|
{'Accept': 'application/vnd.pagerduty+json;version=2'},
|
|
headers,
|
|
'Accept:application/vnd.pagerduty+json;version=2 HTTP header not found'
|
|
)
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_create_window_payload(self, module, url, headers, data=None, method=None):
|
|
payload = json.loads(data)
|
|
window_data = payload['maintenance_window']
|
|
self.assertTrue('start_time' in window_data, '"start_time" is requiered attribute')
|
|
self.assertTrue('end_time' in window_data, '"end_time" is requiered attribute')
|
|
self.assertTrue('services' in window_data, '"services" is requiered attribute')
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_create_window_single_service(self, module, url, headers, data=None, method=None):
|
|
payload = json.loads(data)
|
|
window_data = payload['maintenance_window']
|
|
services = window_data['services']
|
|
self.assertEqual(
|
|
[{'id': 'service_id', 'type': 'service_reference'}],
|
|
services
|
|
)
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_create_window_multiple_service(self, module, url, headers, data=None, method=None):
|
|
payload = json.loads(data)
|
|
window_data = payload['maintenance_window']
|
|
services = window_data['services']
|
|
print(services)
|
|
self.assertEqual(
|
|
[
|
|
{'id': 'service_id_1', 'type': 'service_reference'},
|
|
{'id': 'service_id_2', 'type': 'service_reference'},
|
|
{'id': 'service_id_3', 'type': 'service_reference'},
|
|
],
|
|
services
|
|
)
|
|
return object(), {'status': 201}
|
|
|
|
def _assert_absent_maintenance_window_url(self, module, url, headers, method=None):
|
|
self.assertEqual('https://api.pagerduty.com/maintenance_windows/window_id', url)
|
|
return object(), {'status': 204}
|
|
|
|
def _assert_absent_window_with_v1_compatible_header(self, module, url, headers, method=None):
|
|
self.assertDictContainsSubset(
|
|
{'Accept': 'application/vnd.pagerduty+json;version=2'},
|
|
headers,
|
|
'Accept:application/vnd.pagerduty+json;version=2 HTTP header not found'
|
|
)
|
|
return object(), {'status': 204}
|
|
|
|
def test_ongoing_maintenance_windos_url(self):
|
|
self.pd.ongoing(http_call=self._assert_ongoing_maintenance_windows)
|
|
|
|
def test_ongoing_maintenance_windos_compatibility_header(self):
|
|
self.pd.ongoing(http_call=self._assert_ongoing_window_with_v1_compatible_header)
|
|
|
|
def test_create_maintenance_window_url(self):
|
|
self.pd.create('requester_id', 'service', 1, 0, 'desc', http_call=self._assert_create_a_maintenance_window_url)
|
|
|
|
def test_create_maintenance_window_http_method(self):
|
|
self.pd.create('requester_id', 'service', 1, 0, 'desc', http_call=self._assert_create_a_maintenance_window_http_method)
|
|
|
|
def test_create_maintenance_from_header(self):
|
|
self.pd.create('requester_id', 'service', 1, 0, 'desc', http_call=self._assert_create_a_maintenance_window_from_header)
|
|
|
|
def test_create_maintenance_compatibility_header(self):
|
|
self.pd.create('requester_id', 'service', 1, 0, 'desc', http_call=self._assert_create_window_with_v1_compatible_header)
|
|
|
|
def test_create_maintenance_request_payload(self):
|
|
self.pd.create('requester_id', 'service', 1, 0, 'desc', http_call=self._assert_create_window_payload)
|
|
|
|
def test_create_maintenance_for_single_service(self):
|
|
self.pd.create('requester_id', 'service_id', 1, 0, 'desc', http_call=self._assert_create_window_single_service)
|
|
|
|
def test_create_maintenance_for_multiple_services(self):
|
|
self.pd.create('requester_id', ['service_id_1', 'service_id_2', 'service_id_3'], 1, 0, 'desc', http_call=self._assert_create_window_multiple_service)
|
|
|
|
def test_absent_maintenance_window_url(self):
|
|
self.pd.absent('window_id', http_call=self._assert_absent_maintenance_window_url)
|
|
|
|
def test_absent_maintenance_compatibility_header(self):
|
|
self.pd.absent('window_id', http_call=self._assert_absent_window_with_v1_compatible_header)
|