# Copyright (C) 2009 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. import unittest import isc.cc.data import os import io import sys import socket import http.client import pwd import getpass from optparse import OptionParser from isc.config.config_data import ConfigData, MultiConfigData from isc.config.module_spec import ModuleSpec from isc.testutils.parse_args import TestOptParser, OptsError from bindctl_main import set_bindctl_options from bindctl import cmdparse from bindctl import bindcmd from bindctl.moduleinfo import * from bindctl.exception import * try: from collections import OrderedDict except ImportError: from mycollections import OrderedDict class TestCmdLex(unittest.TestCase): def my_assert_raise(self, exception_type, cmd_line): self.assertRaises(exception_type, cmdparse.BindCmdParser, cmd_line) def testCommandWithoutParameter(self): cmd_parser = cmdparse.BindCmdParser("zone add") assert cmd_parser.module == "zone" assert cmd_parser.command == "add" self.assertEqual(len(cmd_parser.params), 0) def testCommandWithParameters(self): lines = {"zone add zone_name = cnnic.cn, file = cnnic.cn.file master=1.1.1.1", "zone add zone_name = \"cnnic.cn\", file ='cnnic.cn.file' master=1.1.1.1 ", "zone add zone_name = 'cnnic.cn\", file ='cnnic.cn.file' master=1.1.1.1, " } for cmd_line in lines: cmd_parser = cmdparse.BindCmdParser(cmd_line) assert cmd_parser.module == "zone" assert cmd_parser.command == "add" assert cmd_parser.params["zone_name"] == "cnnic.cn" assert cmd_parser.params["file"] == "cnnic.cn.file" assert cmd_parser.params["master"] == '1.1.1.1' def testCommandWithParamters_2(self): '''Test whether the parameters in key=value can be parsed properly.''' cmd_parser = cmdparse.BindCmdParser('zone cmd name = 1:34::2') self.assertEqual(cmd_parser.params['name'], '1:34::2') cmd_parser = cmdparse.BindCmdParser('zone cmd name = 1\"\'34**&2' ' value=44\"\'\"') self.assertEqual(cmd_parser.params['name'], '1\"\'34**&2') self.assertEqual(cmd_parser.params['value'], '44\"\'\"') cmd_parser = cmdparse.BindCmdParser('zone cmd name = 1\"\'34**&2' ',value= 44\"\'\"') self.assertEqual(cmd_parser.params['name'], '1\"\'34**&2') self.assertEqual(cmd_parser.params['value'], '44\"\'\"') cmd_parser = cmdparse.BindCmdParser('zone cmd name = 1\'34**&2' 'value=44\"\'\" value = ' '\"==============\'') self.assertEqual(cmd_parser.params['name'], '1\'34**&2value=44\"\'\"') self.assertEqual(cmd_parser.params['value'], '==============') cmd_parser = cmdparse.BindCmdParser('zone cmd name = \"1234, ' '567890 \" value ==&*/') self.assertEqual(cmd_parser.params['name'], '1234, 567890 ') self.assertEqual(cmd_parser.params['value'], '=&*/') def testCommandWithListParam(self): cmd_parser = cmdparse.BindCmdParser("zone set zone_name='cnnic.cn', " "master='1.1.1.1, 2.2.2.2'") assert cmd_parser.params["master"] == '1.1.1.1, 2.2.2.2' def testCommandWithHelpParam(self): cmd_parser = cmdparse.BindCmdParser("zone add help") assert cmd_parser.params["help"] == "help" cmd_parser = cmdparse.BindCmdParser("zone add help *&)&)*&&$#$^%") assert cmd_parser.params["help"] == "help" self.assertEqual(len(cmd_parser.params), 1) def testCmdModuleNameFormatError(self): self.my_assert_raise(CmdModuleNameFormatError, "zone=good") self.my_assert_raise(CmdModuleNameFormatError, "zo/ne") self.my_assert_raise(CmdModuleNameFormatError, "") self.my_assert_raise(CmdModuleNameFormatError, "=zone") self.my_assert_raise(CmdModuleNameFormatError, "zone,") def testCmdMissCommandNameFormatError(self): self.my_assert_raise(CmdMissCommandNameFormatError, "zone") self.my_assert_raise(CmdMissCommandNameFormatError, "zone ") self.my_assert_raise(CmdMissCommandNameFormatError, "help ") def testCmdCommandNameFormatError(self): self.my_assert_raise(CmdCommandNameFormatError, "zone =d") self.my_assert_raise(CmdCommandNameFormatError, "zone z=d") self.my_assert_raise(CmdCommandNameFormatError, "zone z-d ") self.my_assert_raise(CmdCommandNameFormatError, "zone zdd/") self.my_assert_raise(CmdCommandNameFormatError, "zone zdd/ \"") class TestCmdSyntax(unittest.TestCase): def _create_bindcmd(self): """Create one bindcmd""" tool = bindcmd.BindCmdInterpreter() string_spec = { 'item_type' : 'string', 'item_optional' : False, 'item_default' : ''} int_spec = { 'item_type' : 'integer', 'item_optional' : False, 'item_default' : 10} zone_file_param = ParamInfo(name = "zone_file", param_spec = string_spec) zone_name = ParamInfo(name = 'zone_name', param_spec = string_spec) load_cmd = CommandInfo(name = "load") load_cmd.add_param(zone_file_param) load_cmd.add_param(zone_name) param_master = ParamInfo(name = "master", optional = True, param_spec = string_spec) param_master = ParamInfo(name = "port", optional = True, param_spec = int_spec) param_allow_update = ParamInfo(name = "allow_update", optional = True, param_spec = string_spec) set_cmd = CommandInfo(name = "set") set_cmd.add_param(param_master) set_cmd.add_param(param_allow_update) set_cmd.add_param(zone_name) reload_all_cmd = CommandInfo(name = "reload_all") zone_module = ModuleInfo(name = "zone") zone_module.add_command(load_cmd) zone_module.add_command(set_cmd) zone_module.add_command(reload_all_cmd) tool.add_module_info(zone_module) return tool def setUp(self): self.bindcmd = self._create_bindcmd() def no_assert_raise(self, cmd_line): cmd_parser = cmdparse.BindCmdParser(cmd_line) self.bindcmd._validate_cmd(cmd_parser) def my_assert_raise(self, exception_type, cmd_line): cmd_parser = cmdparse.BindCmdParser(cmd_line) self.assertRaises(exception_type, self.bindcmd._validate_cmd, cmd_parser) def testValidateSuccess(self): self.no_assert_raise("zone load zone_file='cn' zone_name='cn'") self.no_assert_raise("zone load zone_file='cn', zone_name='cn', ") self.no_assert_raise("zone help ") self.no_assert_raise("zone load help ") self.no_assert_raise("zone help help='dd' ") self.no_assert_raise("zone set allow_update='1.1.1.1' zone_name='cn'") self.no_assert_raise("zone set zone_name='cn'") self.my_assert_raise(isc.cc.data.DataTypeError, "zone set zone_name ='cn', port='cn'") self.no_assert_raise("zone reload_all") def testCmdUnknownModuleSyntaxError(self): self.my_assert_raise(CmdUnknownModuleSyntaxError, "zoned d") self.my_assert_raise(CmdUnknownModuleSyntaxError, "dd dd ") def testCmdUnknownCmdSyntaxError(self): self.my_assert_raise(CmdUnknownCmdSyntaxError, "zone dd") def testCmdMissParamSyntaxError(self): self.my_assert_raise(CmdMissParamSyntaxError, "zone load zone_file='cn'") self.my_assert_raise(CmdMissParamSyntaxError, "zone load zone_name='cn'") self.my_assert_raise(CmdMissParamSyntaxError, "zone set allow_update='1.1.1.1'") self.my_assert_raise(CmdMissParamSyntaxError, "zone set ") def testCmdUnknownParamSyntaxError(self): self.my_assert_raise(CmdUnknownParamSyntaxError, "zone load zone_d='cn'") self.my_assert_raise(CmdUnknownParamSyntaxError, "zone reload_all zone_name = 'cn'") self.my_assert_raise(CmdUnknownParamSyntaxError, "zone help a b c") class TestModuleInfo(unittest.TestCase): def test_get_param_name_by_position(self): cmd = CommandInfo('command') cmd.add_param(ParamInfo('name')) cmd.add_param(ParamInfo('age')) cmd.add_param(ParamInfo('data', optional = True)) cmd.add_param(ParamInfo('sex')) self.assertEqual('name', cmd.get_param_name_by_position(0, 2)) self.assertEqual('age', cmd.get_param_name_by_position(1, 2)) self.assertEqual('sex', cmd.get_param_name_by_position(2, 3)) self.assertEqual('data', cmd.get_param_name_by_position(2, 4)) self.assertEqual('data', cmd.get_param_name_by_position(2, 4)) self.assertRaises(KeyError, cmd.get_param_name_by_position, 4, 4) class TestNameSequence(unittest.TestCase): """ Test if the module/command/parameters is saved in the order creation """ def _create_bindcmd(self): """Create one bindcmd""" self._cmd = CommandInfo(name = "load") self.module = ModuleInfo(name = "zone") self.tool = bindcmd.BindCmdInterpreter() for random_str in self.random_names: self._cmd.add_param(ParamInfo(name = random_str)) self.module.add_command(CommandInfo(name = random_str)) self.tool.add_module_info(ModuleInfo(name = random_str)) def setUp(self): self.random_names = ['1erdfeDDWsd', '3fe', '2009erd', 'Fe231', 'tere142', 'rei8WD'] self._create_bindcmd() def testSequence(self): param_names = self._cmd.get_param_names() cmd_names = self.module.get_command_names() module_names = self.tool.get_module_names() i = 0 while i < len(self.random_names): assert self.random_names[i] == param_names[i+1] assert self.random_names[i] == cmd_names[i+1] assert self.random_names[i] == module_names[i+1] i = i + 1 # tine class to fake a UIModuleCCSession, but only the config data # parts for the next set of tests class FakeCCSession(MultiConfigData): def __init__(self): self._local_changes = {} self._current_config = {} self._specifications = {} self.add_foo_spec() def add_foo_spec(self): spec = { "module_name": "foo", "config_data": [ { "item_name": "an_int", "item_type": "integer", "item_optional": False, "item_default": 1 }, { "item_name": "a_list", "item_type": "list", "item_optional": False, "item_default": [], "list_item_spec": { "item_name": "a_string", "item_type": "string", "item_optional": False, "item_default": "bar" } } ] } self.set_specification(ModuleSpec(spec)) # fake socket class FakeSocket(): def __init__(self): self.run = True def connect(self, to): if not self.run: raise socket.error def close(self): self.run = False def send(self, data): if not self.run: raise socket.error return len(data) def makefile(self, type): return self def sendall(self, data): if not self.run: raise socket.error return len(data) class TestConfigCommands(unittest.TestCase): def setUp(self): self.tool = bindcmd.BindCmdInterpreter() mod_info = ModuleInfo(name = "foo") self.tool.add_module_info(mod_info) self.tool.config_data = FakeCCSession() self.stdout_backup = sys.stdout def test_precmd(self): def update_all_modules_info(): raise socket.error def precmd(line): self.tool.precmd(line) self.tool._update_all_modules_info = update_all_modules_info # If line is equals to 'EOF', _update_all_modules_info() # shouldn't be called precmd('EOF') self.assertRaises(socket.error, precmd, 'continue') def test_run(self): def login_to_cmdctl(): return True def cmd_loop(): self.tool._send_message("/module_spec", None) self.tool.login_to_cmdctl = login_to_cmdctl # rewrite cmdloop() to avoid interactive mode self.tool.cmdloop = cmd_loop self.tool.conn.sock = FakeSocket() self.tool.conn.sock.close() # validate log message for socket.err socket_err_output = io.StringIO() sys.stdout = socket_err_output self.assertEqual(1, self.tool.run()) self.assertEqual("Failed to send request, the connection is closed\n", socket_err_output.getvalue()) socket_err_output.close() # validate log message for http.client.CannotSendRequest cannot_send_output = io.StringIO() sys.stdout = cannot_send_output self.assertEqual(1, self.tool.run()) self.assertEqual("Can not send request, the connection is busy\n", cannot_send_output.getvalue()) cannot_send_output.close() def test_apply_cfg_command_int(self): self.tool.location = '/' self.assertEqual((1, MultiConfigData.DEFAULT), self.tool.config_data.get_value("/foo/an_int")) cmd_parser = cmdparse.BindCmdParser('config set identifier=' '"foo/an_int" value="5"') self.tool.apply_config_cmd(cmd_parser) self.assertEqual((5, MultiConfigData.LOCAL), self.tool.config_data.get_value("/foo/an_int")) cmd_parser = cmdparse.BindCmdParser('config unset identifier=' '"foo/an_int"') self.tool.apply_config_cmd(cmd_parser) self.assertEqual((1, MultiConfigData.DEFAULT), self.tool.config_data.get_value("/foo/an_int")) # this should raise a NotFoundError cmd_parser = cmdparse.BindCmdParser('config set identifier=' '"foo/bar" value="[]"') self.assertRaises(isc.cc.data.DataNotFoundError, self.tool.apply_config_cmd, cmd_parser) cmd_parser = cmdparse.BindCmdParser('config unset identifier=' '"foo/bar"') self.assertRaises(isc.cc.data.DataNotFoundError, self.tool.apply_config_cmd, cmd_parser) # this should raise a TypeError cmd_parser = cmdparse.BindCmdParser('config set identifier=' '"foo/an_int" value="[]"') self.assertRaises(isc.cc.data.DataTypeError, self.tool.apply_config_cmd, cmd_parser) # this is a very specific one for use with a set of list tests # to try out the flexibility of the parser (only in the next test) def clt(self, full_cmd_string, item_value): cmd_parser = cmdparse.BindCmdParser(full_cmd_string) self.tool.apply_config_cmd(cmd_parser) self.assertEqual(([item_value], MultiConfigData.LOCAL), self.tool.config_data.get_value("/foo/a_list")) def test_apply_cfg_command_list(self): self.tool.location = '/' self.assertEqual(([], MultiConfigData.DEFAULT), self.tool.config_data.get_value("/foo/a_list")) self.clt("config set identifier=\"foo/a_list\" value=[\"a\"]", "a") self.clt("config set identifier=\"foo/a_list\" value =[\"b\"]", "b") self.clt("config set identifier=\"foo/a_list\" value= [\"c\"]", "c") self.clt("config set identifier=\"foo/a_list\" value = [\"d\"]", "d") self.clt("config set identifier =\"foo/a_list\" value=[\"e\"]", "e") self.clt("config set identifier= \"foo/a_list\" value=[\"f\"]", "f") self.clt("config set identifier = \"foo/a_list\" value=[\"g\"]", "g") self.clt("config set identifier = \"foo/a_list\" value = [\"h\"]", "h") self.clt("config set identifier = \"foo/a_list\" value=[\"i\" ]", "i") self.clt("config set identifier = \"foo/a_list\" value=[ \"j\"]", "j") self.clt("config set identifier = \"foo/a_list\" value=[ \"k\" ]", "k") # this should raise a TypeError cmd_parser = cmdparse.BindCmdParser('config set identifier=' '"foo/a_list" value="a"') self.assertRaises(isc.cc.data.DataTypeError, self.tool.apply_config_cmd, cmd_parser) cmd_parser = cmdparse.BindCmdParser('config set identifier=' '"foo/a_list" value=[1]') self.assertRaises(isc.cc.data.DataTypeError, self.tool.apply_config_cmd, cmd_parser) def tearDown(self): sys.stdout = self.stdout_backup def test_cmd_has_identifier_param(self): module = ModuleInfo(name="test_module") cmd = CommandInfo(name="command_with_identifier") param = ParamInfo(name=bindcmd.CFGITEM_IDENTIFIER_PARAM) cmd.add_param(param) module.add_command(cmd) cmd = CommandInfo(name="command_without_identifier") param = ParamInfo(name="some_argument") cmd.add_param(param) module.add_command(cmd) self.tool.add_module_info(module) cmd_parser = cmdparse.BindCmdParser('test_module ' 'command_with_identifier') self.assertTrue(self.tool._cmd_has_identifier_param(cmd_parser)) cmd_parser = cmdparse.BindCmdParser('test_module ' 'command_without_identifier') self.assertFalse(self.tool._cmd_has_identifier_param(cmd_parser)) cmd_parser = cmdparse.BindCmdParser('badmodule ' 'command_without_identifier') self.assertFalse(self.tool._cmd_has_identifier_param(cmd_parser)) def test_get_identifier_startswith(self): hints = self.tool._get_identifier_startswith("/") self.assertEqual(['foo/an_int', 'foo/a_list'], hints) hints = self.tool._get_identifier_startswith("/foo/an") self.assertEqual(['foo/an_int'], hints) hints = self.tool._get_identifier_startswith("/bar") self.assertEqual([], hints) class FakeBindCmdInterpreter(bindcmd.BindCmdInterpreter): def __init__(self): pass class TestBindCmdInterpreter(unittest.TestCase): def setUp(self): self.old_stdout = sys.stdout def tearDown(self): sys.stdout = self.old_stdout def _create_invalid_csv_file(self, csvfilename): import csv csvfile = open(csvfilename, 'w') writer = csv.writer(csvfile) writer.writerow(['name1']) writer.writerow(['name2']) csvfile.close() def test_csv_file_dir(self): # Checking default value home_dir = pwd.getpwnam(getpass.getuser()).pw_dir self.assertEqual(home_dir + os.sep + '.bind10' + os.sep, bindcmd.BindCmdInterpreter().csv_file_dir) new_csv_dir = '/something/different/' custom_cmd = bindcmd.BindCmdInterpreter(csv_file_dir=new_csv_dir) self.assertEqual(new_csv_dir, custom_cmd.csv_file_dir) def test_get_saved_user_info(self): with open(os.devnull, 'w') as f: sys.stdout = f cmd = bindcmd.BindCmdInterpreter() users = cmd._get_saved_user_info('/notexist', 'csv_file.csv') self.assertEqual([], users) csvfilename = 'csv_file.csv' self._create_invalid_csv_file(csvfilename) users = cmd._get_saved_user_info('./', csvfilename) self.assertEqual([], users) os.remove(csvfilename) class TestCommandLineOptions(unittest.TestCase): def setUp(self): self.parser = TestOptParser() set_bindctl_options(self.parser) def test_csv_file_dir(self): # by default the option is "undefined" (options, _) = self.parser.parse_args([]) self.assertEqual(None, options.csv_file_dir) # specify the option, valid case. (options, _) = self.parser.parse_args(['--csv-file-dir', 'some_dir']) self.assertEqual('some_dir', options.csv_file_dir) # missing option arg; should trigger parser error. self.assertRaises(OptsError, self.parser.parse_args, ['--csv-file-dir']) if __name__== "__main__": unittest.main()