module, and one to update the configuration run-time. These
callbacks are called when 'check_command' is called on the
ModuleCCSession"""
-
+
def __init__(self, spec_file_name, config_handler, command_handler,
cc_session=None, handle_logging_config=True,
socket_file = None):
"""
module_spec = isc.config.module_spec_from_file(spec_file_name)
ConfigData.__init__(self, module_spec)
-
+
self._module_name = module_spec.get_module_name()
-
+
self.set_config_handler(config_handler)
self.set_command_handler(command_handler)
returns nothing.
It calls check_command_without_recvmsg()
to parse the received message.
-
+
If nonblock is True, it just checks if there's a command
and does nothing if there isn't. If nonblock is False, it
waits until it arrives. It temporarily sets timeout to infinity,
"""Parse the given message to see if there is a command or a
configuration update. Calls the corresponding handler
functions if present. Responds on the channel if the
- handler returns a message."""
+ handler returns a message."""
# should we default to an answer? success-by-default? unhandled error?
if msg is not None and not 'result' in msg:
answer = None
answer = create_answer(1, str(exc))
if answer:
self._session.group_reply(env, answer)
-
+
def set_config_handler(self, config_handler):
"""Set the config handler for this module. The handler is a
function that takes the full configuration and handles it.
if not cur_list:
cur_list = []
- if value is None:
+ if value is None and "list_item_spec" in module_spec:
if "item_default" in module_spec["list_item_spec"]:
value = module_spec["list_item_spec"]["item_default"]
if module_spec is None:
raise isc.cc.data.DataNotFoundError("Unknown item " + str(identifier))
+ # for type any, we determine the 'type' by what value is set
+ # (which would be either list or dict)
+ cur_value, _ = self.get_value(identifier)
+ type_any = module_spec['item_type'] == 'any'
+
# the specified element must be a list or a named_set
- if 'list_item_spec' in module_spec:
+ if 'list_item_spec' in module_spec or\
+ (type_any and type(cur_value) == list):
value = None
# in lists, we might get the value with spaces, making it
# the third argument. In that case we interpret both as
value_str += set_value_str
value = isc.cc.data.parse_value_str(value_str)
self._add_value_to_list(identifier, value, module_spec)
- elif 'named_set_item_spec' in module_spec:
+ elif 'named_set_item_spec' in module_spec or\
+ (type_any and type(cur_value) == dict):
item_name = None
item_value = None
if value_str is not None:
- item_name = isc.cc.data.parse_value_str(value_str)
+ #item_name = isc.cc.data.parse_value_str(value_str)
+ item_name = value_str
if set_value_str is not None:
item_value = isc.cc.data.parse_value_str(set_value_str)
else:
if value_str is not None:
value = isc.cc.data.parse_value_str(value_str)
- if 'list_item_spec' in module_spec:
- if value is not None:
+ # for type any, we determine the 'type' by what value is set
+ # (which would be either list or dict)
+ cur_value, _ = self.get_value(identifier)
+ type_any = module_spec['item_type'] == 'any'
+
+ # there's two forms of 'remove from list'; the remove-value-from-list
+ # form, and the 'remove-by-index' form. We can recognize the second
+ # case by value is None
+ if 'list_item_spec' in module_spec or\
+ (type_any and type(cur_value) == list) or\
+ value is None:
+ if not type_any and value is not None:
isc.config.config_data.check_type(module_spec['list_item_spec'], value)
self._remove_value_from_list(identifier, value)
- elif 'named_set_item_spec' in module_spec:
- self._remove_value_from_named_set(identifier, value)
+ elif 'named_set_item_spec' in module_spec or\
+ (type_any and type(cur_value) == dict):
+ self._remove_value_from_named_set(identifier, value_str)
else:
- raise isc.cc.data.DataNotFoundError(str(identifier) + " is not a list or a named_set")
+ raise isc.cc.data.DataNotFoundError(str(identifier) + " is not a list or a named_set " + str(module_spec) + " and type " + str(type(cur_value)))
self.assertRaises(ModuleCCSessionError, parse_answer, { 'result': [] })
self.assertRaises(ModuleCCSessionError, parse_answer, { 'result': [ 'not_an_rcode' ] })
self.assertRaises(ModuleCCSessionError, parse_answer, { 'result': [ 1, 2 ] })
-
+
rcode, val = parse_answer({ 'result': [ 0 ] })
self.assertEqual(0, rcode)
self.assertEqual(None, val)
def spec_file(self, file):
return self.data_path + os.sep + file
-
+
def create_session(self, spec_file_name, config_handler = None,
command_handler = None, cc_session = None):
return ModuleCCSession(self.spec_file(spec_file_name),
self.assertEqual(len(fake_session.message_queue), 1)
self.assertEqual({'result': [1, 'No config_data specification']},
fake_session.get_message('Spec1', None))
-
+
def test_check_command3(self):
fake_session = FakeModuleCCSession()
mccs = self.create_session("spec2.spec", None, None, fake_session)
self.assertEqual(len(fake_session.message_queue), 1)
self.assertEqual({'result': [0]},
fake_session.get_message('Spec2', None))
-
+
def test_check_command4(self):
fake_session = FakeModuleCCSession()
mccs = self.create_session("spec2.spec", None, None, fake_session)
self.assertEqual(len(fake_session.message_queue), 1)
self.assertEqual({'result': [1, 'aaa should be an integer']},
fake_session.get_message('Spec2', None))
-
+
def test_check_command5(self):
fake_session = FakeModuleCCSession()
mccs = self.create_session("spec2.spec", None, None, fake_session)
self.assertEqual(len(fake_session.message_queue), 1)
self.assertEqual({'result': [1, 'aaa should be an integer']},
fake_session.get_message('Spec2', None))
-
+
def test_check_command6(self):
fake_session = FakeModuleCCSession()
mccs = self.create_session("spec2.spec", None, None, fake_session)
self.assertEqual(len(fake_session.message_queue), 1)
self.assertEqual({'result': [1, 'No config_data specification']},
fake_session.get_message('Spec1', None))
-
+
def test_check_command_without_recvmsg2(self):
"copied from test_check_command3"
fake_session = FakeModuleCCSession()
self.assertEqual(len(fake_session.message_queue), 1)
self.assertEqual({'result': [0]},
fake_session.get_message('Spec2', None))
-
+
def test_check_command_without_recvmsg3(self):
"copied from test_check_command7"
fake_session = FakeModuleCCSession()
mccs.check_command_without_recvmsg(cmd, env)
self.assertEqual({'result': [0]},
fake_session.get_message('Spec2', None))
-
+
def test_check_command_block_timeout(self):
"""Check it works if session has timeout and it sets it back."""
def cmd_check(mccs, session):
def set_get_answer(self, name, answer):
self.get_answers[name] = answer
-
+
def set_post_answer(self, name, answer):
self.post_answers[name] = answer
-
+
def send_GET(self, name, arg = None):
if name in self.get_answers:
return self.get_answers[name]
else:
return {}
-
+
def send_POST(self, name, arg = None):
if name in self.post_answers:
return self.post_answers[name]
else:
return fakeAnswer()
-
+
class TestUIModuleCCSession(unittest.TestCase):
def setUp(self):
def spec_file(self, file):
return self.data_path + os.sep + file
-
- def create_uccs2(self, fake_conn):
- module_spec = isc.config.module_spec_from_file(self.spec_file("spec2.spec"))
+
+ def create_uccs(self, fake_conn, specfile="spec2.spec"):
+ module_spec = isc.config.module_spec_from_file(self.spec_file(specfile))
fake_conn.set_get_answer('/module_spec', { module_spec.get_module_name(): module_spec.get_full_spec()})
fake_conn.set_get_answer('/config_data', { 'version': BIND10_CONFIG_DATA_VERSION })
return UIModuleCCSession(fake_conn)
def test_add_remove_value(self):
fake_conn = fakeUIConn()
- uccs = self.create_uccs2(fake_conn)
+ uccs = self.create_uccs(fake_conn)
self.assertRaises(isc.cc.data.DataNotFoundError, uccs.add_value, 1, "a")
self.assertRaises(isc.cc.data.DataNotFoundError, uccs.add_value, "no_such_item", "a")
self.assertRaises(isc.cc.data.DataTypeError,
uccs.remove_value, "Spec2/item5", None)
+ # Test adding and removing values for type = any
+ def test_add_remove_value_any(self):
+ fake_conn = fakeUIConn()
+ uccs = self.create_uccs(fake_conn, "spec40.spec")
+
+ # Test item set of basic types
+ items = [ 1234, "foo", True, False ]
+ items_as_str = [ '1234', 'foo', 'true', 'false' ]
+
+ def test_fails():
+ self.assertRaises(isc.cc.data.DataNotFoundError, uccs.add_value, "Spec40/item1", "foo")
+ self.assertRaises(isc.cc.data.DataNotFoundError, uccs.add_value, "Spec40/item1", "foo", "bar")
+ self.assertRaises(isc.cc.data.DataNotFoundError, uccs.remove_value, "Spec40/item1", "foo")
+ self.assertRaises(isc.cc.data.DataTypeError, uccs.remove_value, "Spec40/item1[0]", None)
+
+ # A few helper functions to perform a number of tests
+ # (to repeat the same test for nested data)
+ def check_list(identifier):
+ for item in items_as_str:
+ uccs.add_value(identifier, item)
+ self.assertEqual((items, 1), uccs.get_value(identifier))
+
+ # Removing from list should work in both ways
+ uccs.remove_value(identifier, "foo")
+ uccs.remove_value(identifier + "[1]", None)
+ self.assertEqual(([1234, False], 1), uccs.get_value(identifier))
+
+ # As should item indexing
+ self.assertEqual((1234, 1), uccs.get_value(identifier + "[0]"))
+ self.assertEqual((False, 1), uccs.get_value(identifier + "[1]"))
+
+ def check_named_set(identifier):
+ for item in items_as_str:
+ # use string version as key as well
+ uccs.add_value(identifier, item, item)
+
+ self.assertEqual((1234, 1), uccs.get_value(identifier + "/1234"))
+ self.assertEqual((True, 1), uccs.get_value(identifier + "/true"))
+
+ for item in items_as_str:
+ # use string version as key as well
+ uccs.remove_value(identifier, item)
+
+
+ # should fail when set to value of primitive type
+ for item in items:
+ uccs.set_value("Spec40/item1", item)
+ test_fails()
+
+ # When set to list, add and remove should work, and its elements
+ # should be considered of type 'any' themselves.
+ uccs.set_value("Spec40/item1", [])
+ check_list("Spec40/item1")
+
+ # When set to dict, it should have the behaviour of a named set
+ uccs.set_value("Spec40/item1", {})
+ check_named_set("Spec40/item1")
+
+ # And, or course, we may need nesting.
+ uccs.set_value("Spec40/item1", { "foo": {}, "bar": [] })
+ check_named_set("Spec40/item1/foo")
+ check_list("Spec40/item1/bar")
+ uccs.set_value("Spec40/item1", [ {}, [] ] )
+ check_named_set("Spec40/item1[0]")
+ check_list("Spec40/item1[1]")
+ uccs.set_value("Spec40/item1", [[[[[[]]]]]] )
+ check_list("Spec40/item1[0][0][0][0][0]")
+ uccs.set_value("Spec40/item1", { 'a': { 'a': { 'a': {} } } } )
+ check_named_set("Spec40/item1/a/a/a")
+
def test_add_dup_value(self):
fake_conn = fakeUIConn()
uccs = self.create_uccs_listtest(fake_conn)
def test_commit(self):
fake_conn = fakeUIConn()
- uccs = self.create_uccs2(fake_conn)
+ uccs = self.create_uccs(fake_conn)
uccs.commit()
uccs._local_changes = {'Spec2': {'item5': [ 'a' ]}}
uccs.commit()
projects / thirdparty / kea.git / commitdiff
