"Fossies" - the Fresh Open Source Software Archive

Member "masakari-9.0.0/masakari/tests/functional/test_vm_notifications.py" (13 May 2020, 3138 Bytes) of package /linux/misc/openstack/masakari-9.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_vm_notifications.py": 8.0.0_vs_9.0.0.

    1 # Copyright (C) 2019 NTT DATA
    2 # All Rights Reserved.
    3 #
    4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 # not use this file except in compliance with the License. You may obtain
    6 # a copy of the License at
    7 #
    8 #      http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 # Unless required by applicable law or agreed to in writing, software
   11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 # License for the specific language governing permissions and limitations
   14 # under the License.
   15 
   16 from oslo_utils import timeutils
   17 
   18 from masakari.objects import fields
   19 from masakari.tests.functional import notification_base as base
   20 
   21 
   22 class NotificationVMTestCase(base.NotificationTestBase):
   23 
   24     NOTIFICATION_TYPE = "VM"
   25     NOTIFICATION_WAIT_INTERVAL = 1
   26     NOTIFICATION_WAIT_PERIOD = 600
   27 
   28     def setUp(self, ha_api_version="1.0"):
   29         super(NotificationVMTestCase, self).setUp(ha_api_version)
   30 
   31     def _test_create_notification(self):
   32         # Create server
   33         server = self.conn.compute.create_server(
   34             name='masakari_test', flavorRef=self.flavors[0],
   35             imageRef=self.image_uuids[0],
   36             networks=[{'uuid': self.private_net}],
   37             metadata={'HA_Enabled': 'True'})
   38 
   39         self.addCleanup(self.conn.compute.delete_server, server)
   40         self.check_server_status(server, 'ACTIVE')
   41 
   42         self.admin_conn.compute.stop_server(server.id)
   43 
   44         self.check_server_status(server, 'SHUTOFF')
   45 
   46         notification = self.admin_conn.ha.create_notification(
   47             type=self.NOTIFICATION_TYPE,
   48             hostname=self.host.name,
   49             generated_time=timeutils.utcnow().replace(microsecond=0),
   50             payload={"instance_uuid": server.id,
   51                      "vir_domain_event": "STOPPED_FAILED",
   52                      "event": "LIFECYCLE"})
   53 
   54         self.check_notification_status(notification,
   55                                        self.NOTIFICATION_WAIT_INTERVAL,
   56                                        self.NOTIFICATION_WAIT_PERIOD)
   57 
   58         notification = self.admin_conn.instance_ha.get_notification(
   59             notification.notification_uuid)
   60 
   61         result = self.admin_conn.compute.get_server(server.id)
   62 
   63         self.assertEqual(fields.NotificationStatus.FINISHED,
   64                          notification.status)
   65         self.assertEqual('ACTIVE', result.status)
   66 
   67         return notification
   68 
   69     def test_create_notification(self):
   70         # Test to create notification for VM notification type
   71 
   72         self._test_create_notification()
   73 
   74 
   75 class NotificationVMTestCase_V1_1(NotificationVMTestCase):
   76 
   77     def setUp(self):
   78         super(NotificationVMTestCase, self).setUp("1.1")
   79 
   80     def test_create_notification(self):
   81         notification = self._test_create_notification()
   82         self.assertIsNotNone(notification.recovery_workflow_details)
   83         recovery_details = notification.recovery_workflow_details
   84         # check the status of each task is successful
   85         for details in recovery_details:
   86             self.assertEqual("SUCCESS", details.state)