rwlaunchpad/test/utest_ro_account.py

   1
   2 #
   3 #   Copyright 2016 RIFT.IO Inc
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17
  18 import asyncio
  19 import sys
  20 import types
  21 import unittest
  22 import uuid
  23 import os
  24
  25 #Setting RIFT_VAR_ROOT if not already set for unit test execution
  26 if "RIFT_VAR_ROOT" not in os.environ:
  27     os.environ['RIFT_VAR_ROOT'] = os.path.join(os.environ['RIFT_INSTALL'], 'var/rift/unittest')
  28
  29 import rift.test.dts
  30 import rift.tasklets.rwnsmtasklet.cloud as cloud
  31 import rift.tasklets.rwnsmtasklet.openmano_nsm as openmano_nsm
  32 import rw_peas
  33
  34 import gi
  35 gi.require_version('RwDtsYang', '1.0')
  36 from gi.repository import (
  37         RwLaunchpadYang as launchpadyang,
  38         RwDts as rwdts,
  39         RwVnfdYang,
  40         RwVnfrYang,
  41         RwNsrYang,
  42         RwNsdYang,
  43         VnfrYang
  44         )
  45
  46
  47 class DescriptorPublisher(object):
  48     def __init__(self, log, dts, loop):
  49         self.log = log
  50         self.loop = loop
  51         self.dts = dts
  52
  53         self._registrations = []
  54
  55     @asyncio.coroutine
  56     def publish(self, w_path, path, desc):
  57         ready_event = asyncio.Event(loop=self.loop)
  58
  59         @asyncio.coroutine
  60         def on_ready(regh, status):
  61             self.log.debug("Create element: %s, obj-type:%s obj:%s",
  62                            path, type(desc), desc)
  63             with self.dts.transaction() as xact:
  64                 regh.create_element(path, desc, xact.xact)
  65             self.log.debug("Created element: %s, obj:%s", path, desc)
  66             ready_event.set()
  67
  68         handler = rift.tasklets.DTS.RegistrationHandler(
  69                 on_ready=on_ready
  70                 )
  71
  72         self.log.debug("Registering path: %s, obj:%s", w_path, desc)
  73         reg = yield from self.dts.register(
  74                 w_path,
  75                 handler,
  76                 flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ
  77                 )
  78         self._registrations.append(reg)
  79         self.log.debug("Registered path : %s", w_path)
  80         yield from ready_event.wait()
  81
  82         return reg
  83
  84     def unpublish_all(self):
  85         self.log.debug("Deregistering all published descriptors")
  86         for reg in self._registrations:
  87             reg.deregister()
  88
  89 class RoAccountDtsTestCase(rift.test.dts.AbstractDTSTest):
  90     @classmethod
  91     def configure_schema(cls):
  92        return launchpadyang.get_schema()
  93
  94     @classmethod
  95     def configure_timeout(cls):
  96         return 240
  97
  98     def configure_test(self, loop, test_id):
  99         self.log.debug("STARTING - %s", test_id)
 100         self.tinfo = self.new_tinfo(str(test_id))
 101         self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop)
 102
 103         self.tinfo_sub = self.new_tinfo(str(test_id) + "_sub")
 104         self.dts_sub = rift.tasklets.DTS(self.tinfo_sub, self.schema, self.loop)
 105
 106         self.publisher = DescriptorPublisher(self.log, self.dts, self.loop)
 107
 108     def tearDown(self):
 109         super().tearDown()
 110
 111     @rift.test.dts.async_test
 112     def test_orch_account_create(self):
 113         orch = cloud.ROAccountPluginSelector(self.dts, self.log, self.loop, None)
 114
 115         yield from orch.register()
 116
 117         # Test if we have a default plugin in case no RO is specified.
 118         assert type(orch.ro_plugin) is cloud.RwNsPlugin
 119         mock_orch_acc = launchpadyang.ResourceOrchestrator.from_dict(
 120                 {'name': 'rift-ro', 'account_type': 'rift_ro', 'rift_ro': {'rift_ro': True}})
 121
 122         # Test rift-ro plugin CREATE
 123         w_xpath = "C,/rw-launchpad:resource-orchestrator"
 124         xpath = w_xpath
 125         yield from self.publisher.publish(w_xpath, xpath, mock_orch_acc)
 126         yield from asyncio.sleep(5, loop=self.loop)
 127
 128         assert type(orch.ro_plugin) is cloud.RwNsPlugin
 129
 130         # Test Openmano plugin CREATE
 131         mock_orch_acc = launchpadyang.ResourceOrchestrator.from_dict(
 132                 {'name': 'openmano',
 133                  'account_type': 'openmano',
 134                  'openmano': {'tenant_id': "abc",
 135                               "port": 9999,
 136                               "host": "10.64.11.77"}})
 137         yield from self.publisher.publish(w_xpath, xpath, mock_orch_acc)
 138         yield from asyncio.sleep(5, loop=self.loop)
 139
 140         assert type(orch.ro_plugin) is openmano_nsm.OpenmanoNsPlugin
 141         assert orch.ro_plugin._cli_api._port  == mock_orch_acc.openmano.port
 142         assert orch.ro_plugin._cli_api._host  == mock_orch_acc.openmano.host
 143
 144         # Test update
 145         mock_orch_acc.openmano.port = 9789
 146         mock_orch_acc.openmano.host = "10.64.11.78"
 147         yield from self.dts.query_update("C,/rw-launchpad:resource-orchestrator",
 148                 rwdts.XactFlag.ADVISE, mock_orch_acc)
 149         assert orch.ro_plugin._cli_api._port  == mock_orch_acc.openmano.port
 150         assert orch.ro_plugin._cli_api._host  == mock_orch_acc.openmano.host
 151
 152         # Test update when a live instance exists
 153         # Exception should be thrown
 154         orch.handle_nsr(None, rwdts.QueryAction.CREATE)
 155         mock_orch_acc.openmano.port = 9788
 156
 157         with self.assertRaises(Exception):
 158             yield from self.dts.query_update("C,/rw-launchpad:resource-orchestrator",
 159                     rwdts.XactFlag.ADVISE, mock_orch_acc)
 160
 161         # Test delete
 162         yield from self.dts.query_delete("C,/rw-launchpad:resource-orchestrator",
 163                 flags=rwdts.XactFlag.ADVISE)
 164         assert orch.ro_plugin == None
 165
 166
 167 def main(argv=sys.argv[1:]):
 168
 169     # The unittest framework requires a program name, so use the name of this
 170     # file instead (we do not want to have to pass a fake program name to main
 171     # when this is called from the interpreter).
 172     unittest.main(
 173             argv=[__file__] + argv,
 174             testRunner=None#xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 175             )
 176
 177 if __name__ == '__main__':
 178     main()