From e0c5189316840be70c541990ee28b046f936fc7b Mon Sep 17 00:00:00 2001 From: Piyush Sharma Date: Sun, 16 Sep 2018 13:34:22 -0700 Subject: [PATCH] Fix a few PEP8 issues Fix a ChunkedEncodingException being raised while logging in --- dlipower/__init__.py | 2 -- dlipower/dlipower.py | 28 +++++++++++++++------------- scripts/dlipower | 2 +- scripts/fence_dli | 39 +++++++++++++++++++-------------------- test/test_dlipower.py | 33 +++++++++++++++++++++------------ 5 files changed, 56 insertions(+), 48 deletions(-) diff --git a/dlipower/__init__.py b/dlipower/__init__.py index ec3bc95..3a0ff3a 100644 --- a/dlipower/__init__.py +++ b/dlipower/__init__.py @@ -6,7 +6,6 @@ import json import os - __version__ = str('0.0.0') __git_version__ = str("") __git_origin__ = str("") @@ -31,4 +30,3 @@ if __git_origin__.endswith('.git'): # pragma: no cover __git_base_url__ = __git_origin__[:-4].strip('/') __source_url__ = __git_base_url__ + '/tree/' + __git_hash__ - diff --git a/dlipower/dlipower.py b/dlipower/dlipower.py index f0182a5..2cc0c3b 100755 --- a/dlipower/dlipower.py +++ b/dlipower/dlipower.py @@ -99,23 +99,23 @@ """ from __future__ import print_function -from bs4 import BeautifulSoup + import hashlib +import json import logging import multiprocessing import os -import json +import time + import requests import requests.exceptions -import time import urllib3 +from bs4 import BeautifulSoup from six.moves.urllib.parse import quote - logger = logging.getLogger(__name__) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - # Global settings TIMEOUT = 20 RETRIES = 3 @@ -130,7 +130,7 @@ CONFIG_FILE = os.path.expanduser('~/.dlipower.conf') -def _call_it(params): # pragma: no cover +def _call_it(params): # pragma: no cover """indirect caller for instance methods and multiprocessing""" instance, name, args = params kwargs = {} @@ -229,7 +229,7 @@ class PowerSwitch(object): def __init__(self, userid=None, password=None, hostname=None, timeout=None, cycletime=None, retries=None, use_https=False): """ - Class initializaton + Class initialisation """ if not retries: retries = RETRIES @@ -349,8 +349,9 @@ def login(self): headers = {'Content-Type': 'application/x-www-form-urlencoded'} try: - response = self.session.post('%s/login.tgi' % self.base_url, headers=headers, data=data, timeout=self.timeout, verify=False) - except requests.exceptions.ConnectTimeout: + response = self.session.post('%s/login.tgi' % self.base_url, headers=headers, data=data, + timeout=self.timeout, verify=False) + except (requests.exceptions.ConnectTimeout, requests.exceptions.ChunkedEncodingError): self.secure_login = False self.session = None return @@ -359,7 +360,8 @@ def login(self): if 'Set-Cookie' in response.headers: self.secure_login = True - def load_configuration(self): + @staticmethod + def load_configuration(): """ Return a configuration dictionary """ if os.path.isfile(CONFIG_FILE): file_h = open(CONFIG_FILE, 'r') @@ -412,7 +414,8 @@ def geturl(self, url='index.htm'): if self.secure_login and self.session: request = self.session.get(full_url, timeout=self.timeout, verify=False) else: - request = requests.get(full_url, auth=(self.userid, self.password,), timeout=self.timeout, verify=False) + request = requests.get(full_url, auth=(self.userid, self.password,), timeout=self.timeout, + verify=False) except requests.exceptions.RequestException as e: logger.warning("Request timed out - %d retries left.", self.retries - i - 1) logger.exception("Caught exception %s", str(e)) @@ -442,7 +445,6 @@ def determine_outlet(self, outlet=None): except ValueError: raise DLIPowerException('Outlet name \'%s\' unknown' % outlet) - def get_outlet_name(self, outlet=0): """ Return the name of the outlet """ outlet = self.determine_outlet(outlet) @@ -561,7 +563,7 @@ def command_on_outlets(self, command, outlets): result = [ value for value in pool.imap( _call_it, - [(self, command, (outlet, )) for outlet in outlets], + [(self, command, (outlet,)) for outlet in outlets], chunksize=1 ) ] diff --git a/scripts/dlipower b/scripts/dlipower index 3e52f9d..dd23a0d 100755 --- a/scripts/dlipower +++ b/scripts/dlipower @@ -73,7 +73,7 @@ def __command_on_outlets(action, outlet_range): if not options.quiet: print(e, file=sys.stderr) sys.exit(1) - + if __name__ == "__main__": usage = "usage: %prog [options] [status|on|off|cycle|get_outlet_name|set_outlet_name] [range|arg]" diff --git a/scripts/fence_dli b/scripts/fence_dli index 7e7df79..93e813c 100755 --- a/scripts/fence_dli +++ b/scripts/fence_dli @@ -24,44 +24,43 @@ import sys # This fencing driver uses the dlipower python module to manage the power switch # the dlipower package installs a dlipower.py script that provides a command line # interface to manage the switch. -RELEASE_VERSION="0.0.5" +RELEASE_VERSION = "0.0.5" device_opt = [ "help", "version", "agent", - "action", "ipaddr", "login", "passwd","port", "nodename", + "action", "ipaddr", "login", "passwd", "port", "nodename", "timeout", "cycletime" ] - if __name__ == "__main__": # Parse the input into a dict named options - options={} - unknown_options=[] + options = {} + unknown_options = [] for line in sys.stdin.readlines(): - temp=line.strip().split('=') + temp = line.strip().split('=') if len(temp) == 2: - key=temp[0].strip() - value=temp[1].strip() + key = temp[0].strip() + value = temp[1].strip() if line[0] != '#' and len(key): if key in device_opt: - options[key]=value + options[key] = value else: unknown_options.append(key) # Print a warning about unknown options that where passed if len(unknown_options): - print('Unknown options, ignoring: ',' '.join(unknown_options), file=sys.stderr) + print('Unknown options, ignoring: ', ' '.join(unknown_options), file=sys.stderr) # Make sure we got all the needed options to do the action requested - if not set(['ipaddr','login','passwd','action']).issubset(set(options)): + if not set(['ipaddr', 'login', 'passwd', 'action']).issubset(set(options)): print( 'Did not receive all required options, missing', - ' '.join(list(set(['ipaddr','login','passwd','action']).difference(set(options)))), + ' '.join(list(set(['ipaddr', 'login', 'passwd', 'action']).difference(set(options)))), file=sys.stderr ) sys.exit(1) - - if 'port' not in options.keys() and options['action'] in ['off','on','reboot','status']: + + if 'port' not in options.keys() and options['action'] in ['off', 'on', 'reboot', 'status']: print('Cannot execute action, no port specified') sys.exit(1) @@ -72,11 +71,11 @@ if __name__ == "__main__": # user running the command. For clarity it's a good idea to pass the options when # using a cluster. if 'cycletime' not in options.keys(): - options['cycletime']=None - + options['cycletime'] = None + if 'timeout' not in options.keys(): - options['timeout']=None - + options['timeout'] = None + switch = dlipower.powerswitch( hostname=options['ipaddr'], userid=options['login'], @@ -91,12 +90,12 @@ if __name__ == "__main__": if options['action'].lower() in ['reboot']: sys.exit(switch.cycle(int(options['port']))) if options['action'].lower() in ['status']: - status=switch.status(int(options['port'])) + status = switch.status(int(options['port'])) if status == 'ON': sys.exit(0) if status == 'Unknown': sys.exit(1) if status == 'OFF': sys.exit(2) - if options['action'].lower() in ['list','monitor']: + if options['action'].lower() in ['list', 'monitor']: sys.exit(switch.printstatus()) diff --git a/test/test_dlipower.py b/test/test_dlipower.py index c8fe4be..420f51c 100644 --- a/test/test_dlipower.py +++ b/test/test_dlipower.py @@ -13,7 +13,8 @@ class TestDLIPowerPro(VCRTestCase): def setUp(self): """ Set up the mock objects to do our unit tests """ super(TestDLIPowerPro, self).setUp() - self.p = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + self.p = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) def test__dlipower__load_configuration(self): self.p.load_configuration() @@ -34,13 +35,15 @@ def test__dlipower__unicode__name(self): self.assertEqual(result, '%s:%s' % (outlet.name, outlet.state)) def test__dlipower__statuslist(self): - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) result = switch.statuslist() self.assertIsInstance(result, list) self.assertEqual(len(result), 8) def test__dlipower__status(self): - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) result = switch.status(1) self.assertIn(result, ['ON', 'OFF']) @@ -54,28 +57,32 @@ def test__powerswitch_user_password(self): def test_status(self): """ Test the status method of the PowerSwitch object """ - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) switch.off(1) status = switch.status(1) self.assertEqual(status, 'OFF') def test_off(self): """ Test the status method of the PowerSwitch object """ - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) switch.off(1) status = switch.status(1) self.assertEqual(status, 'OFF') def test_on(self): """ Test the status method of the PowerSwitch object """ - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) switch.on(1) status = switch.status(1) self.assertEqual(status, 'ON') def test_cycle(self): """ Test the status method of the PowerSwitch object """ - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) switch.cycle(1) status = switch.status(1) self.assertEqual(status, 'ON') @@ -90,7 +97,8 @@ def test_outlet(self): def test_on_state_setter(self): """ Test the state setter to turn on an outlet """ - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) switch[0].state = "ON" status = switch.status(1) self.assertEqual(status, 'ON') @@ -103,7 +111,8 @@ def test_on_outlet(self): def test_off_state_setter(self): """ Test the state setter to turn off an outlet """ - switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, use_https=self.use_https) + switch = PowerSwitch(hostname=self.switch_hostname, userid=self.userid, password=self.password, + use_https=self.use_https) switch[0].state = "OFF" status = switch.status(1) self.assertEqual(status, 'OFF') @@ -122,11 +131,11 @@ def test_powerswitch_verify(self): self.p.verify() def test_outlet_set_name(self): - self.p[0].name='goober' + self.p[0].name = 'goober' self.assertEqual(self.p.get_outlet_name(1), 'goober') def test_determine_outlet(self): - self.p[0].name='goober' + self.p[0].name = 'goober' self.assertEqual(self.p.determine_outlet('goober'), 1) def test__outlet__unicode__magic(self): @@ -143,7 +152,7 @@ def test__outlet__str__magic(self): def test_command_on_outlets(self): for i in range(0, 5): self.p[i].off() - self.p.command_on_outlets('ON', range(1,6)) + self.p.command_on_outlets('ON', range(1, 6)) for i in range(0, 5): self.assertEqual(self.p[i].state, 'ON')