rwlaunchpad/plugins/rwstagingmgr/test/utest_staging_store.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
  20 import argparse
  21 import logging
  22 import io
  23 import os
  24 import sys
  25 import tempfile
  26 import unittest
  27 import xmlrunner
  28
  29 #Setting RIFT_VAR_ROOT if not already set for unit test execution
  30 if "RIFT_VAR_ROOT" not in os.environ:
  31     os.environ['RIFT_VAR_ROOT'] = os.path.join(os.environ['RIFT_INSTALL'], 'var/rift/unittest')
  32
  33 from rift.tasklets.rwstagingmgr.store import StagingFileStore
  34 from rift.mano.utils.project import ManoProject, DEFAULT_PROJECT
  35
  36 import gi
  37 gi.require_version('RwStagingMgmtYang', '1.0')
  38 from gi.repository import (
  39         RwStagingMgmtYang,
  40         )
  41
  42 class MockTasklet(object):
  43     def __init__(self):
  44         self.log = logging.getLogger()
  45         self.projects = {}
  46         project = ManoProject(self.log, name=DEFAULT_PROJECT)
  47         project.publisher = None
  48         self.projects[project.name] = project
  49
  50     def set_delegate(self, store):
  51         self.projects[DEFAULT_PROJECT].publisher = store
  52
  53
  54 class TestSerializer(unittest.TestCase):
  55
  56     def test_staging_area_create(self):
  57         """
  58         1. Verify a valid id is created
  59         2. if the folder and meta files were created.
  60         3. Verify if the meta file has been created.
  61
  62         """
  63         tmp_dir = tempfile.mkdtemp()
  64         tasklet = MockTasklet()
  65         store = StagingFileStore(tasklet, root_dir=tmp_dir)
  66
  67         mock_model = RwStagingMgmtYang.YangData_RwProject_Project_StagingAreas_StagingArea.from_dict({})
  68         stg = store.create_staging_area(mock_model)
  69         mock_id = stg.model.area_id
  70
  71         assert mock_id == store.get_staging_area(mock_id).model.area_id
  72         area_path = os.path.join(store.root_dir, mock_id)
  73         print (area_path)
  74         assert os.path.isdir(area_path)
  75
  76     def test_staging_area_remove(self):
  77         """
  78         1. Verify a valid id is created
  79         2. if the folder and meta files were created.
  80         3. Verify if the meta file has been created.
  81
  82         """
  83         tmp_dir = tempfile.mkdtemp()
  84         tasklet = MockTasklet()
  85         store = StagingFileStore(tasklet, root_dir=tmp_dir)
  86
  87         mock_model = RwStagingMgmtYang.YangData_RwProject_Project_StagingAreas_StagingArea.from_dict({})
  88         # get the wrapped mock model
  89         mock_model = store.create_staging_area(mock_model)
  90         mock_id = mock_model.model.area_id
  91         area_path = os.path.join(store.root_dir, mock_id)
  92
  93         # check if dir actually exists
  94         assert os.path.isdir(area_path)
  95         store.remove_staging_area(mock_model)
  96
  97         assert not os.path.isdir(area_path)
  98
  99 def main(argv=sys.argv[1:]):
 100     logging.basicConfig(format='TEST %(message)s')
 101
 102     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 103     parser = argparse.ArgumentParser()
 104     parser.add_argument('-v', '--verbose', action='store_true')
 105     parser.add_argument('-n', '--no-runner', action='store_true')
 106
 107     args, unknown = parser.parse_known_args(argv)
 108     if args.no_runner:
 109         runner = None
 110
 111     # Set the global logging level
 112     logging.getLogger().setLevel(logging.DEBUG if args.verbose else logging.ERROR)
 113
 114     # The unittest framework requires a program name, so use the name of this
 115     # file instead (we do not want to have to pass a fake program name to main
 116     # when this is called from the interpreter).
 117     unittest.main(argv=[__file__] + unknown + ["-v"], testRunner=runner)
 118
 119 if __name__ == '__main__':
 120     main()