rwlaunchpad/plugins/rwpkgmgr/test/utest_publisher_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 argparse
  20 import asyncio
  21 import logging
  22 import mock
  23 import os
  24 import sys
  25 import unittest
  26 import uuid
  27 import xmlrunner
  28
  29 import gi
  30 gi.require_version('RwDts', '1.0')
  31 gi.require_version('RwPkgMgmtYang', '1.0')
  32 from gi.repository import (
  33         RwDts as rwdts,
  34         RwPkgMgmtYang
  35         )
  36 import rift.tasklets.rwpkgmgr.downloader as downloader
  37 import rift.tasklets.rwpkgmgr.publisher as pkg_publisher
  38 import rift.test.dts
  39
  40
  41 class TestCase(rift.test.dts.AbstractDTSTest):
  42     @classmethod
  43     def configure_schema(cls):
  44         return RwPkgMgmtYang.get_schema()
  45
  46     @classmethod
  47     def configure_timeout(cls):
  48         return 240
  49
  50     def configure_test(self, loop, test_id):
  51         self.log.debug("STARTING - %s", test_id)
  52         self.tinfo = self.new_tinfo(str(test_id))
  53         self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop)
  54
  55         self.job_handler = pkg_publisher.DownloadStatusPublisher(self.log, self.dts, self.loop)
  56
  57     def tearDown(self):
  58         super().tearDown()
  59
  60     @asyncio.coroutine
  61     def get_published_xpaths(self):
  62         published_xpaths = set()
  63
  64         res_iter = yield from self.dts.query_read("D,/rwdts:dts")
  65         for i in res_iter:
  66             res = (yield from i).result
  67             for member in res.member:
  68                 published_xpaths |= {reg.keyspec for reg in member.state.registration if reg.flags == "publisher"}
  69
  70         return published_xpaths
  71
  72     @asyncio.coroutine
  73     def read_xpath(self, xpath):
  74         itr = yield from self.dts.query_read(xpath)
  75
  76         result = None
  77         for fut in itr:
  78             result = yield from fut
  79             return result.result
  80
  81     @rift.test.dts.async_test
  82     def test_download_publisher(self):
  83         yield from self.job_handler.register()
  84         published_xpaths = yield from self.get_published_xpaths()
  85         assert self.job_handler.xpath() in published_xpaths
  86
  87     @rift.test.dts.async_test
  88     def test_publish(self):
  89         """
  90         Asserts:
  91             1. Verify if an update on_download_progess & on_download_finished
  92                triggers a DTS update
  93             2. Verify if the internal store is updated
  94         """
  95         yield from self.job_handler.register()
  96
  97         mock_msg = RwPkgMgmtYang.DownloadJob.from_dict({
  98                 "url": "http://foo/bar",
  99                 "package_id": "123",
 100                 "download_id": str(uuid.uuid4())})
 101
 102         self.job_handler.on_download_progress(mock_msg)
 103         yield from asyncio.sleep(5, loop=self.loop)
 104
 105         itr = yield from self.dts.query_read("/download-jobs/job[download-id='{}']".format(
 106             mock_msg.download_id))
 107
 108         result = None
 109         for fut in itr:
 110             result = yield from fut
 111             result = result.result
 112
 113         print (mock_msg)
 114         assert result == mock_msg
 115
 116         # Modify the msg
 117         mock_msg.url = "http://bar/foo"
 118         self.job_handler.on_download_finished(mock_msg)
 119         yield from asyncio.sleep(5, loop=self.loop)
 120
 121         itr = yield from self.dts.query_read("/download-jobs/job[download-id='{}']".format(
 122             mock_msg.download_id))
 123
 124         result = None
 125         for fut in itr:
 126             result = yield from fut
 127             result = result.result
 128         assert result == mock_msg
 129
 130
 131     @rift.test.dts.async_test
 132     def test_url_download(self):
 133         """
 134         Integration Test:
 135             Test the updates with download/url.py
 136         """
 137         yield from self.job_handler.register()
 138
 139         proxy = mock.MagicMock()
 140
 141         url = "http://boson.eng.riftio.com/common/unittests/rift-shell"
 142         url_downloader = downloader.PackageFileDownloader(url, "1", "/", "VNFD", proxy)
 143
 144         download_id = yield from self.job_handler.register_downloader(url_downloader)
 145         assert download_id is not None
 146
 147         yield from asyncio.sleep(5, loop=self.loop)
 148         xpath = "/download-jobs/job[download-id='{}']".format(
 149             download_id)
 150         result = yield from self.read_xpath(xpath)
 151         print (result)
 152         assert result.status == "COMPLETED"
 153         assert len(self.job_handler.tasks) == 0
 154
 155
 156     @rift.test.dts.async_test
 157     def test_cancelled(self):
 158         """
 159         Integration Test:
 160             1. Test the updates with downloader.py
 161             2. Verifies if cancel triggers the job status to move to cancelled
 162         """
 163         yield from self.job_handler.register()
 164
 165         proxy = mock.MagicMock()
 166         url = "http://boson.eng.riftio.com/common/unittests/Fedora-x86_64-20-20131211.1-sda-ping.qcow2"
 167         url_downloader = downloader.PackageFileDownloader(url, "1", "/", "VNFD", proxy)
 168
 169         download_id = yield from self.job_handler.register_downloader(url_downloader)
 170         assert download_id is not None
 171         xpath = "/download-jobs/job[download-id='{}']".format(
 172             download_id)
 173
 174         yield from asyncio.sleep(3, loop=self.loop)
 175
 176         result = yield from self.read_xpath(xpath)
 177         assert result.status == "IN_PROGRESS"
 178
 179         yield from self.job_handler.cancel_download(download_id)
 180         yield from asyncio.sleep(3, loop=self.loop)
 181         result = yield from self.read_xpath(xpath)
 182         assert result.status == "CANCELLED"
 183         assert len(self.job_handler.tasks) == 0
 184
 185
 186 def main():
 187     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 188
 189     parser = argparse.ArgumentParser()
 190     parser.add_argument('-v', '--verbose', action='store_true')
 191     parser.add_argument('-n', '--no-runner', action='store_true')
 192     args, unittest_args = parser.parse_known_args()
 193     if args.no_runner:
 194         runner = None
 195
 196     TestCase.log_level = logging.DEBUG if args.verbose else logging.WARN
 197
 198     unittest.main(testRunner=runner, argv=[sys.argv[0]] + unittest_args)
 199
 200 if __name__ == '__main__':
 201     main()