Skip to content

Fix a few PEP8 issues #42

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix a few PEP8 issues #42

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions dlipower/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
import json
import os


__version__ = str('0.0.0')
__git_version__ = str("")
__git_origin__ = str("")
Expand All @@ -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__

28 changes: 15 additions & 13 deletions dlipower/dlipower.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 = {}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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')
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
)
]
Expand Down
2 changes: 1 addition & 1 deletion scripts/dlipower
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]"
Expand Down
39 changes: 19 additions & 20 deletions scripts/fence_dli
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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'],
Expand All @@ -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())
33 changes: 21 additions & 12 deletions test/test_dlipower.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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'])

Expand All @@ -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')
Expand All @@ -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')
Expand All @@ -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')
Expand All @@ -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):
Expand All @@ -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')

Expand Down