# 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 errno import sys import socket import ssl import http.client import pwd import getpass import re import json 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 self.printed_messages = [] self.tool._print = self.store_print 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 store_print(self, *args): '''Method to override _print in BindCmdInterpreter. Instead of printing the values, appends the argument tuple to the list in self.printed_messages''' self.printed_messages.append(" ".join(map(str, args))) def __check_printed_message(self, expected_message, printed_message): self.assertIsNotNone(re.match(expected_message, printed_message), "Printed message '" + printed_message + "' does not match '" + expected_message + "'") def __check_printed_messages(self, expected_messages): '''Helper test function to check the printed messages against a list of regexps''' self.assertEqual(len(expected_messages), len(self.printed_messages)) for _ in map(self.__check_printed_message, expected_messages, self.printed_messages): pass def test_try_login(self): # Make sure __try_login raises the correct exception # upon failure of either send_POST or the read() on the # response orig_send_POST = self.tool.send_POST expected_printed_messages = [] try: def send_POST_raiseImmediately(self, params): raise socket.error("test error") self.tool.send_POST = send_POST_raiseImmediately self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar") expected_printed_messages.append( 'Error while sending login information: test error') self.__check_printed_messages(expected_printed_messages) def create_send_POST_raiseOnRead(exception): '''Create a replacement send_POST() method that raises the given exception when read() is called on the value returned from send_POST()''' def send_POST_raiseOnRead(self, params): class MyResponse: def read(self): raise exception return MyResponse() return send_POST_raiseOnRead # basic socket error self.tool.send_POST =\ create_send_POST_raiseOnRead(socket.error("read error")) self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar") expected_printed_messages.append( 'Error while sending login information: read error') self.__check_printed_messages(expected_printed_messages) # connection reset exc = socket.error("connection reset") exc.errno = errno.ECONNRESET self.tool.send_POST =\ create_send_POST_raiseOnRead(exc) self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar") expected_printed_messages.append( 'Error while sending login information: connection reset') self.__check_printed_messages(expected_printed_messages) # 'normal' SSL error exc = ssl.SSLError() self.tool.send_POST =\ create_send_POST_raiseOnRead(exc) self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar") expected_printed_messages.append( 'Error while sending login information: .*') self.__check_printed_messages(expected_printed_messages) # 'EOF' SSL error exc = ssl.SSLError() exc.errno = ssl.SSL_ERROR_EOF self.tool.send_POST =\ create_send_POST_raiseOnRead(exc) self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar") expected_printed_messages.append( 'Error while sending login information: .*') self.__check_printed_messages(expected_printed_messages) # any other exception should be passed through self.tool.send_POST =\ create_send_POST_raiseOnRead(ImportError()) self.assertRaises(ImportError, self.tool._try_login, "foo", "bar") self.__check_printed_messages(expected_printed_messages) finally: self.tool.send_POST = orig_send_POST def test_try_login_calls_cmdctl(self): # Make sure _try_login() makes the right API call to cmdctl. orig_conn = self.tool.conn try: class MyConn: def __init__(self): self.method = None self.url = None self.param = None self.headers = None def request(self, method, url, param, headers): self.method = method self.url = url self.param = param self.headers = headers def getresponse(self): class MyResponse: def __init__(self): self.status = http.client.OK def read(self): class MyData: def decode(self): return json.dumps(True) return MyData() return MyResponse() self.tool.conn = MyConn() self.assertTrue(self.tool._try_login('user32', 'pass64')) self.assertEqual(self.tool.conn.method, 'POST') self.assertEqual(self.tool.conn.url, '/login') self.assertEqual(json.loads(self.tool.conn.param), {"password": "pass64", "username": "user32"}) self.assertIn('cookie', self.tool.conn.headers) finally: self.tool.conn = orig_conn 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() self.assertEqual(1, self.tool.run()) # First few lines may be some kind of heading, or a warning that # Python readline is unavailable, so we do a sub-string check. self.assertIn("Failed to send request, the connection is closed", self.printed_messages) self.assertEqual(1, len(self.printed_messages)) # validate log message for http.client.CannotSendRequest self.assertEqual(1, self.tool.run()) # First few lines may be some kind of heading, or a warning that # Python readline is unavailable, so we do a sub-string check. self.assertIn("Can not send request, the connection is busy", self.printed_messages) self.assertEqual(2, len(self.printed_messages)) 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()