From 580bbea80774b33faeb90f16b973d71e36ed135f Mon Sep 17 00:00:00 2001 From: rshipp Date: Fri, 19 Apr 2019 18:35:31 -0500 Subject: [PATCH] Convert to and from bytes for telnet For #2. --- nut2.py | 120 +++++++++++++------------- tests/mockserver.py | 204 ++++++++++++++++++++++---------------------- 2 files changed, 162 insertions(+), 162 deletions(-) diff --git a/nut2.py b/nut2.py index 872dfd4..44aa35f 100644 --- a/nut2.py +++ b/nut2.py @@ -73,7 +73,7 @@ def __del__(self): # Try to disconnect cleanly when class is deleted. if self._srv_handler: try: - self._srv_handler.write("LOGOUT\n") + self._srv_handler.write(b"LOGOUT\n") self._srv_handler.close() except (telnetlib.socket.error, AttributeError): # The socket is already disconnected. @@ -98,14 +98,14 @@ def _connect(self): timeout=self._timeout) if self._login is not None: - self._srv_handler.write("USERNAME %s\n" % self._login) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"USERNAME %s\n" % self._login.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if not result == "OK\n": raise PyNUTError(result.replace("\n", "")) if self._password is not None: - self._srv_handler.write("PASSWORD %s\n" % self._password) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"PASSWORD %s\n" % self._password.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if not result == "OK\n": raise PyNUTError(result.replace("\n", "")) except telnetlib.socket.error: @@ -115,8 +115,8 @@ def description(self, ups): """Returns the description for a given UPS.""" logging.debug("description called...") - self._srv_handler.write("GET UPSDESC %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET UPSDESC %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') try: return result.split('"')[1].strip() except IndexError: @@ -130,13 +130,13 @@ def list_ups(self): """ logging.debug("list_ups from server") - self._srv_handler.write("LIST UPS\n") - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST UPS\n") + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST UPS\n": raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST UPS\n", - self._timeout) + result = self._srv_handler.read_until(b"END LIST UPS\n", + self._timeout).decode('utf-8') ups_dict = {} for line in result.split("\n"): @@ -154,13 +154,13 @@ def list_vars(self, ups): """ logging.debug("list_vars called...") - self._srv_handler.write("LIST VAR %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST VAR %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST VAR %s\n" % ups: raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST VAR %s\n" % ups, - self._timeout) + result = self._srv_handler.read_until(b"END LIST VAR %s\n" % ups.encode('utf-8'), + self._timeout).decode('utf-8') offset = len("VAR %s " % ups) end_offset = 0 - (len("END LIST VAR %s\n" % ups) + 1) @@ -179,13 +179,13 @@ def list_commands(self, ups): """ logging.debug("list_commands called...") - self._srv_handler.write("LIST CMD %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST CMD %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST CMD %s\n" % ups: raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST CMD %s\n" % ups, - self._timeout) + result = self._srv_handler.read_until(b"END LIST CMD %s\n" % ups.encode('utf-8'), + self._timeout).decode('utf-8') offset = len("CMD %s " % ups) end_offset = 0 - (len("END LIST CMD %s\n" % ups) + 1) @@ -195,8 +195,8 @@ def list_commands(self, ups): # For each var we try to get the available description try: - self._srv_handler.write("GET CMDDESC %s %s\n" % (ups, command)) - temp = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET CMDDESC %s %s\n" % (ups.encode('utf-8'), command.encode('utf-8'))) + temp = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if temp.startswith("CMDDESC"): desc_offset = len("CMDDESC %s %s " % (ups, command)) commands[command] = temp[desc_offset:-1].split('"')[1] @@ -219,15 +219,15 @@ def list_clients(self, ups=None): raise PyNUTError("%s is not a valid UPS" % ups) if ups: - self._srv_handler.write("LIST CLIENTS %s\n" % ups) + self._srv_handler.write(b"LIST CLIENTS %s\n" % ups.encode('utf-8')) else: - self._srv_handler.write("LIST CLIENTS\n") - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST CLIENTS\n") + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST CLIENTS\n": raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST CLIENTS\n", - self._timeout) + result = self._srv_handler.read_until(b"END LIST CLIENTS\n", + self._timeout).decode('utf-8') clients = {} for line in result.split("\n"): @@ -247,13 +247,13 @@ def list_rw_vars(self, ups): """ logging.debug("list_vars from '%s'...", ups) - self._srv_handler.write("LIST RW %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST RW %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST RW %s\n" % ups: raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST RW %s\n" % ups, - self._timeout) + result = self._srv_handler.read_until(b"END LIST RW %s\n" % ups.encode('utf-8'), + self._timeout).decode('utf-8') offset = len("VAR %s" % ups) end_offset = 0 - (len("END LIST RW %s\n" % ups) + 1) @@ -271,13 +271,13 @@ def list_enum(self, ups, var): """ logging.debug("list_enum from '%s'...", ups) - self._srv_handler.write("LIST ENUM %s %s\n" % (ups, var)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST ENUM %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST ENUM %s %s\n" % (ups, var): raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST ENUM %s %s\n" % (ups, var), - self._timeout) + result = self._srv_handler.read_until(b"END LIST ENUM %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8')), + self._timeout).decode('utf-8') offset = len("ENUM %s %s" % (ups, var)) end_offset = 0 - (len("END LIST ENUM %s %s\n" % (ups, var)) + 1) @@ -294,13 +294,13 @@ def list_range(self, ups, var): """ logging.debug("list_range from '%s'...", ups) - self._srv_handler.write("LIST RANGE %s %s\n" % (ups, var)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"LIST RANGE %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "BEGIN LIST RANGE %s %s\n" % (ups, var): raise PyNUTError(result.replace("\n", "")) - result = self._srv_handler.read_until("END LIST RANGE %s %s\n" % (ups, var), - self._timeout) + result = self._srv_handler.read_until(b"END LIST RANGE %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8')), + self._timeout).decode('utf-8') offset = len("RANGE %s %s" % (ups, var)) end_offset = 0 - (len("END LIST RANGE %s %s\n" % (ups, var)) + 1) @@ -318,8 +318,8 @@ def set_var(self, ups, var, value): """ logging.debug("set_var '%s' from '%s' to '%s'", var, ups, value) - self._srv_handler.write("SET VAR %s %s %s\n" % (ups, var, value)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"SET VAR %s %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8'), value.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "OK\n": raise PyNUTError(result.replace("\n", "")) @@ -327,8 +327,8 @@ def get_var(self, ups, var): """Get the value of a variable.""" logging.debug("get_var called...") - self._srv_handler.write("GET VAR %s %s\n" % (ups, var)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET VAR %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') try: # result = 'VAR %s %s "%s"\n' % (ups, var, value) return result.split('"')[1].strip() @@ -344,8 +344,8 @@ def var_description(self, ups, var): """Get a variable's description.""" logging.debug("var_description called...") - self._srv_handler.write("GET DESC %s %s\n" % (ups, var)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET DESC %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') try: # result = 'DESC %s %s "%s"\n' % (ups, var, description) return result.split('"')[1].strip() @@ -356,8 +356,8 @@ def var_type(self, ups, var): """Get a variable's type.""" logging.debug("var_type called...") - self._srv_handler.write("GET TYPE %s %s\n" % (ups, var)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET TYPE %s %s\n" % (ups.encode('utf-8'), var.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') try: # result = 'TYPE %s %s %s\n' % (ups, var, type) type_ = ' '.join(result.split(' ')[3:]).strip() @@ -372,8 +372,8 @@ def command_description(self, ups, command): """Get a command's description.""" logging.debug("command_description called...") - self._srv_handler.write("GET CMDDESC %s %s\n" % (ups, command)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET CMDDESC %s %s\n" % (ups.encode('utf-8'), command.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') try: # result = 'CMDDESC %s %s "%s"' % (ups, command, description) return result.split('"')[1].strip() @@ -384,8 +384,8 @@ def run_command(self, ups, command): """Send a command to the specified UPS.""" logging.debug("run_command called...") - self._srv_handler.write("INSTCMD %s %s\n" % (ups, command)) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"INSTCMD %s %s\n" % (ups.encode('utf-8'), command.encode('utf-8'))) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "OK\n": raise PyNUTError(result.replace("\n", "")) @@ -393,14 +393,14 @@ def fsd(self, ups): """Send MASTER and FSD commands.""" logging.debug("MASTER called...") - self._srv_handler.write("MASTER %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"MASTER %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "OK MASTER-GRANTED\n": raise PyNUTError(("Master level function are not available", "")) logging.debug("FSD called...") - self._srv_handler.write("FSD %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"FSD %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') if result != "OK FSD-SET\n": raise PyNUTError(result.replace("\n", "")) @@ -410,8 +410,8 @@ def num_logins(self, ups): """ logging.debug("num_logins called on '%s'...", ups) - self._srv_handler.write("GET NUMLOGINS %s\n" % ups) - result = self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"GET NUMLOGINS %s\n" % ups.encode('utf-8')) + result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') try: # result = "NUMLOGINS %s %s\n" % (ups, int(numlogins)) return int(result.split(' ')[2].strip()) @@ -422,12 +422,12 @@ def help(self): """Send HELP command.""" logging.debug("HELP called...") - self._srv_handler.write("HELP\n") - return self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"HELP\n") + return self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') def ver(self): """Send VER command.""" logging.debug("VER called...") - self._srv_handler.write("VER\n") - return self._srv_handler.read_until("\n", self._timeout) + self._srv_handler.write(b"VER\n") + return self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') diff --git a/tests/mockserver.py b/tests/mockserver.py index d782e52..ead4d04 100644 --- a/tests/mockserver.py +++ b/tests/mockserver.py @@ -3,8 +3,8 @@ class MockServer(object): def __init__(self, host=None, port=None, broken=True, ok=True, broken_username=False, timeout=None): - self.valid = "test" - self.valid_desc = '"Test UPS 1"' + self.valid = b"test" + self.valid_desc = b'"Test UPS 1"' self.broken = broken self.ok = ok self.broken_username = broken_username @@ -20,115 +20,115 @@ def close(self): pass def run_command(self): - if self.broken and not self.broken_username and self.command == "USERNAME %s\n" % self.valid: - return 'OK\n' + if self.broken and not self.broken_username and self.command == b"USERNAME %s\n" % self.valid: + return b'OK\n' elif self.broken: - return 'ERR\n' - elif self.command == "HELP\n": - return 'Commands: HELP VER GET LIST SET INSTCMD LOGIN LOGOUT USERNAME PASSWORD STARTTLS\n' - elif self.command == "VER\n": - return 'Network UPS Tools upsd 2.7.1 - http://www.networkupstools.org/\n' - elif self.command == "GET CMDDESC %s %s\n" % (self.valid, self.valid): - return 'CMDDESC '+self.valid+' '+self.valid+' '+self.valid_desc+'\n' - elif self.command == "LIST UPS\n" and self.first: + return b'ERR\n' + elif self.command == b"HELP\n": + return b'Commands: HELP VER GET LIST SET INSTCMD LOGIN LOGOUT USERNAME PASSWORD STARTTLS\n' + elif self.command == b"VER\n": + return b'Network UPS Tools upsd 2.7.1 - http://www.networkupstools.org/\n' + elif self.command == b"GET CMDDESC %s %s\n" % (self.valid, self.valid): + return b'CMDDESC '+self.valid+b' '+self.valid+b' '+self.valid_desc+b'\n' + elif self.command == b"LIST UPS\n" and self.first: self.first = False - return 'BEGIN LIST UPS\n' - elif self.command == "LIST UPS\n": - return 'UPS '+self.valid+' '+self.valid_desc+'\nUPS Test_UPS2 "Test UPS 2"\nEND LIST UPS\n' - elif self.command == "LIST VAR %s\n" % self.valid and self.first: + return b'BEGIN LIST UPS\n' + elif self.command == b"LIST UPS\n": + return b'UPS '+self.valid+b' '+self.valid_desc+b'\nUPS Test_UPS2 "Test UPS 2"\nEND LIST UPS\n' + elif self.command == b"LIST VAR %s\n" % self.valid and self.first: self.first = False - return 'BEGIN LIST VAR '+self.valid+'\n' - elif self.command == "LIST VAR %s\n" % self.valid: - return 'VAR '+self.valid+' battery.charge "100"\nVAR '+self.valid+' battery.voltage "14.44"\nEND LIST VAR '+self.valid+'\n' - elif self.command.startswith("LIST VAR"): - return 'ERR INVALID-ARGUMENT\n' - elif self.command == "LIST CMD %s\n" % self.valid and self.first: + return b'BEGIN LIST VAR '+self.valid+b'\n' + elif self.command == b"LIST VAR %s\n" % self.valid: + return b'VAR '+self.valid+b' battery.charge "100"\nVAR '+self.valid+b' battery.voltage "14.44"\nEND LIST VAR '+self.valid+b'\n' + elif self.command.startswith(b"LIST VAR"): + return b'ERR INVALID-ARGUMENT\n' + elif self.command == b"LIST CMD %s\n" % self.valid and self.first: self.first = False - return 'BEGIN LIST CMD '+self.valid+'\n' - elif self.command == "LIST CMD %s\n" % self.valid: - return 'CMD '+self.valid+' '+self.valid+'\nEND LIST CMD '+self.valid+'\n' - elif self.command.startswith("LIST CMD"): - return 'ERR INVALID-ARGUMENT\n' - elif self.command == "LIST RW %s\n" % self.valid and self.first: + return b'BEGIN LIST CMD '+self.valid+b'\n' + elif self.command == b"LIST CMD %s\n" % self.valid: + return b'CMD '+self.valid+b' '+self.valid+b'\nEND LIST CMD '+self.valid+b'\n' + elif self.command.startswith(b"LIST CMD"): + return b'ERR INVALID-ARGUMENT\n' + elif self.command == b"LIST RW %s\n" % self.valid and self.first: self.first = False - return 'BEGIN LIST RW '+self.valid+'\n' - elif self.command == "LIST RW %s\n" % self.valid: - return 'RW '+self.valid+' '+self.valid+' "test"\nEND LIST RW '+self.valid+'\n' - elif self.command.startswith("LIST RW"): - return 'ERR INVALID-ARGUMENT\n' - elif self.command == "LIST CLIENTS %s\n" % self.valid and self.first: + return b'BEGIN LIST RW '+self.valid+b'\n' + elif self.command == b"LIST RW %s\n" % self.valid: + return b'RW '+self.valid+b' '+self.valid+b' "test"\nEND LIST RW '+self.valid+b'\n' + elif self.command.startswith(b"LIST RW"): + return b'ERR INVALID-ARGUMENT\n' + elif self.command == b"LIST CLIENTS %s\n" % self.valid and self.first: self.first = False - return 'BEGIN LIST CLIENTS\n' - elif self.command == "LIST CLIENTS %s\n" % self.valid: - return 'CLIENT '+self.valid+' '+self.valid+'\nEND LIST CLIENTS\n' - elif self.command.startswith("LIST CLIENTS"): - return 'ERR INVALID-ARGUMENT\n' - elif self.command == "LIST ENUM %s %s\n" % (self.valid, self.valid) and self.first: + return b'BEGIN LIST CLIENTS\n' + elif self.command == b"LIST CLIENTS %s\n" % self.valid: + return b'CLIENT '+self.valid+b' '+self.valid+b'\nEND LIST CLIENTS\n' + elif self.command.startswith(b"LIST CLIENTS"): + return b'ERR INVALID-ARGUMENT\n' + elif self.command == b"LIST ENUM %s %s\n" % (self.valid, self.valid) and self.first: self.first = False - return 'BEGIN LIST ENUM %s %s\n' % (self.valid, self.valid) - elif self.command == "LIST ENUM %s %s\n" % (self.valid, self.valid): - return 'ENUM %s %s %s\nEND LIST ENUM %s %s\n' % (self.valid, + return b'BEGIN LIST ENUM %s %s\n' % (self.valid, self.valid) + elif self.command == b"LIST ENUM %s %s\n" % (self.valid, self.valid): + return b'ENUM %s %s %s\nEND LIST ENUM %s %s\n' % (self.valid, self.valid, self.valid_desc, self.valid, self.valid) - elif self.command == "LIST RANGE %s %s\n" % (self.valid, self.valid) and self.first: + elif self.command == b"LIST RANGE %s %s\n" % (self.valid, self.valid) and self.first: self.first = False - return 'BEGIN LIST RANGE %s %s\n' % (self.valid, self.valid) - elif self.command == "LIST RANGE %s %s\n" % (self.valid, self.valid): - return 'RANGE %s %s %s %s\nEND LIST RANGE %s %s\n' % (self.valid, + return b'BEGIN LIST RANGE %s %s\n' % (self.valid, self.valid) + elif self.command == b"LIST RANGE %s %s\n" % (self.valid, self.valid): + return b'RANGE %s %s %s %s\nEND LIST RANGE %s %s\n' % (self.valid, self.valid, self.valid_desc, self.valid_desc, self.valid, self.valid) - elif self.command == "SET VAR %s %s %s\n" % (self.valid, self.valid, self.valid): - return 'OK\n' - elif self.command.startswith("SET"): - return 'ERR ACCESS-DENIED\n' - elif self.command == "INSTCMD %s %s\n"% (self.valid, self.valid): - return 'OK\n' - elif self.command.startswith("INSTCMD"): - return 'ERR CMD-NOT-SUPPORTED\n' + elif self.command == b"SET VAR %s %s %s\n" % (self.valid, self.valid, self.valid): + return b'OK\n' + elif self.command.startswith(b"SET"): + return b'ERR ACCESS-DENIED\n' + elif self.command == b"INSTCMD %s %s\n"% (self.valid, self.valid): + return b'OK\n' + elif self.command.startswith(b"INSTCMD"): + return b'ERR CMD-NOT-SUPPORTED\n' # TODO: LOGIN/LOGOUT commands - elif self.command == "USERNAME %s\n" % self.valid: - return 'OK\n' - elif self.command.startswith("USERNAME"): - return 'ERR\n' # FIXME: What does it say on invalid password? - elif self.command == "PASSWORD %s\n" % self.valid: - return 'OK\n' - elif self.command.startswith("PASSWORD"): - return 'ERR\n' # FIXME: ^ - elif self.command == "STARTTLS\n": - return 'ERR FEATURE-NOT-CONFIGURED\n' - elif self.command == "MASTER %s\n" % self.valid: - return 'OK MASTER-GRANTED\n' - elif self.command == "FSD %s\n" % self.valid and self.ok: - return 'OK FSD-SET\n' - elif self.command == "FSD %s\n" % self.valid: - return 'ERR\n' - elif self.command == "GET NUMLOGINS %s\n" % self.valid: - return 'NUMLOGINS %s 1\n' % self.valid - elif self.command.startswith("GET NUMLOGINS"): - return 'ERR UNKNOWN-UPS\n' - elif self.command == "GET UPSDESC %s\n" % self.valid: - return 'UPSDESC %s %s\n' % (self.valid, self.valid_desc) - elif self.command.startswith("GET UPSDESC"): - return 'ERR UNKNOWN-UPS\n' - elif self.command == "GET VAR %s %s\n" % (self.valid, self.valid): - return 'VAR %s %s "100"\n' % (self.valid, self.valid) - elif self.command.startswith("GET VAR %s" % self.valid): - return 'ERR VAR-NOT-SUPPORTED\n' - elif self.command.startswith("GET VAR "): - return 'ERR UNKNOWN-UPS\n' - elif self.command.startswith("GET VAR"): - return 'ERR INVALID-ARGUMENT\n' - elif self.command == "GET TYPE %s %s\n" % (self.valid, self.valid): - return 'TYPE %s %s RW STRING:3\n' % (self.valid, self.valid) - elif self.command.startswith("GET TYPE %s" % self.valid): - return 'ERR VAR-NOT-SUPPORTED\n' - elif self.command.startswith("GET TYPE"): - return 'ERR INVALID-ARGUMENT\n' - elif self.command == "GET DESC %s %s\n" % (self.valid, self.valid): - return 'DESC %s %s %s\n' % (self.valid, self.valid, self.valid_desc) - elif self.command.startswith("GET DESC"): - return 'ERR-INVALID-ARGUMENT\n' - elif self.command == "GET CMDDESC %s %s" % (self.valid, self.valid): - return 'CMDDESC %s %s %s\n' % (self.valid, self.valid, self.valid_desc) - elif self.command.startswith("GET CMDDESC"): - return 'ERR INVALID-ARGUMENT' + elif self.command == b"USERNAME %s\n" % self.valid: + return b'OK\n' + elif self.command.startswith(b"USERNAME"): + return b'ERR\n' # FIXME: What does it say on invalid password? + elif self.command == b"PASSWORD %s\n" % self.valid: + return b'OK\n' + elif self.command.startswith(b"PASSWORD"): + return b'ERR\n' # FIXME: ^ + elif self.command == b"STARTTLS\n": + return b'ERR FEATURE-NOT-CONFIGURED\n' + elif self.command == b"MASTER %s\n" % self.valid: + return b'OK MASTER-GRANTED\n' + elif self.command == b"FSD %s\n" % self.valid and self.ok: + return b'OK FSD-SET\n' + elif self.command == b"FSD %s\n" % self.valid: + return b'ERR\n' + elif self.command == b"GET NUMLOGINS %s\n" % self.valid: + return b'NUMLOGINS %s 1\n' % self.valid + elif self.command.startswith(b"GET NUMLOGINS"): + return b'ERR UNKNOWN-UPS\n' + elif self.command == b"GET UPSDESC %s\n" % self.valid: + return b'UPSDESC %s %s\n' % (self.valid, self.valid_desc) + elif self.command.startswith(b"GET UPSDESC"): + return b'ERR UNKNOWN-UPS\n' + elif self.command == b"GET VAR %s %s\n" % (self.valid, self.valid): + return b'VAR %s %s "100"\n' % (self.valid, self.valid) + elif self.command.startswith(b"GET VAR %s" % self.valid): + return b'ERR VAR-NOT-SUPPORTED\n' + elif self.command.startswith(b"GET VAR "): + return b'ERR UNKNOWN-UPS\n' + elif self.command.startswith(b"GET VAR"): + return b'ERR INVALID-ARGUMENT\n' + elif self.command == b"GET TYPE %s %s\n" % (self.valid, self.valid): + return b'TYPE %s %s RW STRING:3\n' % (self.valid, self.valid) + elif self.command.startswith(b"GET TYPE %s" % self.valid): + return b'ERR VAR-NOT-SUPPORTED\n' + elif self.command.startswith(b"GET TYPE"): + return b'ERR INVALID-ARGUMENT\n' + elif self.command == b"GET DESC %s %s\n" % (self.valid, self.valid): + return b'DESC %s %s %s\n' % (self.valid, self.valid, self.valid_desc) + elif self.command.startswith(b"GET DESC"): + return b'ERR-INVALID-ARGUMENT\n' + elif self.command == b"GET CMDDESC %s %s" % (self.valid, self.valid): + return b'CMDDESC %s %s %s\n' % (self.valid, self.valid, self.valid_desc) + elif self.command.startswith(b"GET CMDDESC"): + return b'ERR INVALID-ARGUMENT' else: - return 'ERR UNKNOWN-COMMAND\n' + return b'ERR UNKNOWN-COMMAND\n'