Commit 97a57742 authored by Charles Ferguson's avatar Charles Ferguson
Browse files

Pylint fixes.

Some pylint fixes, including some that appear to have been actual bugs -
the inspect module was never imported, so it's not clear why the test
code worked.
parent 135660ea
......@@ -49,7 +49,6 @@ try:
except ImportError:
# No fcntl module; might be running on non-POSIX system
fcntl = None
pass
try:
# Python 2 has Queue with a capital
......@@ -166,7 +165,7 @@ class StreamedInput(object):
fd = self.stdoutfd
if fcntl:
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) # pylint: disable=no-member
else:
print("Warning: Cannot mark subprocess output as non-blocking")
......@@ -361,7 +360,7 @@ class StreamedInput(object):
try:
return self.data_buffer.get_nowait()
except Queue.Empty:
except queue.Empty:
return None
def is_running(self):
......
......@@ -8,6 +8,9 @@ Class: Functional
Type: Unit test
"""
# Some tests just check that we don't crash.
# pylint: disable=no-self-use
import inspect
import os
import psutil
......@@ -238,7 +241,7 @@ class Test41DataFunctionObjectAPIs(BaseDataFunction):
with self.assertRaises(streamedinput.StreamedInputNotSimpleError):
for _ in si:
self.assertTrue(False, "This should never happen")
self.assertTrue(False, "There should not be items here") # pylint: disable=redundant-unittest-assert
def test_002_read_fails(self):
self.data_expects = ['foo\n']
......@@ -320,6 +323,7 @@ class Test50CompleteFunction(BaseCompleteFunction):
class Test80Repr(BaseStreamedInput):
called = None
def test_001_before_start(self):
si = streamedinput.StreamedInput(['true'])
......@@ -339,7 +343,7 @@ class Test80Repr(BaseStreamedInput):
self.assertIn("data_chunks=0", repr(si))
def test_003_with_datafunc(self):
def dummy(data):
def dummy(_):
pass
si = streamedinput.StreamedInput(['true'], data_function=dummy)
......@@ -362,7 +366,7 @@ class Test80Repr(BaseStreamedInput):
def test_005_running_state(self):
self.called = False
def test_func(data):
def test_func(_):
self.assertIn('started=True', repr(si))
self.assertIn("stopped=running", repr(si))
self.called = True
......@@ -404,7 +408,7 @@ class Test80Repr(BaseStreamedInput):
time.sleep(0.2)
self.assertTrue(si.eof(), "We should be at the end of the stream")
else:
self.assertTrue(False, "WE SHOULD NOT GET HERE")
self.assertTrue(False, "Invalid data received") # pylint: disable=redundant-unittest-assert
self.called = True
si = streamedinput.StreamedInput('echo foo && sleep 1 && echo bar', shell=True, data_function=test_func)
......@@ -428,8 +432,10 @@ class BaseThreadedInput(BaseStreamedInput):
frames = inspect.stack()
try:
caller = frames[1]
# We'll check whether the name of the module we're in matches the
# caller. If so, we're really inside this test code.
if caller[0].f_globals['__name__'] == \
really_call_ThreadedStreamedInput.func_globals['__name__']:
really_call_ThreadedStreamedInput.__globals__['__name__']:
# We're in this test code, so divert to the threaded call
return streamedinput.ThreadedStreamedInput(*args, **kwargs)
else:
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment