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