Commit 155e2cb4 authored by Charles Ferguson's avatar Charles Ferguson
Browse files

Moved all the command declarations to the top.

All the command declarations are now at the top of the file within
a check for the platform. Hopefully this should make it easier to
manange.
parent 92b8b8d0
...@@ -22,10 +22,23 @@ import unittest ...@@ -22,10 +22,23 @@ import unittest
import nose import nose
if True: # A command whose return is true
command_true = 'rem' command_true = None
# A command that generates output once per second
command_output_per_second = None
if sys.platform == 'darwin':
command_true = ['true']
command_output_per_second = ['netstat', '-i', '-w1']
elif sys.platform.startswith('linux'):
command_true = ['true']
command_output_per_second = ['ping', '-c', '4', '127.0.0.1']
elif sys.platform == 'win32':
command_true = ['rem']
command_output_per_second = ['ping', '127.0.0.1']
else: else:
command_true = 'true' raise NotImplementedError
streamedinput = None streamedinput = None
...@@ -79,17 +92,17 @@ class Test01Basics(BaseStreamedInput): ...@@ -79,17 +92,17 @@ class Test01Basics(BaseStreamedInput):
class Test02StartStop(BaseStreamedInput): class Test02StartStop(BaseStreamedInput):
def test_001_start_stop(self): def test_001_start_stop(self):
si = streamedinput.StreamedInput([command_true], shell=True) si = streamedinput.StreamedInput(command_true, shell=True)
si.start() si.start()
si.stop() si.stop()
def test_002_stop(self): def test_002_stop(self):
# Stop is always safe. # Stop is always safe.
si = streamedinput.StreamedInput([command_true], shell=True) si = streamedinput.StreamedInput(command_true, shell=True)
si.stop() si.stop()
def test_003_start_start(self): def test_003_start_start(self):
si = streamedinput.StreamedInput([command_true], shell=True) si = streamedinput.StreamedInput(command_true, shell=True)
si.start() si.start()
with self.assertRaises(streamedinput.StreamedInputCannotStartError): with self.assertRaises(streamedinput.StreamedInputCannotStartError):
si.start() si.start()
...@@ -104,11 +117,11 @@ class Test02StartStop(BaseStreamedInput): ...@@ -104,11 +117,11 @@ class Test02StartStop(BaseStreamedInput):
class Test10DataLength(BaseStreamedInput): class Test10DataLength(BaseStreamedInput):
def test_001_before_start(self): def test_001_before_start(self):
si = streamedinput.StreamedInput([command_true], shell=True) si = streamedinput.StreamedInput(command_true, shell=True)
self.assertEqual(len(si), 0) self.assertEqual(len(si), 0)
def test_002_after_start_empty(self): def test_002_after_start_empty(self):
si = streamedinput.StreamedInput([command_true], shell=True) si = streamedinput.StreamedInput(command_true, shell=True)
si.start() si.start()
self.assertEqual(len(si), 0) self.assertEqual(len(si), 0)
...@@ -127,12 +140,12 @@ class Test10DataLength(BaseStreamedInput): ...@@ -127,12 +140,12 @@ class Test10DataLength(BaseStreamedInput):
class Test20EOF(BaseStreamedInput): class Test20EOF(BaseStreamedInput):
def test_001_before_start(self): def test_001_before_start(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
with self.assertRaises(streamedinput.StreamedInputNotStartedError): with self.assertRaises(streamedinput.StreamedInputNotStartedError):
self.assertFalse(si.eof()) self.assertFalse(si.eof())
def test_002_after_start_empty(self): def test_002_after_start_empty(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
si.start() si.start()
self.assertTrue(si.eof()) self.assertTrue(si.eof())
...@@ -147,12 +160,12 @@ class Test20EOF(BaseStreamedInput): ...@@ -147,12 +160,12 @@ class Test20EOF(BaseStreamedInput):
class Test30Read(BaseStreamedInput): class Test30Read(BaseStreamedInput):
def test_001_before_start(self): def test_001_before_start(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
with self.assertRaises(streamedinput.StreamedInputNotStartedError): with self.assertRaises(streamedinput.StreamedInputNotStartedError):
self.assertIsNone(si.read()) self.assertIsNone(si.read())
def test_002_after_start_empty(self): def test_002_after_start_empty(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
si.start() si.start()
self.assertIsNone(si.read()) self.assertIsNone(si.read())
...@@ -166,7 +179,7 @@ class Test30Read(BaseStreamedInput): ...@@ -166,7 +179,7 @@ class Test30Read(BaseStreamedInput):
class Test31Iterate(BaseStreamedInput): class Test31Iterate(BaseStreamedInput):
def test_001_before_start(self): def test_001_before_start(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
number_iterated = 0 number_iterated = 0
with self.assertRaises(streamedinput.StreamedInputNotStartedError): with self.assertRaises(streamedinput.StreamedInputNotStartedError):
for _ in si: for _ in si:
...@@ -174,7 +187,7 @@ class Test31Iterate(BaseStreamedInput): ...@@ -174,7 +187,7 @@ class Test31Iterate(BaseStreamedInput):
self.assertEqual(number_iterated, 0, "Should have iterated no items") self.assertEqual(number_iterated, 0, "Should have iterated no items")
def test_002_after_start_empty(self): def test_002_after_start_empty(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
si.start() si.start()
number_iterated = 0 number_iterated = 0
for _ in si: for _ in si:
...@@ -269,22 +282,22 @@ class Test42DataFunctionEOF(BaseDataFunction): ...@@ -269,22 +282,22 @@ class Test42DataFunctionEOF(BaseDataFunction):
def test_001_before_start(self): def test_001_before_start(self):
self.data_expects = [] self.data_expects = []
si = streamedinput.StreamedInput([command_true], data_function=self.data_func) si = streamedinput.StreamedInput(command_true, data_function=self.data_func)
with self.assertRaises(streamedinput.StreamedInputNotStartedError): with self.assertRaises(streamedinput.StreamedInputNotStartedError):
self.assertFalse(si.eof()) self.assertTrue(si.eof(), "We should be at the end of the data")
def test_002_after_start_empty(self): def test_002_after_start_empty(self):
self.data_expects = [] self.data_expects = []
si = streamedinput.StreamedInput([command_true], data_function=self.data_func) si = streamedinput.StreamedInput(command_true, data_function=self.data_func)
si.start() si.start()
self.assertTrue(si.eof()) self.assertTrue(si.eof(), "We should be at the end of the data")
def test_003_after_start_1line(self): def test_003_after_start_1line(self):
self.data_expects = ['foo\n'] self.data_expects = ['foo\n']
si = streamedinput.StreamedInput('echo foo', shell=True, data_function=self.data_func) si = streamedinput.StreamedInput('echo foo', shell=True, data_function=self.data_func)
si.start() si.start()
# All the data was already delivered, so we should be at the end. # All the data was already delivered, so we should be at the end.
self.assertTrue(si.eof()) self.assertTrue(si.eof(), "We should be at the end of the data")
class BaseCompleteFunction(BaseStreamedInput): class BaseCompleteFunction(BaseStreamedInput):
...@@ -316,11 +329,11 @@ class BaseCompleteFunction(BaseStreamedInput): ...@@ -316,11 +329,11 @@ class BaseCompleteFunction(BaseStreamedInput):
class Test50CompleteFunction(BaseCompleteFunction): class Test50CompleteFunction(BaseCompleteFunction):
def test_001_not_called_before_start(self): def test_001_not_called_before_start(self):
_ = streamedinput.StreamedInput([command_true], complete_function=self.complete_func) _ = streamedinput.StreamedInput(command_true, complete_function=self.complete_func)
self.assertNoComplete() self.assertNoComplete()
def test_002_called_after_start(self): def test_002_called_after_start(self):
si = streamedinput.StreamedInput([command_true], complete_function=self.complete_func) si = streamedinput.StreamedInput(command_true, complete_function=self.complete_func)
self.complete_expected = True self.complete_expected = True
si.start() si.start()
self.wait() self.wait()
...@@ -331,7 +344,7 @@ class Test80Repr(BaseStreamedInput): ...@@ -331,7 +344,7 @@ class Test80Repr(BaseStreamedInput):
called = None called = None
def test_001_before_start(self): def test_001_before_start(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
self.assertIn('started=False', repr(si)) self.assertIn('started=False', repr(si))
self.assertIn("stopped=n/a", repr(si)) self.assertIn("stopped=n/a", repr(si))
self.assertIn('shell=False', repr(si)) self.assertIn('shell=False', repr(si))
...@@ -339,7 +352,7 @@ class Test80Repr(BaseStreamedInput): ...@@ -339,7 +352,7 @@ class Test80Repr(BaseStreamedInput):
self.assertIn("data_chunks=0", repr(si)) self.assertIn("data_chunks=0", repr(si))
def test_002_started(self): def test_002_started(self):
si = streamedinput.StreamedInput([command_true]) si = streamedinput.StreamedInput(command_true)
si.start() si.start()
self.assertIn('started=True', repr(si)) self.assertIn('started=True', repr(si))
self.assertIn("stopped=stopped", repr(si)) self.assertIn("stopped=stopped", repr(si))
...@@ -351,7 +364,7 @@ class Test80Repr(BaseStreamedInput): ...@@ -351,7 +364,7 @@ class Test80Repr(BaseStreamedInput):
def dummy(_): def dummy(_):
pass pass
si = streamedinput.StreamedInput([command_true], data_function=dummy) si = streamedinput.StreamedInput(command_true, data_function=dummy)
si.start() si.start()
self.assertIn('started=True', repr(si)) self.assertIn('started=True', repr(si))
self.assertIn("stopped=stopped", repr(si)) self.assertIn("stopped=stopped", repr(si))
...@@ -360,7 +373,7 @@ class Test80Repr(BaseStreamedInput): ...@@ -360,7 +373,7 @@ class Test80Repr(BaseStreamedInput):
self.assertNotIn("data_chunks=0", repr(si)) self.assertNotIn("data_chunks=0", repr(si))
def test_004_shell(self): def test_004_shell(self):
si = streamedinput.StreamedInput([command_true], shell=True) si = streamedinput.StreamedInput(command_true, shell=True)
si.start() si.start()
self.assertIn('started=True', repr(si)) self.assertIn('started=True', repr(si))
self.assertIn("stopped=stopped", repr(si)) self.assertIn("stopped=stopped", repr(si))
...@@ -513,21 +526,15 @@ class TestThreaded90ThreadControl(BaseThreadedInput): ...@@ -513,21 +526,15 @@ class TestThreaded90ThreadControl(BaseThreadedInput):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super(TestThreaded90ThreadControl, self).__init__(*args, **kwargs) super(TestThreaded90ThreadControl, self).__init__(*args, **kwargs)
if sys.platform == 'darwin':
self.cmd = ['netstat', '-i', '-w1']
elif sys.platform.startswith('linux'):
self.cmd = ['ping', '-c', '4', '127.0.0.1']
else:
raise NotImplementedError
self.si = None self.si = None
def tearDown(self): def tearDown(self):
self.si.stop() if self.si is not None:
self.si.stop()
super(TestThreaded90ThreadControl, self).tearDown() super(TestThreaded90ThreadControl, self).tearDown()
def test_001_start_and_stop_long_process(self): def test_001_start_and_stop_long_process(self):
self.si = streamedinput.StreamedInput(self.cmd) self.si = streamedinput.StreamedInput(command_output_per_second)
self.si.start() self.si.start()
time.sleep(0.1) time.sleep(0.1)
initial_datalen = len(self.si) initial_datalen = len(self.si)
...@@ -553,7 +560,7 @@ class TestThreaded90ThreadControl(BaseThreadedInput): ...@@ -553,7 +560,7 @@ class TestThreaded90ThreadControl(BaseThreadedInput):
self.assertTrue(self.si.eof(), "We should now be at the end of the list, and the process should be gone") self.assertTrue(self.si.eof(), "We should now be at the end of the list, and the process should be gone")
def test_002_context_handler_stops(self): def test_002_context_handler_stops(self):
with streamedinput.ThreadedStreamedInput(self.cmd) as self.si: with streamedinput.ThreadedStreamedInput(command_output_per_second) as self.si:
test_process = psutil.Process(self.si.proc.pid) test_process = psutil.Process(self.si.proc.pid)
self.assertTrue(self.si.is_running(), "Process should be running") self.assertTrue(self.si.is_running(), "Process should be running")
......
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