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 import gi
  37 import requests
  38 from tornado.platform.asyncio import AsyncIOMainLoop
  39 from tornado.ioloop import IOLoop
  40 from concurrent.futures.thread import ThreadPoolExecutor
  41 from concurrent.futures.process import ProcessPoolExecutor
  42 gi.require_version('RwDts', '1.0')
  43 gi.require_version('RwPkgMgmtYang', '1.0')
  44 from gi.repository import (
  45         RwDts as rwdts,
  46         RwPkgMgmtYang,
  47         RwVnfdYang
  48
  49         )
  50
  51 import rift.tasklets.rwlaunchpad.uploader as uploader
  52 import rift.tasklets.rwlaunchpad.message as message
  53 import rift.tasklets.rwlaunchpad.export as export
  54 import rift.test.dts
  55 import mock
  56
  57 TEST_STRING = "foobar"
  58
  59 class TestCase(rift.test.dts.AbstractDTSTest):
  60     @classmethod
  61     def configure_schema(cls):
  62         return RwPkgMgmtYang.get_schema()
  63
  64     @classmethod
  65     def configure_timeout(cls):
  66         return 240
  67
  68     def configure_test(self, loop, test_id):
  69         self.log.debug("STARTING - %s", test_id)
  70         self.tinfo = self.new_tinfo(str(test_id))
  71         self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop)
  72
  73
  74         mock_vnfd_catalog = mock.MagicMock()
  75         self.uid, path = self.create_mock_package()
  76
  77         mock_vnfd = RwVnfdYang.YangData_Vnfd_VnfdCatalog_Vnfd.from_dict({
  78               "id": self.uid
  79             })
  80         mock_vnfd_catalog = {self.uid: mock_vnfd}
  81
  82         self.app = uploader.UploaderApplication(
  83                 self.log,
  84                 self.dts,
  85                 self.loop,
  86                 vnfd_catalog=mock_vnfd_catalog)
  87
  88         AsyncIOMainLoop().install()
  89         self.server = tornado.httpserver.HTTPServer(
  90             self.app,
  91             io_loop=IOLoop.current(),
  92         )
  93
  94     def tearDown(self):
  95         super().tearDown()
  96
  97     def create_mock_package(self):
  98         uid = str(uuid.uuid4())
  99         path = os.path.join(
 100                 os.getenv('RIFT_ARTIFACTS'),
 101                 "launchpad/packages/vnfd",
 102                 uid)
 103
 104         package_path = os.path.join(path, "pong_vnfd")
 105
 106         os.makedirs(package_path)
 107         open(os.path.join(path, "pong_vnfd.xml"), "wb").close()
 108         open(os.path.join(path, "logo.png"), "wb").close()
 109
 110         return uid, path
 111
 112     @rift.test.dts.async_test
 113     def test_package_create_rpc(self):
 114         """
 115             1. Verify the package-create RPC handler
 116             2. Check if the log messages are updated which will be used by UI
 117                 for polling
 118             3. Verify the package-update RPC handler
 119             4. Check if the log messages are updated which will be used by UI
 120                 for polling
 121         """
 122         yield from self.app.register()
 123         ip = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageCreate.from_dict({
 124                 "package_type": "VNFD",
 125                 "external_url":  "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
 126                 })
 127
 128         rpc_out = yield from self.dts.query_rpc(
 129                     "I,/rw-pkg-mgmt:package-create",
 130                     rwdts.XactFlag.TRACE,
 131                     ip)
 132
 133         trans_id = None
 134         for itr in rpc_out:
 135             result = yield from itr
 136             trans_id = result.result.transaction_id
 137
 138         assert trans_id is not None
 139
 140         yield from asyncio.sleep(5, loop=self.loop)
 141         # Verify the message logs
 142         data = self.app.messages[trans_id]
 143         assert data is not None
 144         data = data[1]
 145         assert type(data) is message.DownloadSuccess
 146
 147         # Update
 148         ip = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageUpdate.from_dict({
 149                 "package_type": "VNFD",
 150                 "external_url":  "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
 151                 })
 152         rpc_out = yield from self.dts.query_rpc(
 153                     "I,/rw-pkg-mgmt:package-update",
 154                     rwdts.XactFlag.TRACE,
 155                     ip)
 156
 157         trans_id = None
 158         for itr in rpc_out:
 159             result = yield from itr
 160             trans_id = result.result.transaction_id
 161
 162         assert trans_id is not None
 163         yield from asyncio.sleep(5, loop=self.loop)
 164         # Verify the message logs
 165         data = self.app.messages[trans_id]
 166         assert data is not None
 167         data = data[1]
 168         assert type(data) is message.DownloadSuccess
 169
 170
 171     @rift.test.dts.async_test
 172     def test_package_export(self):
 173         """
 174             1. Verify if the package export RPC handler work
 175             2. A file is actually generated in the exports dir.
 176         """
 177         yield from self.app.register()
 178         ip = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageExport.from_dict({
 179                 "package_type": "VNFD",
 180                 "package_id": self.uid
 181                 })
 182
 183         rpc_out = yield from self.dts.query_rpc(
 184                     "I,/rw-pkg-mgmt:package-export",
 185                     rwdts.XactFlag.TRACE,
 186                     ip)
 187
 188         trans_id = None
 189         filename = None
 190         for itr in rpc_out:
 191             result = yield from itr
 192             trans_id = result.result.transaction_id
 193             filename = result.result.filename
 194
 195         assert trans_id is not None
 196
 197         # Verify the message logs
 198         data = self.app.messages[trans_id]
 199         assert data is not None
 200         data = data[-1]
 201         assert type(data) is export.ExportSuccess
 202         path = os.path.join(
 203                 os.getenv("RIFT_ARTIFACTS"),
 204                 "launchpad/exports",
 205                 filename)
 206
 207
 208         print (path)
 209         assert os.path.isfile(path)
 210
 211
 212 def main():
 213     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 214
 215     parser = argparse.ArgumentParser()
 216     parser.add_argument('-v', '--verbose', action='store_true')
 217     parser.add_argument('-n', '--no-runner', action='store_true')
 218     args, unittest_args = parser.parse_known_args()
 219     if args.no_runner:
 220         runner = None
 221     logging.basicConfig(format='TEST %(message)s')
 222     logging.getLogger().setLevel(logging.DEBUG)
 223
 224
 225     unittest.main(testRunner=runner, argv=[sys.argv[0]] + unittest_args)
 226
 227 if __name__ == '__main__':
 228     main()