lib/xos-synchronizer/xos-synchronizer-tests/test_loadmodels.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import unittest
from mock import patch, Mock, MagicMock

from io import StringIO
import functools
import os
import sys

test_path = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))


def mock_listdir(dir_map, dir):
    """ mock os.listdir() """
    return dir_map.get(dir, [])


def mock_exists(file_map, fn):
    """ mock os.path.exists() """
    return (fn in file_map)


def mock_open(orig_open, file_map, fn, *args, **kwargs):
    """ mock file open() """
    if fn in file_map:
        return StringIO(file_map[fn])
    else:
        return orig_open(fn, *args, **kwargs)


class ItemList(object):
    """ mock the various items within a LoadModelsRequest protobuf """

    def __init__(self):
        self.items = []

    def add(self):
        item = Mock()
        self.items.append(item)
        return item


class MockLoadModelsRequest(object):
    """ mock a LoadModelsRequest protobuf """

    def __init__(self, *args, **kwargs):
        for (k, v) in kwargs.items():
            setattr(self, k, v)
        self.xprotos = ItemList()
        self.decls = ItemList()
        self.attics = ItemList()
        self.convenience_methods = ItemList()
        self.migrations = ItemList()


class TestLoadModels(unittest.TestCase):
    def setUp(self):
        self.sys_path_save = sys.path
        self.cwd_save = os.getcwd()

        config = os.path.join(test_path, "test_config.yaml")
        from xosconfig import Config

        Config.clear()
        Config.init(config, "synchronizer-config-schema.yaml")

        from xossynchronizer import loadmodels
        from xossynchronizer.loadmodels import ModelLoadClient
        self.loadmodels = loadmodels

        self.api = MagicMock()
        self.api.dynamicload_pb2.LoadModelsRequest = MockLoadModelsRequest
        self.loader = ModelLoadClient(self.api)

    def tearDown(self):
        sys.path = self.sys_path_save
        os.chdir(self.cwd_save)

    def test_upload_models(self):
        dir_map = {"models_dir": ["models.xproto", "models.py"],
                   "models_dir/convenience": ["convenience1.py"],
                   "models_dir/../migrations": ["migration1.py", "migration2.py"]}

        file_map = {"models_dir/models.xproto": u"some xproto",
                    "models_dir/models.py": u"print `python models file`",
                    "models_dir/convenience": u"directory",
                    "models_dir/convenience/convenience1.py": u"print `python convenience file`",
                    "models_dir/../migrations": u"directory",
                    "models_dir/../migrations/migration1.py": u"print `first migration`",
                    "models_dir/../migrations/migration2.py": u"print `second migration`"}

        orig_open = open
        with patch("os.listdir", side_effect=functools.partial(mock_listdir, dir_map)), \
                patch("os.path.exists", side_effect=functools.partial(mock_exists, file_map)), \
                patch("__builtin__.open", side_effect=functools.partial(mock_open, orig_open, file_map)):
            self.loader.upload_models("myservice", "models_dir", "1.2")

            request = self.api.dynamicload.LoadModels.call_args[0][0]
            self.assertEqual(request.name, "myservice")
            self.assertEqual(request.version, "1.2")

            self.assertEqual(len(request.xprotos.items), 1)
            self.assertEqual(request.xprotos.items[0].filename, "models.xproto")
            self.assertEqual(request.xprotos.items[0].contents, u"some xproto")

            self.assertEqual(len(request.decls.items), 1)
            self.assertEqual(request.decls.items[0].filename, "models.py")
            self.assertEqual(request.decls.items[0].contents, u"print `python models file`")

            self.assertEqual(len(request.convenience_methods.items), 1)
            self.assertEqual(request.convenience_methods.items[0].filename, "convenience1.py")
            self.assertEqual(request.convenience_methods.items[0].contents, u"print `python convenience file`")

            self.assertEqual(len(request.migrations.items), 2)
            self.assertEqual(request.migrations.items[0].filename, "migration1.py")
            self.assertEqual(request.migrations.items[0].contents, u"print `first migration`")
            self.assertEqual(request.migrations.items[1].filename, "migration2.py")
            self.assertEqual(request.migrations.items[1].contents, u"print `second migration`")


if __name__ == "__main__":
    unittest.main()