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 import xmlrunner
  25
  26 #Setting RIFT_VAR_ROOT if not already set for unit test execution
  27 if "RIFT_VAR_ROOT" not in os.environ:
  28     os.environ['RIFT_VAR_ROOT'] = os.path.join(os.environ['RIFT_INSTALL'], 'var/rift/unittest')
  29
  30 import rift.test.dts
  31 import rift.tasklets.rwnsmtasklet.cloud as cloud
  32 import rift.tasklets.rwnsmtasklet.rwnsmplugin as rwnsmplugin
  33 import rift.tasklets.rwnsmtasklet.openmano_nsm as openmano_nsm
  34 from rift.mano.utils.project import ManoProject
  35 import rw_peas
  36
  37 import gi
  38 gi.require_version('RwDts', '1.0')
  39 from gi.repository import (
  40         RwRoAccountYang as roaccountyang,
  41         RwDts as rwdts,
  42         RwProjectVnfdYang as RwVnfdYang,
  43         RwVnfrYang,
  44         RwNsrYang,
  45         RwProjectNsdYang as RwNsdYang,
  46         VnfrYang,
  47         )
  48
  49
  50 class DescriptorPublisher(object):
  51     def __init__(self, log, dts, loop):
  52         self.log = log
  53         self.loop = loop
  54         self.dts = dts
  55         self._registrations = []
  56
  57     @asyncio.coroutine
  58     def update(self, xpath, desc):
  59         self._registrations[-1].update_element(xpath, desc)
  60
  61     @asyncio.coroutine
  62     def delete(self, xpath):
  63         self._registrations[-1].delete_element(xpath)
  64
  65     @asyncio.coroutine
  66     def publish(self, w_path, path, desc):
  67         ready_event = asyncio.Event(loop=self.loop)
  68
  69         @asyncio.coroutine
  70         def on_ready(regh, status):
  71             self.log.debug("Create element: %s, obj-type:%s obj:%s",
  72                            path, type(desc), desc)
  73             with self.dts.transaction() as xact:
  74                 regh.create_element(path, desc, xact.xact)
  75             self.log.debug("Created element: %s, obj:%s", path, desc)
  76             ready_event.set()
  77
  78         handler = rift.tasklets.DTS.RegistrationHandler(
  79                 on_ready=on_ready
  80                 )
  81
  82         self.log.debug("Registering path: %s, obj:%s", w_path, desc)
  83
  84         reg = yield from self.dts.register(
  85                 w_path,
  86                 handler,
  87                 flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ
  88                 )
  89
  90         self._registrations.append(reg)
  91         self.log.debug("Registered path : %s", w_path)
  92         yield from ready_event.wait()
  93
  94         return reg
  95
  96     def unpublish_all(self):
  97         self.log.debug("Deregistering all published descriptors")
  98         for reg in self._registrations:
  99             reg.deregister()
 100
 101 class RoAccountDtsTestCase(rift.test.dts.AbstractDTSTest):
 102     @classmethod
 103     def configure_schema(cls):
 104        return roaccountyang.get_schema()
 105
 106     @classmethod
 107     def configure_timeout(cls):
 108         return 240
 109
 110     def configure_test(self, loop, test_id):
 111         self.log.debug("STARTING - %s", test_id)
 112         self.tinfo = self.new_tinfo(str(test_id))
 113         self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop)
 114         self.project = ManoProject(self.log)
 115
 116         self.tinfo_sub = self.new_tinfo(str(test_id) + "_sub")
 117         self.dts_sub = rift.tasklets.DTS(self.tinfo_sub, self.schema, self.loop)
 118
 119         self.publisher = DescriptorPublisher(self.log, self.dts, self.loop)
 120
 121     def tearDown(self):
 122         super().tearDown()
 123
 124     @rift.test.dts.async_test
 125     def test_orch_account_create(self):
 126         ro_cfg_sub = cloud.ROAccountConfigSubscriber(self.dts, self.log, self.loop, self.project, None)
 127         yield from ro_cfg_sub.register()
 128
 129         ro_plugin = ro_cfg_sub.get_ro_plugin(account_name=None)
 130         # Test if we have a default plugin in case no RO is specified.
 131         assert type(ro_plugin) is rwnsmplugin.RwNsPlugin
 132
 133         # Test rift-ro plugin CREATE
 134         w_xpath = self.project.add_project("C,/rw-ro-account:ro-account/rw-ro-account:account")
 135         xpath = w_xpath + "[rw-ro-account:name='openmano']"
 136
 137         # Test Openmano plugin CREATE
 138         mock_orch_acc = roaccountyang.YangData_RwProject_Project_RoAccount_Account.from_dict(
 139                 {'name': 'openmano',
 140                  'ro_account_type': 'openmano',
 141                  'openmano': {'tenant_id': "abc",
 142                               "port": 9999,
 143                               "host": "10.64.11.77"}})
 144
 145         yield from self.publisher.publish(w_xpath, xpath, mock_orch_acc)
 146         yield from asyncio.sleep(5, loop=self.loop)
 147
 148         ro_plugin = ro_cfg_sub.get_ro_plugin(account_name='openmano')
 149         assert type(ro_plugin) is openmano_nsm.OpenmanoNsPlugin
 150
 151         # Test update
 152         mock_orch_acc.openmano.port = 9789
 153         mock_orch_acc.openmano.host = "10.64.11.78"
 154         yield from self.publisher.update(xpath, mock_orch_acc)
 155         yield from asyncio.sleep(5, loop=self.loop)
 156
 157         #Since update means delete followed by a insert get the new ro_plugin.
 158         ro_plugin = ro_cfg_sub.get_ro_plugin(account_name='openmano')
 159         assert ro_plugin._cli_api._port  == mock_orch_acc.openmano.port
 160         assert ro_plugin._cli_api._host  == mock_orch_acc.openmano.host
 161
 162         # Test delete to be implemented. right now facing some dts issues.
 163         # Use DescriptorPublisher delete for deletion
 164
 165 def main(argv=sys.argv[1:]):
 166
 167     # The unittest framework requires a program name, so use the name of this
 168     # file instead (we do not want to have to pass a fake program name to main
 169     # when this is called from the interpreter).
 170     unittest.main(
 171             argv=[__file__] + argv,
 172             testRunner=xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 173             )
 174
 175 if __name__ == '__main__':
 176     main()