commit 8ca6f48505a5cd4d51636a25bd5e68431e89b2d0 Author: Amr Keleg Date: Tue Aug 20 15:43:20 2019 +0200 Avoid code duplication in testing modules Both printtest and proctest use methods to communicate with the underlying command. These methods are common and shouldn't be implemented in both modules. diff --git a/tests/basictest.py b/tests/basictest.py new file mode 100644 index 0000000..0aff6a3 --- /dev/null +++ b/tests/basictest.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +import signal +class Alarm(Exception): + pass + +class BasicTest: + def alarmHandler(self, signum, frame): + raise Alarm + + def withTimeout(self, seconds, cmd, *args, **kwds): + signal.signal(signal.SIGALRM, self.alarmHandler) + signal.alarm(seconds) + ret = cmd(*args, **kwds) + signal.alarm(0) # reset the alarm + return ret + + def communicateFlush(self, string, process): + if string: + process.stdin.write(string.encode('utf-8')) + process.stdin.write(b'\0') + process.stdin.flush() + + output = [] + char = None + try: + char = self.withTimeout(2, process.stdout.read, 1) + except Alarm: + pass + while char and char != b'\0': + output.append(char) + try: + char = self.withTimeout(2, process.stdout.read, 1) + except Alarm: + break # send what we got up till now + + return b"".join(output).decode('utf-8') diff --git a/tests/printtest.py b/tests/printtest.py index ea22e2f..b0fa390 100644 --- a/tests/printtest.py +++ b/tests/printtest.py @@ -4,12 +4,9 @@ import itertools from subprocess import Popen, PIPE, call from tempfile import mkdtemp from shutil import rmtree +from basictest import BasicTest -import signal -class Alarm(Exception): - pass - -class PrintTest(): +class PrintTest(BasicTest): """See lt_print test for how to use this. Override runTest if you don't want to use NUL flushing.""" @@ -18,32 +15,6 @@ class PrintTest(): expectedOutput = itertools.repeat("") expectedRetCodeFail = False - def alarmHandler(self, signum, frame): - raise Alarm - - def withTimeout(self, seconds, cmd, *args, **kwds): - signal.signal(signal.SIGALRM, self.alarmHandler) - signal.alarm(seconds) - ret = cmd(*args, **kwds) - signal.alarm(0) # reset the alarm - return ret - - def communicateFlush(self): - output = [] - char = None - try: - char = self.withTimeout(2, self.printresult.stdout.read, 1) - except Alarm: - pass - while char and char != b'\0': - output.append(char) - try: - char = self.withTimeout(2, self.printresult.stdout.read, 1) - except Alarm: - break # send what we got up till now - - return b"".join(output).decode('utf-8') - def compileTest(self, tmpd): self.assertEqual(0, call(["../lttoolbox/lt-comp", self.printdir, @@ -59,7 +30,7 @@ class PrintTest(): stdout=PIPE, stderr=PIPE) - self.assertEqual(self.communicateFlush(), self.expectedOutput) + self.assertEqual(self.communicateFlush(None, self.printresult), self.expectedOutput) self.printresult.communicate() # let it terminate self.printresult.stdout.close() diff --git a/tests/proctest.py b/tests/proctest.py index dff7969..79dac20 100644 --- a/tests/proctest.py +++ b/tests/proctest.py @@ -4,12 +4,9 @@ import itertools from subprocess import Popen, PIPE, call from tempfile import mkdtemp from shutil import rmtree +from basictest import BasicTest -import signal -class Alarm(Exception): - pass - -class ProcTest(): +class ProcTest(BasicTest): """See lt_proc test for how to use this. Override runTest if you don't want to use NUL flushing.""" @@ -20,36 +17,6 @@ class ProcTest(): expectedOutputs = itertools.repeat("") expectedRetCodeFail = False - def alarmHandler(self, signum, frame): - raise Alarm - - def withTimeout(self, seconds, cmd, *args, **kwds): - signal.signal(signal.SIGALRM, self.alarmHandler) - signal.alarm(seconds) - ret = cmd(*args, **kwds) - signal.alarm(0) # reset the alarm - return ret - - def communicateFlush(self, string): - self.proc.stdin.write(string.encode('utf-8')) - self.proc.stdin.write(b'\0') - self.proc.stdin.flush() - - output = [] - char = None - try: - char = self.withTimeout(2, self.proc.stdout.read, 1) - except Alarm: - pass - while char and char != b'\0': - output.append(char) - try: - char = self.withTimeout(2, self.proc.stdout.read, 1) - except Alarm: - break # send what we got up till now - - return b"".join(output).decode('utf-8') - def compileTest(self, tmpd): self.assertEqual(0, call(["../lttoolbox/lt-comp", self.procdir, @@ -69,7 +36,7 @@ class ProcTest(): self.assertEqual(len(self.inputs), len(self.expectedOutputs)) for inp, exp in zip(self.inputs, self.expectedOutputs): - self.assertEqual(self.communicateFlush(inp+"[][\n]"), + self.assertEqual(self.communicateFlush(inp+"[][\n]", self.proc), exp+"[][\n]") self.proc.communicate() # let it terminate