Commit 570dd496 authored by Charles Ferguson's avatar Charles Ferguson
Browse files

Merge branch 'windows-support-work' into 'master'

Windows support work

See merge request !1
parents 706a3dfe 78545200
......@@ -41,6 +41,7 @@ Example usage
import errno
import os
import select
import shlex
import subprocess
import time
import threading
......@@ -189,7 +190,12 @@ class StreamedInput(object):
try:
stderr = subprocess.STDOUT if self.keep_stderr else open(os.devnull, 'w')
self.proc = subprocess.Popen(self.command, shell=self.shell,
command = self.command
if os.name == 'nt':
if isinstance(command, basestring):
command = shlex.split(command)
self.proc = subprocess.Popen(command, shell=self.shell,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=stderr,
bufsize=0)
......
......@@ -22,6 +22,24 @@ import unittest
import nose
# A command whose return is true
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:
raise NotImplementedError
streamedinput = None
# Ensure that the Coverage module is configured to instrument the package we
......@@ -74,17 +92,17 @@ class Test01Basics(BaseStreamedInput):
class Test02StartStop(BaseStreamedInput):
def test_001_start_stop(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true, shell=True)
si.start()
si.stop()
def test_002_stop(self):
# Stop is always safe.
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true, shell=True)
si.stop()
def test_003_start_start(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true, shell=True)
si.start()
with self.assertRaises(streamedinput.StreamedInputCannotStartError):
si.start()
......@@ -99,16 +117,16 @@ class Test02StartStop(BaseStreamedInput):
class Test10DataLength(BaseStreamedInput):
def test_001_before_start(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true, shell=True)
self.assertEqual(len(si), 0)
def test_002_after_start_empty(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true, shell=True)
si.start()
self.assertEqual(len(si), 0)
def test_003_after_start_1line(self):
si = streamedinput.StreamedInput(['echo', 'foo'])
si = streamedinput.StreamedInput(['echo', 'foo'], shell=True)
si.start()
self.assertEqual(len(si), 1)
......@@ -122,12 +140,12 @@ class Test10DataLength(BaseStreamedInput):
class Test20EOF(BaseStreamedInput):
def test_001_before_start(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
with self.assertRaises(streamedinput.StreamedInputNotStartedError):
self.assertFalse(si.eof())
def test_002_after_start_empty(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
si.start()
self.assertTrue(si.eof())
......@@ -142,12 +160,12 @@ class Test20EOF(BaseStreamedInput):
class Test30Read(BaseStreamedInput):
def test_001_before_start(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
with self.assertRaises(streamedinput.StreamedInputNotStartedError):
self.assertIsNone(si.read())
def test_002_after_start_empty(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
si.start()
self.assertIsNone(si.read())
......@@ -161,7 +179,7 @@ class Test30Read(BaseStreamedInput):
class Test31Iterate(BaseStreamedInput):
def test_001_before_start(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
number_iterated = 0
with self.assertRaises(streamedinput.StreamedInputNotStartedError):
for _ in si:
......@@ -169,7 +187,7 @@ class Test31Iterate(BaseStreamedInput):
self.assertEqual(number_iterated, 0, "Should have iterated no items")
def test_002_after_start_empty(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
si.start()
number_iterated = 0
for _ in si:
......@@ -264,22 +282,22 @@ class Test42DataFunctionEOF(BaseDataFunction):
def test_001_before_start(self):
self.data_expects = []
si = streamedinput.StreamedInput(['true'], data_function=self.data_func)
si = streamedinput.StreamedInput(command_true, data_function=self.data_func)
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):
self.data_expects = []
si = streamedinput.StreamedInput(['true'], data_function=self.data_func)
si = streamedinput.StreamedInput(command_true, data_function=self.data_func)
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):
self.data_expects = ['foo\n']
si = streamedinput.StreamedInput('echo foo', shell=True, data_function=self.data_func)
si.start()
# 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):
......@@ -311,11 +329,11 @@ class BaseCompleteFunction(BaseStreamedInput):
class Test50CompleteFunction(BaseCompleteFunction):
def test_001_not_called_before_start(self):
_ = streamedinput.StreamedInput(['true'], complete_function=self.complete_func)
_ = streamedinput.StreamedInput(command_true, complete_function=self.complete_func)
self.assertNoComplete()
def test_002_called_after_start(self):
si = streamedinput.StreamedInput(['true'], complete_function=self.complete_func)
si = streamedinput.StreamedInput(command_true, complete_function=self.complete_func)
self.complete_expected = True
si.start()
self.wait()
......@@ -326,7 +344,7 @@ class Test80Repr(BaseStreamedInput):
called = None
def test_001_before_start(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
self.assertIn('started=False', repr(si))
self.assertIn("stopped=n/a", repr(si))
self.assertIn('shell=False', repr(si))
......@@ -334,7 +352,7 @@ class Test80Repr(BaseStreamedInput):
self.assertIn("data_chunks=0", repr(si))
def test_002_started(self):
si = streamedinput.StreamedInput(['true'])
si = streamedinput.StreamedInput(command_true)
si.start()
self.assertIn('started=True', repr(si))
self.assertIn("stopped=stopped", repr(si))
......@@ -346,7 +364,7 @@ class Test80Repr(BaseStreamedInput):
def dummy(_):
pass
si = streamedinput.StreamedInput(['true'], data_function=dummy)
si = streamedinput.StreamedInput(command_true, data_function=dummy)
si.start()
self.assertIn('started=True', repr(si))
self.assertIn("stopped=stopped", repr(si))
......@@ -355,7 +373,7 @@ class Test80Repr(BaseStreamedInput):
self.assertNotIn("data_chunks=0", repr(si))
def test_004_shell(self):
si = streamedinput.StreamedInput(['true'], shell=True)
si = streamedinput.StreamedInput(command_true, shell=True)
si.start()
self.assertIn('started=True', repr(si))
self.assertIn("stopped=stopped", repr(si))
......@@ -508,21 +526,15 @@ class TestThreaded90ThreadControl(BaseThreadedInput):
def __init__(self, *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
def tearDown(self):
if self.si is not None:
self.si.stop()
super(TestThreaded90ThreadControl, self).tearDown()
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()
time.sleep(0.1)
initial_datalen = len(self.si)
......@@ -548,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")
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)
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