diff --git a/nut2.py b/nut2.py index 99ea8a0..4876210 100644 --- a/nut2.py +++ b/nut2.py @@ -29,7 +29,8 @@ __version__ = '2.1.1' __all__ = ['PyNUTError', 'PyNUTClient'] -logging.basicConfig(level=logging.WARNING, format="[%(levelname)s] %(message)s") +logger = logging.getLogger(__name__) +logger.addHandler(logging.NullHandler()) class PyNUTError(Exception): @@ -38,7 +39,7 @@ class PyNUTError(Exception): class PyNUTClient(object): """Access NUT (Network UPS Tools) servers.""" - def __init__(self, host="127.0.0.1", port=3493, login=None, password=None, debug=False, timeout=5, connect=True): + def __init__(self, host="127.0.0.1", port=3493, login=None, password=None, timeout=5, connect=True): """Class initialization method. host : Host to connect (defaults to 127.0.0.1). @@ -46,18 +47,12 @@ def __init__(self, host="127.0.0.1", port=3493, login=None, password=None, debug login : Login used to connect to NUT server (defaults to None for no authentication). password : Password used when using authentication (defaults to None). - debug : Boolean, put class in debug mode (prints everything - on console, defaults to False). timeout : Timeout used to wait for network response (defaults to 5 seconds). """ - if debug: - # Print DEBUG messages to the console. - logging.getLogger().setLevel(logging.DEBUG) - - logging.debug("Class initialization...") - logging.debug(" -> Host = %s (port %s)", host, port) - logging.debug(" -> Login = '%s' / '%s'", login, password) + logger.debug("Class initialization...") + logger.debug(" -> Host = %s (port %s)", host, port) + logger.debug(" -> Login = '%s' / '%s'", login, password) self._host = host self._port = port @@ -91,7 +86,7 @@ def _connect(self): If login/pass was specified, the class tries to authenticate. An error is raised if something goes wrong. """ - logging.debug("Connecting to host") + logger.debug("Connecting to host") try: self._srv_handler = telnetlib.Telnet(self._host, self._port, @@ -113,7 +108,7 @@ def _connect(self): def description(self, ups): """Returns the description for a given UPS.""" - logging.debug("description called...") + logger.debug("description called...") 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') @@ -128,7 +123,7 @@ def list_ups(self): The result is a dictionary containing 'key->val' pairs of 'UPSName' and 'UPS Description'. """ - logging.debug("list_ups from server") + logger.debug("list_ups from server") self._srv_handler.write(b"LIST UPS\n") result = self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') @@ -152,7 +147,7 @@ def list_vars(self, ups): The result is a dictionary containing 'key->val' pairs of all available vars. """ - logging.debug("list_vars called...") + logger.debug("list_vars called...") 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') @@ -177,7 +172,7 @@ def list_commands(self, ups): The result is a dict object with command name as key and a description of the command as value. """ - logging.debug("list_commands called...") + logger.debug("list_commands called...") 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') @@ -213,7 +208,7 @@ def list_clients(self, ups=None): The result is a dictionary containing 'key->val' pairs of 'UPSName' and a list of clients. """ - logging.debug("list_clients from '%s'...", ups or "server") + logger.debug("list_clients from '%s'...", ups or "server") if ups and (ups not in self.list_ups()): raise PyNUTError("%s is not a valid UPS" % ups) @@ -245,7 +240,7 @@ def list_rw_vars(self, ups): The result is presented as a dictionary containing 'key->val' pairs. """ - logging.debug("list_vars from '%s'...", ups) + logger.debug("list_vars from '%s'...", ups) 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') @@ -269,7 +264,7 @@ def list_enum(self, ups, var): The result is presented as a list. """ - logging.debug("list_enum from '%s'...", ups) + logger.debug("list_enum from '%s'...", ups) 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') @@ -282,7 +277,7 @@ def list_enum(self, ups, var): end_offset = 0 - (len("END LIST ENUM %s %s\n" % (ups, var)) + 1) try: - return [ c[offset:].split('"')[1].strip() + return [ c[offset:].split('"')[1].strip() for c in result[:end_offset].split("\n") ] except IndexError: raise PyNUTError(result.replace("\n", "")) @@ -292,7 +287,7 @@ def list_range(self, ups, var): The result is presented as a list. """ - logging.debug("list_range from '%s'...", ups) + logger.debug("list_range from '%s'...", ups) 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') @@ -305,7 +300,7 @@ def list_range(self, ups, var): end_offset = 0 - (len("END LIST RANGE %s %s\n" % (ups, var)) + 1) try: - return [ c[offset:].split('"')[1].strip() + return [ c[offset:].split('"')[1].strip() for c in result[:end_offset].split("\n") ] except IndexError: raise PyNUTError(result.replace("\n", "")) @@ -316,7 +311,7 @@ def set_var(self, ups, var, value): The variable must be a writable value (cf list_rw_vars) and you must have the proper rights to set it (maybe login/password). """ - logging.debug("set_var '%s' from '%s' to '%s'", var, ups, value) + logger.debug("set_var '%s' from '%s' to '%s'", var, ups, value) 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') @@ -325,7 +320,7 @@ def set_var(self, ups, var, value): def get_var(self, ups, var): """Get the value of a variable.""" - logging.debug("get_var called...") + logger.debug("get_var called...") 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') @@ -342,7 +337,7 @@ def get(self, ups, var): def var_description(self, ups, var): """Get a variable's description.""" - logging.debug("var_description called...") + logger.debug("var_description called...") 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') @@ -354,7 +349,7 @@ def var_description(self, ups, var): def var_type(self, ups, var): """Get a variable's type.""" - logging.debug("var_type called...") + logger.debug("var_type called...") 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') @@ -370,7 +365,7 @@ def var_type(self, ups, var): def command_description(self, ups, command): """Get a command's description.""" - logging.debug("command_description called...") + logger.debug("command_description called...") 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') @@ -382,7 +377,7 @@ def command_description(self, ups, command): def run_command(self, ups, command): """Send a command to the specified UPS.""" - logging.debug("run_command called...") + logger.debug("run_command called...") 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') @@ -391,14 +386,14 @@ def run_command(self, ups, command): def fsd(self, ups): """Send MASTER and FSD commands.""" - logging.debug("MASTER called...") + logger.debug("MASTER called...") 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...") + logger.debug("FSD called...") 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": @@ -408,7 +403,7 @@ def num_logins(self, ups): """Send GET NUMLOGINS command to get the number of users logged into a given UPS. """ - logging.debug("num_logins called on '%s'...", ups) + logger.debug("num_logins called on '%s'...", ups) 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') @@ -420,14 +415,14 @@ def num_logins(self, ups): def help(self): """Send HELP command.""" - logging.debug("HELP called...") + logger.debug("HELP called...") 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...") + logger.debug("VER called...") self._srv_handler.write(b"VER\n") return self._srv_handler.read_until(b"\n", self._timeout).decode('utf-8') diff --git a/tests/test_nutclient.py b/tests/test_nutclient.py index 7216c5b..2f587ea 100644 --- a/tests/test_nutclient.py +++ b/tests/test_nutclient.py @@ -3,14 +3,18 @@ # This source code is provided for testing/debuging purpose ;) -from nut2 import PyNUTClient +import logging import sys +import nut2 +from nut2 import PyNUTClient + if __name__ == "__main__" : + nut2.logger.setLevel(logging.DEBUG) print( "PyNUTClient test..." ) - nut = PyNUTClient( debug=True ) - #nut = PyNUTClient( login="upsadmin", password="upsadmin", debug=True ) + nut = PyNUTClient() + #nut = PyNUTClient( login="upsadmin", password="upsadmin" ) print( 80*"-" + "\nTesting 'GetUPSList' :") result = nut.GetUPSList( ) diff --git a/tests/testclient.py b/tests/testclient.py index d16bf18..7a968e5 100644 --- a/tests/testclient.py +++ b/tests/testclient.py @@ -1,21 +1,24 @@ import unittest from mockserver import MockServer import telnetlib +import logging try: from mock import Mock except ImportError: from unittest.mock import Mock +import nut2 from nut2 import PyNUTClient, PyNUTError class TestClient(unittest.TestCase): def setUp(self): - self.client = PyNUTClient(connect=False, debug=True) + nut2.logger.setLevel(logging.DEBUG) + self.client = PyNUTClient(connect=False) self.client._srv_handler = MockServer(broken=False) - self.broken_client = PyNUTClient(connect=False, debug=True) + self.broken_client = PyNUTClient(connect=False) self.broken_client._srv_handler = MockServer(broken=True) - self.not_ok_client = PyNUTClient(connect=False, debug=True) + self.not_ok_client = PyNUTClient(connect=False) self.not_ok_client._srv_handler = MockServer(ok=False, broken=False) self.valid = "test" @@ -45,7 +48,7 @@ def test_connect(self): def test_connect_debug(self): try: - PyNUTClient(debug=True) + PyNUTClient() except Exception: assert(False) @@ -57,8 +60,7 @@ def test_connect_broken(self): def test_connect_credentials(self): try: - PyNUTClient(login=self.valid, password=self.valid, - debug=True) + PyNUTClient(login=self.valid, password=self.valid) except TypeError: pass except PyNUTError: @@ -69,8 +71,7 @@ def test_connect_credentials(self): def test_connect_credentials_username_ok(self): try: telnetlib.Telnet = MockServer - PyNUTClient(login=self.valid, password=self.valid, - debug=True) + PyNUTClient(login=self.valid, password=self.valid) except TypeError: pass except PyNUTError: @@ -229,7 +230,7 @@ def test_command_description_broken(self): self.valid, self.valid) def test_list_enum(self): - self.assertEquals(self.client.list_enum(self.valid, self.valid), + self.assertEquals(self.client.list_enum(self.valid, self.valid), [self.valid_desc]) def test_list_enum_broken(self): @@ -237,7 +238,7 @@ def test_list_enum_broken(self): self.valid, self.valid) def test_list_range(self): - self.assertEquals(self.client.list_range(self.valid, self.valid), + self.assertEquals(self.client.list_range(self.valid, self.valid), [self.valid_desc]) def test_list_range_broken(self): @@ -245,7 +246,7 @@ def test_list_range_broken(self): self.valid, self.valid) def test_var_type(self): - self.assertEquals(self.client.var_type(self.valid, self.valid), + self.assertEquals(self.client.var_type(self.valid, self.valid), "RW STRING:3") def test_var_type_broken(self):