kea/src/lib/python/isc/config/tests/cfgmgr_test.py

# Copyright (C) 2010  Internet Systems Consortium.
#
# Permission to use, copy, modify, and distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
# DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
# FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

#
# Tests for the configuration manager module
#

import unittest
import os
from isc.config.cfgmgr import *
from isc.config import config_data
from unittest_fakesession import FakeModuleCCSession

class TestConfigManagerData(unittest.TestCase):
    def setUp(self):
        self.data_path = os.environ['CONFIG_TESTDATA_PATH']
        self.writable_data_path = os.environ['CONFIG_WR_TESTDATA_PATH']
        self.config_manager_data = ConfigManagerData(self.writable_data_path,
                                                     file_name="b10-config.db")
        self.assertTrue(self.config_manager_data)

    def test_abs_file(self):
        """
        Test what happens if we give the config manager an absolute path.
        It shouldn't append the data path to it.
        """
        abs_path = self.data_path + os.sep + "b10-config-imaginary.db"
        data = ConfigManagerData(self.data_path, abs_path)
        self.assertEqual(abs_path, data.db_filename)
        self.assertEqual(self.data_path, data.data_path)

    def test_init(self):
        self.assertEqual(self.config_manager_data.data['version'],
                         config_data.BIND10_CONFIG_DATA_VERSION)
        self.assertEqual(self.config_manager_data.data_path,
                         self.writable_data_path)
        self.assertEqual(self.config_manager_data.db_filename,
                         self.writable_data_path + os.sep + "b10-config.db")

    def test_read_from_file(self):
        ConfigManagerData.read_from_file(self.writable_data_path, "b10-config.db")
        self.assertRaises(ConfigManagerDataEmpty,
                          ConfigManagerData.read_from_file,
                          "doesnotexist", "b10-config.db")
        self.assertRaises(ConfigManagerDataReadError,
                          ConfigManagerData.read_from_file,
                          self.data_path, "b10-config-bad1.db")
        self.assertRaises(ConfigManagerDataReadError,
                          ConfigManagerData.read_from_file,
                          self.data_path, "b10-config-bad2.db")
        self.assertRaises(ConfigManagerDataReadError,
                          ConfigManagerData.read_from_file,
                          self.data_path, "b10-config-bad3.db")
        self.assertRaises(ConfigManagerDataReadError,
                          ConfigManagerData.read_from_file,
                          self.data_path, "b10-config-bad4.db")

    def test_write_to_file(self):
        output_file_name = "b10-config-write-test"
        self.config_manager_data.write_to_file(output_file_name)
        new_config = ConfigManagerData(self.data_path, output_file_name)
        self.assertEqual(self.config_manager_data, new_config)
        os.remove(output_file_name)

    def check_existence(self, files, should_exist=[], should_not_exist=[]):
        """Helper function for test_rename_config_file.
           Arguments:
           files: array of file names to check.
           should_exist: array of indices, the files in 'files' with these
                         indices should exist.
           should_not_exist: array of indices, the files in 'files' with
                             these indices should not exist."""
        for n in should_exist:
            self.assertTrue(os.path.exists(files[n]))
        for n in should_not_exist:
            self.assertFalse(os.path.exists(files[n]))

    def test_rename_config_file(self):
        # test file names, put in array for easy cleanup
        filenames = [ "b10-config-rename-test",
                      "b10-config-rename-test.bak",
                      "b10-config-rename-test.bak.1",
                      "b10-config-rename-test.bak.2" ]

        for filename in filenames:
            if os.path.exists(filename):
                os.remove(filename)

        # The original does not exist, so the new one should not be created
        self.config_manager_data.rename_config_file(filenames[0])
        self.check_existence(filenames, [], [0, 1, 2, 3])

        # now create a file to rename, and call rename again
        self.config_manager_data.write_to_file(filenames[0])
        self.config_manager_data.rename_config_file(filenames[0])
        self.check_existence(filenames, [1], [0, 2, 3])

        # If backup already exists, give it a new name automatically
        self.config_manager_data.write_to_file(filenames[0])
        self.config_manager_data.rename_config_file(filenames[0])
        self.check_existence(filenames, [1, 2], [0, 3])

        # If backup already exists, give it a new name automatically with
        # increasing postfix
        self.config_manager_data.write_to_file(filenames[0])
        self.config_manager_data.rename_config_file(filenames[0])
        self.check_existence(filenames, [1, 2, 3], [0])

        # Test with explicit renamed file argument
        self.config_manager_data.rename_config_file(filenames[1],
                                                    filenames[0])
        self.check_existence(filenames, [0, 2, 3], [1])

        # clean up again to be nice
        for filename in filenames:
            if os.path.exists(filename):
                os.remove(filename)

    def test_equality(self):
        # tests the __eq__ function. Equality is only defined
        # by equality of the .data element. If data_path or db_filename
        # are different, but the contents are the same, it's still
        # considered equal
        cfd1 = ConfigManagerData(self.data_path, file_name="b10-config.db")
        cfd2 = ConfigManagerData(self.data_path, file_name="b10-config.db")
        self.assertEqual(cfd1, cfd2)
        cfd2.data_path = "some/unknown/path"
        self.assertEqual(cfd1, cfd2)
        cfd2.db_filename = "bad_file.name"
        self.assertEqual(cfd1, cfd2)
        cfd2.data['test'] = { 'a': [ 1, 2, 3]}
        self.assertNotEqual(cfd1, cfd2)


class TestConfigManager(unittest.TestCase):

    def setUp(self):
        self.data_path = os.environ['CONFIG_TESTDATA_PATH']
        self.writable_data_path = os.environ['CONFIG_WR_TESTDATA_PATH']
        self.fake_session = FakeModuleCCSession()
        self.cm = ConfigManager(self.writable_data_path,
                                database_filename="b10-config.db",
                                session=self.fake_session)
        self.name = "TestModule"
        self.spec = isc.config.module_spec_from_file(self.data_path + os.sep + "/spec2.spec")

    def test_paths(self):
        """
        Test data_path and database filename is passed trough to
        underlying ConfigManagerData.
        """
        cm = ConfigManager("datapath", "filename", self.fake_session)
        self.assertEqual("datapath" + os.sep + "filename",
                         cm.config.db_filename)
        # It should preserve it while reading
        cm.read_config()
        self.assertEqual("datapath" + os.sep + "filename",
                         cm.config.db_filename)

    def test_init(self):
        self.assertEqual(self.cm.module_specs, {})
        self.assertEqual(self.cm.data_path, self.writable_data_path)
        self.assertIsNotNone(self.cm.config)
        self.assertTrue(self.fake_session.has_subscription("ConfigManager"))
        self.assertTrue(self.fake_session.has_subscription("Boss", "ConfigManager"))
        self.assertFalse(self.cm.running)

    def test_notify_boss(self):
        self.cm.notify_boss()
        msg = self.fake_session.get_message("Boss", None)
        self.assertTrue(msg)
        # this one is actually wrong, but 'current status quo'
        self.assertEqual(msg, {"running": "ConfigManager"})

    def test_set_module_spec(self):
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        self.assertNotIn(module_spec.get_module_name(), self.cm.virtual_modules)

    def test_remove_module_spec(self):
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.remove_module_spec(module_spec.get_module_name())
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.assertNotIn(module_spec.get_module_name(), self.cm.virtual_modules)

    def test_add_remove_virtual_module(self):
        module_spec = isc.config.module_spec.module_spec_from_file(
            self.data_path + os.sep + "spec1.spec")
        check_func = lambda: True
        # Make sure it's not in there before
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.assertNotIn(module_spec.get_module_name(), self.cm.virtual_modules)
        # Add it there
        self.cm.set_virtual_module(module_spec, check_func)
        # Check it's in there
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        self.assertEqual(self.cm.module_specs[module_spec.get_module_name()],
                      module_spec)
        self.assertEqual(self.cm.virtual_modules[module_spec.get_module_name()],
                      check_func)
        # Remove it again
        self.cm.remove_module_spec(module_spec.get_module_name())
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.assertNotIn(module_spec.get_module_name(), self.cm.virtual_modules)

    def test_get_module_spec(self):
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        module_spec2 = self.cm.get_module_spec(module_spec.get_module_name())
        self.assertEqual(module_spec.get_full_spec(), module_spec2)

        self.assertEqual({}, self.cm.get_module_spec("nosuchmodule"))

    def test_get_config_spec(self):
        config_spec = self.cm.get_config_spec()
        self.assertEqual(config_spec, {})
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        config_spec = self.cm.get_config_spec()
        self.assertEqual(config_spec, { 'Spec1': None })
        self.cm.remove_module_spec('Spec1')
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        config_spec = self.cm.get_config_spec()
        self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())
        config_spec = self.cm.get_config_spec('Spec2')
        self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())


    def test_get_commands_spec(self):
        commands_spec = self.cm.get_commands_spec()
        self.assertEqual(commands_spec, {})
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        commands_spec = self.cm.get_commands_spec()
        self.assertEqual(commands_spec, { 'Spec1': None })
        self.cm.remove_module_spec('Spec1')
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        commands_spec = self.cm.get_commands_spec()
        self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())
        commands_spec = self.cm.get_commands_spec('Spec2')
        self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())

    def test_get_statistics_spec(self):
        statistics_spec = self.cm.get_statistics_spec()
        self.assertEqual(statistics_spec, {})
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        statistics_spec = self.cm.get_statistics_spec()
        self.assertEqual(statistics_spec, { 'Spec1': None })
        self.cm.remove_module_spec('Spec1')
        module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
        self.assertNotIn(module_spec.get_module_name(), self.cm.module_specs)
        self.cm.set_module_spec(module_spec)
        self.assertIn(module_spec.get_module_name(), self.cm.module_specs)
        statistics_spec = self.cm.get_statistics_spec()
        self.assertEqual(statistics_spec['Spec2'], module_spec.get_statistics_spec())
        statistics_spec = self.cm.get_statistics_spec('Spec2')
        self.assertEqual(statistics_spec['Spec2'], module_spec.get_statistics_spec())

    def test_read_config(self):
        self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION})
        self.cm.data_path = "/no_such_path"
        self.cm.read_config()
        self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION})

    def test_write_config(self):
        # tested in ConfigManagerData tests
        pass

    def _handle_msg_helper(self, msg, expected_answer):
        answer = self.cm.handle_msg(msg)
        self.assertEqual(expected_answer, answer)

    def test_handle_msg_basic_commands(self):
        # Some basic commands, where not much interaction happens, just
        # check the result
        self._handle_msg_helper({},
            { 'result': [ 1, 'Unknown message format: {}']})
        self._handle_msg_helper("",
            { 'result': [ 1, 'Unknown message format: ']})
        self._handle_msg_helper({ "command": [ "badcommand" ] },
            { 'result': [ 1, "Unknown command: badcommand"]})
        self._handle_msg_helper({ "command": [ "get_commands_spec" ] },
                                { 'result': [ 0, {} ]})
        self._handle_msg_helper({ "command": [ "get_statistics_spec" ] },
                                { 'result': [ 0, {} ]})
        self._handle_msg_helper({ "command": [ "get_module_spec" ] },
                                { 'result': [ 0, {} ]})
        self._handle_msg_helper({ "command": [ "get_module_spec",
                                               { "module_name": "Spec2" } ] },
                                { 'result': [ 0, {} ]})
        self._handle_msg_helper({ "command": [ "get_module_spec", 1 ] },
                                {'result': [1, 'Bad get_module_spec command, '+
                                               'argument not a dict']})
        self._handle_msg_helper({ "command": [ "get_module_spec", { } ] },
                                {'result': [1, 'Bad module_name in '+
                                               'get_module_spec command']})
        self._handle_msg_helper({ "command": [ "get_config" ] },
                                { 'result': [ 0, { 'version':
                                    config_data.BIND10_CONFIG_DATA_VERSION }]})
        self._handle_msg_helper({ "command": [ "get_config",
                                    { "module_name": "nosuchmodule" } ] },
                                {'result': [0, { 'version':
                                    config_data.BIND10_CONFIG_DATA_VERSION }]})
        self._handle_msg_helper({ "command": [ "get_config", 1 ] },
                                {'result': [1, 'Bad get_config command, '+
                                               'argument not a dict']})
        self._handle_msg_helper({ "command": [ "get_config", { } ] },
                                {'result': [1, 'Bad module_name in '+
                                               'get_config command']})
        self._handle_msg_helper({ "command": [ "set_config" ] },
                                {'result': [1, 'Wrong number of arguments']})
        self._handle_msg_helper({ "command": [ "set_config", [{}]] },
                                {'result': [0]})

        self.assertEqual(len(self.fake_session.message_queue), 0)

    def test_handle_msg_module_and_stats_commands(self):
        self._handle_msg_helper({ "command":
                                  ["module_spec", self.spec.get_full_spec()]
                                },
                                {'result': [0]})
        # There should be a message on the queue about the 'new' Spec2 module
        # from ConfigManager to Cmdctl, containing its name and full
        # specification
        self.assertEqual(ccsession.create_command(
                            ccsession.COMMAND_MODULE_SPECIFICATION_UPDATE,
                            [ self.spec.get_module_name(),
                              self.spec.get_full_spec()]),
                         self.fake_session.get_message("Cmdctl", None))

        self._handle_msg_helper({ "command": [ "module_spec", { 'foo': 1 } ] },
                                {'result': [1, 'Error in data definition: no '+
                                               'module_name in module_spec']})

        self._handle_msg_helper({ "command": [ "get_module_spec" ] },
                                { 'result': [ 0, { self.spec.get_module_name():
                                                 self.spec.get_full_spec() } ]})
        self._handle_msg_helper({ "command": [ "get_module_spec",
                                               { "module_name" : "Spec2" } ] },
                                { 'result': [ 0, self.spec.get_full_spec() ] })
        self._handle_msg_helper({ "command": [ "get_commands_spec" ] },
                                { 'result': [ 0, { self.spec.get_module_name():
                                              self.spec.get_commands_spec()}]})
        self._handle_msg_helper({ "command": [ "get_statistics_spec" ] },
                                { 'result': [ 0, { self.spec.get_module_name():
                                             self.spec.get_statistics_spec()}]})


    def __test_handle_msg_update_config_helper(self, new_config):
        # Helper function for the common pattern in
        # test_handle_msg_update_config; send 'set config', check for
        # update message, check if config has indeed been updated

        my_ok_answer = { 'result': [ 0 ] }
        # Send the 'ok' that cfgmgr expects back to the fake queue first
        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")

        config_version = config_data.BIND10_CONFIG_DATA_VERSION
        self._handle_msg_helper({ "command": [ "set_config",
                                               [ { "version": config_version,
                                                 self.name: new_config } ] ] },
                                my_ok_answer)

        # The cfgmgr should have eaten the ok message, and sent out an update
        # message
        self.assertEqual(len(self.fake_session.message_queue), 1)
        self.assertEqual({'command': [ 'config_update', new_config]},
                         self.fake_session.get_message(self.name, None))

        # Config should have been updated
        self.assertEqual(self.cm.config.data, {self.name: new_config,
                            'version': config_version})

        # and the queue should now be empty again
        self.assertEqual(len(self.fake_session.message_queue), 0)

    def test_handle_msg_update_config(self):
        # Update the configuration and check results a few times
        # only work the first time
        self.__test_handle_msg_update_config_helper({ "test": 123 })

        self.__test_handle_msg_update_config_helper({ "test": 124 })

        self.__test_handle_msg_update_config_helper({ "test": 125 })

        self.__test_handle_msg_update_config_helper({ "test": 126 })

        # Now send an error result (i.e. config not accepted)
        my_bad_answer = { 'result': [1, "bad config"] }
        self.fake_session.group_sendmsg(my_bad_answer, "ConfigManager")
        self._handle_msg_helper({ "command": [ "set_config",
                                               [self.name, { "test": 127 }] ] },
                                my_bad_answer )
        self.assertEqual(len(self.fake_session.message_queue), 1)
        self.assertEqual({'command': [ 'config_update', {'test': 127}]},
                         self.fake_session.get_message(self.name, None))
        # Config should not be updated due to the error
        self.cm.read_config()
        self.assertEqual(self.cm.config.data, { self.name: {'test': 126},
                            'version': config_data.BIND10_CONFIG_DATA_VERSION})

        self.assertEqual(len(self.fake_session.message_queue), 0)

        self.fake_session.group_sendmsg(None, 'ConfigManager')
        self._handle_msg_helper({ "command": [ "set_config", [ ] ] },
                                {'result': [1, 'Wrong number of arguments']} )
        self._handle_msg_helper({ "command": [ "set_config",
                                               [ self.name, { "test": 128 }]]},
                                { 'result': [1, 'No answer message '+
                                                'from TestModule']} )

        # This command should leave a message to the TestModule to update its
        # configuration (since the TestModule did not eat it)
        self.assertEqual(len(self.fake_session.message_queue), 1)
        self.assertEqual(
            ccsession.create_command(ccsession.COMMAND_CONFIG_UPDATE,
                                     { "test": 128 }),
            self.fake_session.get_message("TestModule", None))

        # Make sure queue is empty now
        self.assertEqual(len(self.fake_session.message_queue), 0)

        # Shutdown should result in 'ok' answer
        self._handle_msg_helper({ "command":
                                  ["shutdown"]
                                },
                                {'result': [0]})

    def test_stopping_message(self):
        # Update the system by announcing this module
        self._handle_msg_helper({ "command":
                                  ["module_spec", self.spec.get_full_spec()]
                                },
                                {'result': [0]})

        # This causes a update to be sent from the ConfigManager to the CmdCtl
        # channel, containing the new module's name and full specification
        self.assertEqual(ccsession.create_command(
                            ccsession.COMMAND_MODULE_SPECIFICATION_UPDATE,
                            [ self.spec.get_module_name(),
                              self.spec.get_full_spec()]),
                         self.fake_session.get_message("Cmdctl", None))

        # A stopping message should get no response, but should cause another
        # message to be sent, if it is a known module
        self._handle_msg_helper({ "command": [ "stopping",
                                               { "module_name": "Spec2"}] },
                                None)
        self.assertEqual(len(self.fake_session.message_queue), 1)
        self.assertEqual({'command': [ 'module_specification_update',
                                       ['Spec2', None] ] },
                         self.fake_session.get_message("Cmdctl", None))

        # but if the 'stopping' module is either unknown or not running,
        # no followup message should be sent
        self._handle_msg_helper({ "command":
                                  [ "stopping",
                                    { "module_name": "NoSuchModule" } ] },
                                None)
        self.assertEqual(len(self.fake_session.message_queue), 0)

    def test_set_config_virtual(self):
        """Test that if the module is virtual, we don't send it over the
           message bus, but call the checking function.
           """
        # We run the same three times, with different return values
        def single_test(value, returnFunc, expectedResult):
            # Because closures can't assign to closed-in variables, we pass
            # it trough self
            self.called_with = None
            def check_test(new_data):
                self.called_with = new_data
                return returnFunc()

            # Register our virtual module
            self.cm.set_virtual_module(self.spec, check_test)
            # The fake session will throw now if it tries to read a response.
            # Handy, we don't need to find a complicated way to check for it.
            result = self.cm.handle_msg(ccsession.create_command(
                        ccsession.COMMAND_SET_CONFIG,
                        [self.spec.get_module_name(), { "item1": value }]))
            # Check the correct result is passed and our function was called
            # With correct data
            self.assertEqual(self.called_with['item1'], value)
            self.assertEqual(result, {'result': expectedResult})
            if expectedResult[0] == 0:
                # Check it provided the correct notification
                self.assertEqual(len(self.fake_session.message_queue), 1)
                self.assertEqual({'command': [ 'config_update',
                                 {'item1': value}]},
                                 self.fake_session.get_message('Spec2', None))
                # and the queue should now be empty again
                self.assertEqual(len(self.fake_session.message_queue), 0)
            else:
                # It shouldn't send anything on error
                self.assertEqual(len(self.fake_session.message_queue), 0)

        # Success
        single_test(5, lambda: None, [0])
        # Graceful error
        single_test(6, lambda: "Just error", [1, "Just error"])
        # Exception from the checker
        def raiser():
            raise Exception("Just exception")
        single_test(7, raiser, [1, "Exception: Just exception"])

    def test_set_config_all(self):
        my_ok_answer = { 'result': [ 0 ] }

        self.assertEqual({"version": 2}, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value1": 123 }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value1": 123 }
                         }, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value1": 124 }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value1": 124 }
                         }, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value2": True }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value1": 124,
                                    "value2": True
                                  }
                         }, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value3": [ 1, 2, 3 ] }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value1": 124,
                                    "value2": True,
                                    "value3": [ 1, 2, 3 ]
                                  }
                         }, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value2": False }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value1": 124,
                                    "value2": False,
                                    "value3": [ 1, 2, 3 ]
                                  }
                         }, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value1": None }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value2": False,
                                    "value3": [ 1, 2, 3 ]
                                  }
                         }, self.cm.config.data)

        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
        self.cm.handle_msg(ccsession.create_command(
            ccsession.COMMAND_SET_CONFIG, ["test", { "value3": [ 1 ] }]))
        self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
                          "test": { "value2": False,
                                    "value3": [ 1 ]
                                  }
                         }, self.cm.config.data)


    def test_run(self):
        self.fake_session.group_sendmsg({ "command": [ "get_commands_spec" ] }, "ConfigManager")
        self.fake_session.group_sendmsg({ "command": [ "get_statistics_spec" ] }, "ConfigManager")
        self.fake_session.group_sendmsg({ "command": [ "stopping", { "module_name": "FooModule" } ] }, "ConfigManager")
        self.fake_session.group_sendmsg({ "command": [ "shutdown" ] }, "ConfigManager")
        self.assertEqual(len(self.fake_session.message_queue), 4)
        self.cm.run()
        # All commands should have been read out by run()
        # Three of the commands should have been responded to, so the queue
        # should now contain three answers
        self.assertEqual(len(self.fake_session.message_queue), 3)


if __name__ == '__main__':
    if not 'CONFIG_TESTDATA_PATH' in os.environ or not 'CONFIG_WR_TESTDATA_PATH' in os.environ:
        print("You need to set the environment variable CONFIG_TESTDATA_PATH and CONFIG_WR_TESTDATA_PATH to point to the directory containing the test data files")
        exit(1)
    isc.log.init("unittests")
    isc.log.resetUnitTestRootLogger()
    unittest.main()