self.csv_file_dir = pwd.getpwnam(getpass.getuser()).pw_dir + \
os.sep + '.bind10' + os.sep
+ def _print(self, *args):
+ '''Simple wrapper around calls to print that can be overridden in
+ unit tests.'''
+ print(*args)
+
def _get_session_id(self):
'''Generate one session id for the connection. '''
rand = os.urandom(16)
return 1
self.cmdloop()
- print('\nExit from bindctl')
+ self._print('\nExit from bindctl')
return 0
except FailToLogin as err:
# error already printed when this was raised, ignoring
return 1
except KeyboardInterrupt:
- print('\nExit from bindctl')
+ self._print('\nExit from bindctl')
return 0
except socket.error as err:
- print('Failed to send request, the connection is closed')
+ self._print('Failed to send request, the connection is closed')
return 1
except http.client.CannotSendRequest:
- print('Can not send request, the connection is busy')
+ self._print('Can not send request, the connection is busy')
return 1
def _get_saved_user_info(self, dir, file_name):
for row in users_info:
users.append([row[0], row[1]])
except (IOError, IndexError) as err:
- print("Error reading saved username and password from %s%s: %s" % (dir, file_name, err))
+ self._print("Error reading saved username and password "
+ "from %s%s: %s" % (dir, file_name, err))
finally:
if csvfile:
csvfile.close()
writer.writerow([username, passwd])
csvfile.close()
except IOError as err:
- print("Error saving user information:", err)
- print("user info file name: %s%s" % (dir, file_name))
+ self._print("Error saving user information:", err)
+ self._print("user info file name: %s%s" % (dir, file_name))
return False
return True
- def __try_login(self, username, password):
+ def _try_login(self, username, password):
'''
- Attempts to log in to bindctl by sending a POST with
+ Attempts to log in to cmdctl by sending a POST with
the given username and password.
On success of the POST (mind, not the login, only the network
operation), returns a tuple (response, data).
On failure, raises a FailToLogin exception, and prints some
information on the failure.
+ This call is essentially 'private', but made 'protected' for
+ easier testing.
'''
param = {'username': username, 'password' : password}
try:
data = response.read().decode()
return (response, data)
except socket.error as err:
- print("Socket error while sending login information: ", err)
+ self._print("Socket error while sending login information: ", err)
if err.errno == errno.ECONNRESET:
- print("Please check the logs of b10-cmdctl, there may be a "
- "problem accepting SSL connections, such as a "
- "permission problem on the server certificate file.")
+ self._print("Please check the logs of b10-cmdctl, there may "
+ "be a problem accepting SSL connections, such "
+ "as a permission problem on the server "
+ "certificate file.")
raise FailToLogin()
def login_to_cmdctl(self):
# Look at existing username/password combinations and try to log in
users = self._get_saved_user_info(self.csv_file_dir, CSV_FILE_NAME)
for row in users:
- response, data = self.__try_login(row[0], row[1])
+ response, data = self._try_login(row[0], row[1])
if response.status == http.client.OK:
# Is interactive?
if sys.stdin.isatty():
- print(data + ' login as ' + row[0])
+ self._print(data + ' login as ' + row[0])
return True
# No valid logins were found, prompt the user for a username/password
count = 0
- print('No stored password file found, please see sections '
+ self._print('No stored password file found, please see sections '
'"Configuration specification for b10-cmdctl" and "bindctl '
'command-line options" of the BIND 10 guide.')
while True:
count = count + 1
if count > 3:
- print("Too many authentication failures")
+ self._print("Too many authentication failures")
return False
username = input("Username: ")
passwd = getpass.getpass()
- response, data = self.__try_login(username, passwd)
- print(data)
+ response, data = self._try_login(username, passwd)
+ self._print(data)
if response.status == http.client.OK:
self._save_user_info(username, passwd, self.csv_file_dir,
pass
def do_help(self, name):
- print(CONST_BINDCTL_HELP)
+ self._print(CONST_BINDCTL_HELP)
for k in self.modules.values():
n = k.get_name()
if len(n) >= CONST_BINDCTL_HELP_INDENT_WIDTH:
- print(" %s" % n)
- print(textwrap.fill(k.get_desc(),
- initial_indent=" ",
- subsequent_indent=" " +
- " " * CONST_BINDCTL_HELP_INDENT_WIDTH,
- width=70))
+ self._print(" %s" % n)
+ self._print(textwrap.fill(k.get_desc(),
+ initial_indent=" ",
+ subsequent_indent=" " +
+ " " * CONST_BINDCTL_HELP_INDENT_WIDTH,
+ width=70))
else:
- print(textwrap.fill("%s%s%s" %
- (k.get_name(),
- " "*(CONST_BINDCTL_HELP_INDENT_WIDTH - len(k.get_name())),
- k.get_desc()),
- initial_indent=" ",
- subsequent_indent=" " +
- " " * CONST_BINDCTL_HELP_INDENT_WIDTH,
- width=70))
+ self._print(textwrap.fill("%s%s%s" %
+ (k.get_name(),
+ " "*(CONST_BINDCTL_HELP_INDENT_WIDTH -
+ len(k.get_name())),
+ k.get_desc()),
+ initial_indent=" ",
+ subsequent_indent=" " +
+ " " * CONST_BINDCTL_HELP_INDENT_WIDTH,
+ width=70))
def onecmd(self, line):
if line == 'EOF' or line.lower() == "quit":
self._validate_cmd(cmd)
self._handle_cmd(cmd)
except (IOError, http.client.HTTPException) as err:
- print('Error: ', err)
+ self._print('Error: ', err)
except BindCtlException as err:
- print("Error! ", err)
+ self._print("Error! ", err)
self._print_correct_usage(err)
except isc.cc.data.DataTypeError as err:
- print("Error! ", err)
+ self._print("Error! ", err)
except isc.cc.data.DataTypeError as dte:
- print("Error: " + str(dte))
+ self._print("Error: " + str(dte))
except isc.cc.data.DataNotFoundError as dnfe:
- print("Error: " + str(dnfe))
+ self._print("Error: " + str(dnfe))
except isc.cc.data.DataAlreadyPresentError as dape:
- print("Error: " + str(dape))
+ self._print("Error: " + str(dape))
except KeyError as ke:
- print("Error: missing " + str(ke))
+ self._print("Error: missing " + str(ke))
def _print_correct_usage(self, ept):
if isinstance(ept, CmdUnknownModuleSyntaxError):
module_name = identifier.split('/')[1]
if module_name != "" and (self.config_data is None or \
not self.config_data.have_specification(module_name)):
- print("Error: Module '" + module_name + "' unknown or not running")
+ self._print("Error: Module '" + module_name +
+ "' unknown or not running")
return
if cmd.command == "show":
#identifier
identifier += cmd.params['argument']
else:
- print("Error: unknown argument " + cmd.params['argument'] + ", or multiple identifiers given")
+ self._print("Error: unknown argument " +
+ cmd.params['argument'] +
+ ", or multiple identifiers given")
return
values = self.config_data.get_value_maps(identifier, show_all)
for value_map in values:
line += "(default)"
if value_map['modified']:
line += "(modified)"
- print(line)
+ self._print(line)
elif cmd.command == "show_json":
if identifier == "":
- print("Need at least the module to show the configuration in JSON format")
+ self._print("Need at least the module to show the "
+ "configuration in JSON format")
else:
data, default = self.config_data.get_value(identifier)
- print(json.dumps(data))
+ self._print(json.dumps(data))
elif cmd.command == "add":
self.config_data.add_value(identifier,
cmd.params.get('value_or_name'),
self.config_data.remove_value(identifier, None)
elif cmd.command == "set":
if 'identifier' not in cmd.params:
- print("Error: missing identifier or value")
+ self._print("Error: missing identifier or value")
else:
parsed_value = None
try:
try:
self.config_data.commit()
except isc.config.ModuleCCSessionError as mcse:
- print(str(mcse))
+ self._print(str(mcse))
elif cmd.command == "diff":
- print(self.config_data.get_local_changes())
+ self._print(self.config_data.get_local_changes())
elif cmd.command == "go":
self.go(identifier)
# check if exists, if not, revert and error
v,d = self.config_data.get_value(new_location)
if v is None:
- print("Error: " + identifier + " not found")
+ self._print("Error: " + identifier + " not found")
return
self.location = new_location
with open(command.params['filename']) as command_file:
commands = command_file.readlines()
except IOError as ioe:
- print("Error: " + str(ioe))
+ self._print("Error: " + str(ioe))
return
elif command_sets.has_command_set(command.command):
commands = command_sets.get_commands(command.command)
def __show_execute_commands(self, commands):
'''Prints the command list without executing them'''
for line in commands:
- print(line.strip())
+ self._print(line.strip())
def __apply_execute_commands(self, commands):
'''Applies the configuration commands from the given iterator.
for line in commands:
line = line.strip()
if verbose:
- print(line)
+ self._print(line)
if line.startswith('#') or len(line) == 0:
continue
elif line.startswith('!'):
if re.match('^!echo ', line, re.I) and len(line) > 6:
- print(line[6:])
+ self._print(line[6:])
elif re.match('^!verbose\s+on\s*$', line, re.I):
verbose = True
elif re.match('^!verbose\s+off$', line, re.I):
verbose = False
else:
- print("Warning: ignoring unknown directive: " + line)
+ self._print("Warning: ignoring unknown directive: " +
+ line)
else:
cmd = BindCmdParser(line)
self._validate_cmd(cmd)
isc.cc.data.DataNotFoundError,
isc.cc.data.DataAlreadyPresentError,
KeyError) as err:
- print('Error: ', err)
- print()
- print('Depending on the contents of the script, and which')
- print('commands it has called, there can be committed and')
- print('local changes. It is advised to check your settings,')
- print('and revert local changes with "config revert".')
+ self._print('Error: ', err)
+ self._print()
+ self._print('Depending on the contents of the script, and which')
+ self._print('commands it has called, there can be committed and')
+ self._print('local changes. It is advised to check your settings')
+ self._print(', and revert local changes with "config revert".')
def apply_cmd(self, cmd):
'''Handles a general module command'''
# The reply is a string containing JSON data,
# parse it, then prettyprint
if data != "" and data != "{}":
- print(json.dumps(json.loads(data), sort_keys=True, indent=4))
+ self._print(json.dumps(json.loads(data), sort_keys=True,
+ indent=4))
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():
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 test_try_login(self):
# Make sure __try_login raises the correct exception
# upon failure of either send_POST or the read() on the
raise socket.error("test error")
self.tool.send_POST = send_POST_raiseImmediately
- self.assertRaises(FailToLogin,
- self.tool._BindCmdInterpreter__try_login,
- "foo", "bar")
-
- def send_POST_raiseOnRead(self, params):
- class MyResponse:
- def read(self):
- raise socket.error("read error")
- return MyResponse()
-
- self.tool.send_POST = send_POST_raiseOnRead
- self.assertRaises(FailToLogin,
- self.tool._BindCmdInterpreter__try_login,
- "foo", "bar")
+ self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
+ self.assertIn('Socket error while sending login information: test error',
+ self.printed_messages)
+ self.assertEqual(1, len(self.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
+
+ self.tool.send_POST =\
+ create_send_POST_raiseOnRead(socket.error("read error"))
+ self.assertRaises(FailToLogin, self.tool._try_login, "foo", "bar")
+ self.assertIn('Socket error while sending login information: read error',
+ self.printed_messages)
+ self.assertEqual(2, len(self.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.assertEqual(2, len(self.printed_messages))
+
finally:
self.tool.send_POST = orig_send_POST
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())
# 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",
- socket_err_output.getvalue())
-
- socket_err_output.close()
+ self.printed_messages)
+ self.assertEqual(1, len(self.printed_messages))
# validate log message for http.client.CannotSendRequest
- cannot_send_output = io.StringIO()
- sys.stdout = cannot_send_output
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",
- cannot_send_output.getvalue())
-
- cannot_send_output.close()
+ self.printed_messages)
+ self.assertEqual(2, len(self.printed_messages))
def test_apply_cfg_command_int(self):
self.tool.location = '/'
projects / thirdparty / kea.git / commitdiff
