diff --git a/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_alarm_manager.py b/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_alarm_manager.py index 13bf1153a..e30863667 100644 --- a/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_alarm_manager.py +++ b/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_alarm_manager.py @@ -1,4 +1,4 @@ -# Copyright (c) 2020-2022 Wind River Systems, Inc. +# Copyright (c) 2020-2024 Wind River Systems, Inc. # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at @@ -12,60 +12,67 @@ # under the License. # -from six.moves import http_client - from dcmanager.db.sqlalchemy import api as db_api - -from dcmanager.tests.unit.api import test_root_controller as testroot +from dcmanager.tests.unit.api.test_root_controller import DCManagerApiTest from dcmanager.tests import utils -FAKE_URL = '/v1.0/alarms' -FAKE_TENANT = utils.UUID1 -FAKE_ID = '1' -FAKE_HEADERS = {'X-Tenant-Id': FAKE_TENANT, 'X_ROLE': 'admin,member,reader', - 'X-Identity-Status': 'Confirmed', 'X-Project-Name': 'admin'} +class BaseTestSubcloudAlarmController(DCManagerApiTest): + """Base class for testing the SubcloudAlarmController""" -class TestSubcloudAlarmController(testroot.DCManagerApiTest): def setUp(self): - super(TestSubcloudAlarmController, self).setUp() - self.ctx = utils.dummy_context() + super().setUp() - def test_get_alarms(self): - get_url = FAKE_URL - subcloud_summary = [{'region_name': 'subcloud1', - 'uuid': utils.UUID2, - 'critical_alarms': 1, - 'major_alarms': 2, - 'minor_alarms': 3, - 'warnings': 0, - 'cloud_status': 'critical'}, - {'region_name': 'subcloud2', - 'uuid': utils.UUID3, - 'critical_alarms': 0, - 'major_alarms': 2, - 'minor_alarms': 3, - 'warnings': 4, - 'cloud_status': 'degraded'}] + self.url = '/v1.0/alarms' - db_api.subcloud_alarms_create(self.ctx, - 'subcloud2', - values={'uuid': utils.UUID3, - 'critical_alarms': 0, - 'major_alarms': 2, - 'minor_alarms': 3, - 'warnings': 4, - 'cloud_status': 'degraded'}) - db_api.subcloud_alarms_create(self.ctx, - 'subcloud1', - values={'uuid': utils.UUID2, - 'critical_alarms': 1, - 'major_alarms': 2, - 'minor_alarms': 3, - 'warnings': 0, - 'cloud_status': 'critical'}) - response = self.app.get(get_url, headers=FAKE_HEADERS) - self.assertEqual(response.content_type, 'application/json') - self.assertEqual(response.status_code, http_client.OK) +class TestSubcloudAlarmController(BaseTestSubcloudAlarmController): + """Test class for SubcloudAlarmController""" + + def setUp(self): + super().setUp() + + def test_unmapped_method(self): + """Test requesting an unmapped method results in success with null content""" + + self.method = self.app.put + + response = self._send_request() + + self._assert_response(response) + self.assertEqual(response.text, 'null') + + +class TestSubcloudAlarmControllerGet(BaseTestSubcloudAlarmController): + """Test class for get requests""" + + def setUp(self): + super().setUp() + + self.method = self.app.get + + def test_get_succeeds(self): + """Test get succeeds""" + + subcloud1_values = { + 'uuid': utils.UUID2, 'critical_alarms': 1, 'major_alarms': 2, + 'minor_alarms': 3, 'warnings': 0, 'cloud_status': 'critical' + } + + subcloud2_values = { + 'uuid': utils.UUID3, 'critical_alarms': 0, 'major_alarms': 2, + 'minor_alarms': 3, 'warnings': 4, 'cloud_status': 'degraded' + } + + subcloud_summary = [ + {'region_name': 'subcloud1', **subcloud1_values}, + {'region_name': 'subcloud2', **subcloud2_values} + ] + + db_api.subcloud_alarms_create(self.ctx, 'subcloud2', values=subcloud2_values) + db_api.subcloud_alarms_create(self.ctx, 'subcloud1', values=subcloud1_values) + + response = self._send_request() + + self._assert_response(response) self.assertEqual(subcloud_summary, response.json.get('alarm_summary')) diff --git a/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_notifications.py b/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_notifications.py index ec8a0b19c..4ade54de3 100644 --- a/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_notifications.py +++ b/distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_notifications.py @@ -1,4 +1,4 @@ -# Copyright (c) 2021-2022 Wind River Systems, Inc. +# Copyright (c) 2021-2024 Wind River Systems, Inc. # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at @@ -12,43 +12,93 @@ # under the License. # +import http.client import json -import mock -from six.moves import http_client -from dcmanager.audit import rpcapi as audit_rpc_client -from dcmanager.tests.unit.api import test_root_controller as testroot -from dcmanager.tests import utils - -FAKE_URL = '/v1.0/notifications' -FAKE_TENANT = utils.UUID1 -FAKE_HEADERS = {'X-Tenant-Id': FAKE_TENANT, 'X_ROLE': 'admin,member,reader', - 'X-Identity-Status': 'Confirmed', 'X-Project-Name': 'admin'} +from dcmanager.tests.unit.api.test_root_controller import DCManagerApiTest +from dcmanager.tests.unit.common import consts as test_consts -class TestNotificationsController(testroot.DCManagerApiTest): +class BaseTestNotificationsController(DCManagerApiTest): + """Base class for testing the NotificationsController""" + def setUp(self): - super(TestNotificationsController, self).setUp() - self.ctx = utils.dummy_context() + super().setUp() - @mock.patch.object(audit_rpc_client, 'ManagerAuditClient') - def test_post(self, mock_audit_rpc_client): - mock_audit_rpc_client().trigger_load_audit.return_value = None - post_url = FAKE_URL - params = json.dumps({'events': ['platform-upgrade-completed']}) - response = self.app.post(post_url, params=params, headers=FAKE_HEADERS) - self.assertEqual(response.content_type, 'application/json') - self.assertEqual(response.status_code, http_client.OK) - mock_audit_rpc_client().trigger_load_audit.assert_called_once_with( - mock.ANY) + self.url = '/v1.0/notifications' - @mock.patch.object(audit_rpc_client, 'ManagerAuditClient') - def test_post_k8(self, mock_audit_rpc_client): - mock_audit_rpc_client().trigger_kubernetes_audit.return_value = None - post_url = FAKE_URL - params = json.dumps({'events': ['k8s-upgrade-completed']}) - response = self.app.post(post_url, params=params, headers=FAKE_HEADERS) - self.assertEqual(response.content_type, 'application/json') - self.assertEqual(response.status_code, http_client.OK) - mock_audit_rpc_client().trigger_kubernetes_audit.assert_called_once_with( - mock.ANY) + self._mock_audit_rpc_client() + + +class TestNotificationsController(BaseTestNotificationsController): + """Test class for NotificationsController""" + + def setUp(self): + super().setUp() + + def test_unmapped_method(self): + """Test requesting an unmapped method results in success with null content""" + + self.method = self.app.put + + response = self._send_request() + + self._assert_response(response) + self.assertEqual(response.text, 'null') + + +class TestNotificationsControllerPost(BaseTestNotificationsController): + """Test class for post requests""" + + def setUp(self): + super().setUp() + + self.method = self.app.post + + def test_post_succeeds_with_platform_upgrade_completed(self): + """Test post succeeds with platform upgrade completed event""" + + self.params = json.dumps({'events': ['platform-upgrade-completed']}) + + response = self._send_request() + + self.mock_audit_rpc_client().trigger_load_audit.assert_called_once() + + self._assert_response(response) + + def test_post_succeeds_with_k8s_upgrade_completed(self): + """Test post succeeds with k8s upgrade completed event""" + + self.params = json.dumps({'events': ['k8s-upgrade-completed']}) + + response = self._send_request() + + self.mock_audit_rpc_client().trigger_kubernetes_audit.assert_called_once() + + self._assert_response(response) + + def test_post_succeeds_with_kube_rootca_update_completed_in_events(self): + """Tests post succeeds when kube-rootca-update-completed in events""" + + self.params = json.dumps({'events': ['kube-rootca-update-completed']}) + + response = self._send_request() + + self.mock_audit_rpc_client().\ + trigger_kube_rootca_update_audit.assert_called_once() + + self._assert_response(response) + + def test_post_fails_without_events_in_request_body(self): + """Tests post fails when body doesn't have events""" + + self.params = json.dumps({}) + + response = self._send_request() + + self._assert_response( + response, http.client.BAD_REQUEST, content_type=test_consts.TEXT_PLAIN + ) + self._assert_pecan( + http.client.BAD_REQUEST, "Missing required notification events" + )