Commit 27ae5d02 authored by Charles Ferguson's avatar Charles Ferguson
Browse files

Some initial work to try to make the streamed input code work on Windows.

Done some time back.
parent 6538fd3c
......@@ -40,6 +40,7 @@ Example usage
import os
import select
import shlex
import subprocess
import time
import threading
......@@ -187,7 +188,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,11 @@ import unittest
import nose
if True:
command_true = 'rem'
else:
command_true = 'true'
streamedinput = None
# Ensure that the Coverage module is configured to instrument the package we
......@@ -74,17 +79,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 +104,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 +127,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 +147,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 +166,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 +174,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,13 +269,13 @@ 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())
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())
......@@ -311,11 +316,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 +331,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 +339,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 +351,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 +360,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))
......
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