rwlaunchpad/plugins/rwlaunchpadtasklet/test/utest_uploader_app_dts.py

   1 #!/usr/bin/env python3
   2
   3 #
   4 #   Copyright 2016 RIFT.IO Inc
   5 #
   6 #   Licensed under the Apache License, Version 2.0 (the "License");
   7 #   you may not use this file except in compliance with the License.
   8 #   You may obtain a copy of the License at
   9 #
  10 #       http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 #   Unless required by applicable law or agreed to in writing, software
  13 #   distributed under the License is distributed on an "AS IS" BASIS,
  14 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 #   See the License for the specific language governing permissions and
  16 #   limitations under the License.
  17 #
  18
  19 import asyncio
  20 import argparse
  21 import os
  22 import logging
  23
  24 import shutil
  25 import stat
  26 import sys
  27 import unittest
  28 import uuid
  29 import xmlrunner
  30 import tornado
  31 import tornado.escape
  32 import tornado.ioloop
  33 import tornado.web
  34 import tornado.httputil
  35
  36 #Setting RIFT_VAR_ROOT if not already set for unit test execution
  37 if "RIFT_VAR_ROOT" not in os.environ:
  38     os.environ['RIFT_VAR_ROOT'] = os.path.join(os.environ['RIFT_INSTALL'], 'var/rift/unittest')
  39
  40 import gi
  41 import requests
  42 from tornado.platform.asyncio import AsyncIOMainLoop
  43 from tornado.ioloop import IOLoop
  44 from concurrent.futures.thread import ThreadPoolExecutor
  45 from concurrent.futures.process import ProcessPoolExecutor
  46 gi.require_version('RwDts', '1.0')
  47 gi.require_version('RwPkgMgmtYang', '1.0')
  48 from gi.repository import (
  49         RwDts as rwdts,
  50         RwPkgMgmtYang,
  51         RwVnfdYang
  52
  53         )
  54
  55 import rift.tasklets.rwlaunchpad.uploader as uploader
  56 import rift.tasklets.rwlaunchpad.message as message
  57 import rift.tasklets.rwlaunchpad.export as export
  58 import rift.test.dts
  59 import mock
  60
  61 TEST_STRING = "foobar"
  62
  63 class TestCase(rift.test.dts.AbstractDTSTest):
  64     @classmethod
  65     def configure_schema(cls):
  66         return RwPkgMgmtYang.get_schema()
  67
  68     @classmethod
  69     def configure_timeout(cls):
  70         return 240
  71
  72     def configure_test(self, loop, test_id):
  73         self.log.debug("STARTING - %s", test_id)
  74         self.tinfo = self.new_tinfo(str(test_id))
  75         self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop)
  76
  77
  78         mock_vnfd_catalog = mock.MagicMock()
  79         self.uid, path = self.create_mock_package()
  80
  81         mock_vnfd = RwVnfdYang.YangData_Vnfd_VnfdCatalog_Vnfd.from_dict({
  82               "id": self.uid
  83             })
  84         mock_vnfd_catalog = {self.uid: mock_vnfd}
  85
  86         self.app = uploader.UploaderApplication(
  87                 self.log,
  88                 self.dts,
  89                 self.loop,
  90                 vnfd_catalog=mock_vnfd_catalog)
  91
  92         AsyncIOMainLoop().install()
  93         self.server = tornado.httpserver.HTTPServer(
  94             self.app,
  95             io_loop=IOLoop.current(),
  96         )
  97
  98     def tearDown(self):
  99         super().tearDown()
 100
 101     def create_mock_package(self):
 102         uid = str(uuid.uuid4())
 103         path = os.path.join(
 104                 os.getenv('RIFT_VAR_ROOT'),
 105                 "launchpad/packages/vnfd",
 106                 uid)
 107
 108         package_path = os.path.join(path, "pong_vnfd")
 109
 110         os.makedirs(package_path)
 111         open(os.path.join(path, "pong_vnfd.xml"), "wb").close()
 112         open(os.path.join(path, "logo.png"), "wb").close()
 113
 114         return uid, path
 115
 116     @rift.test.dts.async_test
 117     def test_package_create_rpc(self):
 118         """
 119             1. Verify the package-create RPC handler
 120             2. Check if the log messages are updated which will be used by UI
 121                 for polling
 122             3. Verify the package-update RPC handler
 123             4. Check if the log messages are updated which will be used by UI
 124                 for polling
 125         """
 126         yield from self.app.register()
 127         ip = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageCreate.from_dict({
 128                 "package_type": "VNFD",
 129                 "external_url":  "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
 130                 })
 131
 132         rpc_out = yield from self.dts.query_rpc(
 133                     "I,/rw-pkg-mgmt:package-create",
 134                     rwdts.XactFlag.TRACE,
 135                     ip)
 136
 137         trans_id = None
 138         for itr in rpc_out:
 139             result = yield from itr
 140             trans_id = result.result.transaction_id
 141
 142         assert trans_id is not None
 143
 144         yield from asyncio.sleep(5, loop=self.loop)
 145         # Verify the message logs
 146         data = self.app.messages[trans_id]
 147         assert data is not None
 148         data = data[1]
 149         assert type(data) is message.DownloadSuccess
 150
 151         # Update
 152         ip = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageUpdate.from_dict({
 153                 "package_type": "VNFD",
 154                 "external_url":  "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
 155                 })
 156         rpc_out = yield from self.dts.query_rpc(
 157                     "I,/rw-pkg-mgmt:package-update",
 158                     rwdts.XactFlag.TRACE,
 159                     ip)
 160
 161         trans_id = None
 162         for itr in rpc_out:
 163             result = yield from itr
 164             trans_id = result.result.transaction_id
 165
 166         assert trans_id is not None
 167         yield from asyncio.sleep(5, loop=self.loop)
 168         # Verify the message logs
 169         data = self.app.messages[trans_id]
 170         assert data is not None
 171         data = data[1]
 172         assert type(data) is message.DownloadSuccess
 173
 174
 175     @rift.test.dts.async_test
 176     def test_package_export(self):
 177         """
 178             1. Verify if the package export RPC handler work
 179             2. A file is actually generated in the exports dir.
 180         """
 181         yield from self.app.register()
 182         ip = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageExport.from_dict({
 183                 "package_type": "VNFD",
 184                 "package_id": self.uid
 185                 })
 186
 187         rpc_out = yield from self.dts.query_rpc(
 188                     "I,/rw-pkg-mgmt:package-export",
 189                     rwdts.XactFlag.TRACE,
 190                     ip)
 191
 192         trans_id = None
 193         filename = None
 194         for itr in rpc_out:
 195             result = yield from itr
 196             trans_id = result.result.transaction_id
 197             filename = result.result.filename
 198
 199         assert trans_id is not None
 200
 201         # Verify the message logs
 202         data = self.app.messages[trans_id]
 203         assert data is not None
 204         data = data[-1]
 205         assert type(data) is export.ExportSuccess
 206         path = os.path.join(
 207                 os.getenv("RIFT_VAR_ROOT"),
 208                 "launchpad/exports",
 209                 filename)
 210
 211
 212         print (path)
 213         assert os.path.isfile(path)
 214
 215
 216 def main():
 217     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 218
 219     parser = argparse.ArgumentParser()
 220     parser.add_argument('-v', '--verbose', action='store_true')
 221     parser.add_argument('-n', '--no-runner', action='store_true')
 222     args, unittest_args = parser.parse_known_args()
 223     if args.no_runner:
 224         runner = None
 225     logging.basicConfig(format='TEST %(message)s')
 226     logging.getLogger().setLevel(logging.DEBUG)
 227
 228
 229     unittest.main(testRunner=runner, argv=[sys.argv[0]] + unittest_args)
 230
 231 if __name__ == '__main__':
 232     main()