diff --git a/bin/doxec b/bin/doxec
index c421d7a1495b0d57e71241f7f977f68f45deb893..81fc64e97a7a86d551ace23aa0f06b1dc2bf2379 100755
--- a/bin/doxec
+++ b/bin/doxec
@@ -18,6 +18,9 @@ if __name__ == "__main__":
parser.add_argument("--version", action="store_true",
help="Prints the version of the doxec package and exits.")
+ parser.add_argument("--short", action="store_true",
+ help="Suppresses the standard output of operations.")
+
parser.add_argument("documents", metavar="DOCUMENT", nargs="+", default=[],
help="A document from which the code examples should be parsed and "
"executed")
@@ -30,16 +33,18 @@ if __name__ == "__main__":
parser = doxec.parser[args.syntax]
- def monitor(op, exception):
- print("running %s(%s) ... " % (op.command, op.args))
- if exception is not None:
- print(chr(27) + "[31m", end="")
- print(exception, end="")
- print(chr(27) + "[0m")
+ print("Copyright (c) 2017 Frank Sauerburger")
- fail_count = 0
for doc_path in args.documents:
doc = doxec.Document(doc_path, syntax=parser)
- fail_count += not doc.run(monitor=monitor)
+ monitor = doxec.Monitor(doc_path, short=args.short)
+ doc.run(monitor=monitor)
+
+
+ print("-"*80)
+ color = "\033[31m" if monitor.fail_count > 0 else "\033[32m"
+
+ print(color + "Failed: %5d\033[0m" % monitor.fail_count)
+ print("Total: %5d" % monitor.total_count)
- sys.exit(fail_count)
+ sys.exit(monitor.fail_count)
diff --git a/doxec/__init__.py b/doxec/__init__.py
index 29eca7f14bc65c99801585ac8c1245ca199102c1..1792ae586beb0f6b6b35ad09580072710257af8c 100644
--- a/doxec/__init__.py
+++ b/doxec/__init__.py
@@ -1,6 +1,7 @@
import abc
import re
+import os
import subprocess
__version__ = "0.1.1"
@@ -21,7 +22,7 @@ class Operation(metaclass=abc.ABCMeta):
"""
op_store = []
- def __init__(self, args, content):
+ def __init__(self, args, content, line=None):
"""
Creates an operation. The args argument is the token after the
operation key word, content is the markdown block after the operation
@@ -41,9 +42,15 @@ class Operation(metaclass=abc.ABCMeta):
return cls(args, content)
@abc.abstractmethod
- def execute(self):
+ def execute(self, log=None):
"""
- Performs the operation represented by this object.
+ Performs the operation represented by this object. The optional
+ function is called, when the operation produces output. A list of
+ lines is passed to the log function. The log function might be called
+ multiple times.
+
+ An exception is thrown, when the method encounters problems, or a test
+ fails.
"""
pass
@@ -65,39 +72,57 @@ class OpWrite(Operation):
"""
command = "write"
- def execute(self):
+ def execute(self, log=None):
with open(self.args, "w") as f:
for line in self.content:
print(line, file=f)
+ def __str__(self):
+ return "%s(%s)" % (self.command, self.args)
+
class OpAppend(Operation):
"""
This operation performs a 'append to file' operation.
"""
command = "append"
- def execute(self):
+ def execute(self, log=None):
with open(self.args, "a") as f:
for line in self.content:
print(line, file=f)
+ def __str__(self):
+ return "%s(%s)" % (self.command, self.args)
+
class OpConsole(Operation):
"""
- This operation runs all lines starting with $ in the console. The
- operation raises an error, if the return code is not zero.
+ This operation runs all lines starting with $ in the console. All other
+ lines are ignored. The operation raises an error, if the return code is
+ not zero.
"""
command = "console"
- def execute(self):
- job = subprocess.Popen("/bin/bash", stdin=subprocess.PIPE,
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- script = "\n".join([l[1:] for l in self.content if l.startswith("$")])
- (stdoutdata, stderrdata) = job.communicate(script.encode('utf8'))
- if job.returncode != 0:
- raise TestException("Script failed with %d:" % job.returncode,
- stdoutdata.decode('utf8'), stderrdata.decode('utf8'))
- return stdoutdata
+ def execute(self, log=None):
+ if log is None:
+ log = lambda x: None
+
+ script = [l[1:].lstrip() for l in self.content if l.startswith("$")]
+ for command in script:
+ log(["$ %s" % command])
+ job = subprocess.Popen("/bin/bash", stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ (stdoutdata, stderrdata) = job.communicate(command.encode('utf8'))
+ output = stdoutdata.decode('utf8')
+ output = re.split(r'\r?\n', output)
+ if len(output) > 0 and output[-1] == '':
+ del output[-1]
+ log(output)
+ if job.returncode != 0:
+ raise TestException("Script failed with return code %d:" % job.returncode,
+ stdoutdata.decode('utf8'), stderrdata.decode('utf8'))
+ def __str__(self):
+ return "%s" % self.command
class OpConsoleOutput(Operation):
@@ -107,11 +132,14 @@ class OpConsoleOutput(Operation):
"""
command = "console_output"
- def execute(self):
+ def execute(self, log=None):
+ if log is None:
+ log = lambda x: None
+
commands = [] # items are (command, [output lines])
for line in self.content:
if line.startswith("$"):
- commands.append((line[1:], []))
+ commands.append((line[1:].lstrip(), []))
elif len(commands) == 0:
# no command yet
continue
@@ -119,19 +147,29 @@ class OpConsoleOutput(Operation):
commands[-1][1].append(line)
for command, lines in commands:
+ log(["$ %s" % command])
job = subprocess.Popen("/bin/bash", stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(stdoutdata, stderrdata) = job.communicate(command.encode('utf8'))
if job.returncode != 0:
- raise TestException("Script failed with %d:" % job.returncode,
+ raise TestException("Script failed with return code %d:" % job.returncode,
stdoutdata.decode('utf8'), stderrdata.decode('utf8'))
output = stdoutdata.decode('utf8')
output = re.split(r'\r?\n', output)
if len(output) > 0 and output[-1] == '':
del output[-1]
+ log(output)
if lines != output:
- raise TestException("Output differs", lines, output)
+ first_offending = None
+ for l, o in zip(lines, output):
+ if l != o:
+ first_offending = "First mismatch: %s != %s " % (repr(l), repr(o))
+ break
+ raise TestException("Output differs", first_offending,
+ "Expected: %s" % repr(lines), "Actual: %s" % repr(output))
+ def __str__(self):
+ return "%s" % self.command
# add operations to op_store
Operation.op_store.append(OpConsole)
@@ -156,9 +194,10 @@ class DoxecSyntax(metaclass=abc.ABCMeta):
the command content. These parts are also removed from the line list.
The method modifies the list in place.
- The return value is a triplet of (command, argument, content). None is
- returned, if no valid magic tag could be found. The content item is a
- list of code lines.
+ The return value is a tuple of (command, argument, content, length).
+ None is returned, if no valid magic tag could be found. The content
+ item is a list of code lines. The length is the number of lines, which
+ belong to this operation.
"""
pass
@@ -184,12 +223,15 @@ class Markdown(DoxecSyntax):
See DoxecSyntax.
"""
while len(lines) > 0:
+ start_line_count = len(lines)
cmd = Markdown.parse_command(lines[0])
del lines[0]
if cmd is not None:
code = Markdown.parse_code(lines)
if code is not None:
- return cmd[0], cmd[1], code
+ end_line_count = len(lines)
+ length = start_line_count - end_line_count
+ return cmd[0], cmd[1], code, length
return None
@@ -282,50 +324,137 @@ class Document:
operations within the object are executed.
"""
- def __init__(self, document_path, syntax=Markdown):
+ def __init__(self, document_path, syntax):
"""
Creates a new document object and parses the given document.
+ Parses the given file and appends all operations defined in the
+ string to the internal storage. The syntax class is used to parse the
+ file.
"""
- self.operations = []
+ self.operations = [] # each item is a tuple of (line, operation)
with open(document_path) as f:
- self.parse(f.read(), syntax)
-
- def parse(self, string, syntax):
- """
- Parses the content string and appends all operations defined in the
- string to the internal storage.
- """
- lines = re.split("\r?\n", string)
- while len(lines) > 0:
- op_tuple = syntax.parse(lines)
- if op_tuple is None:
- continue
- op_obj = Operation.factory(*op_tuple)
- if op_obj is None:
- continue
- self.operations.append(op_obj)
+ lines = re.split("\r?\n", f.read())
+ line_count = len(lines)
+ while len(lines) > 0:
+ op_tuple = syntax.parse(lines)
+ if op_tuple is None:
+ continue
+ command, args, content, length = op_tuple
+ op_obj = Operation.factory(command, args, content)
+ if op_obj is None:
+ continue
+ line_number = line_count - len(lines) + 1 - length
+ self.operations.append((line_number, op_obj))
def run(self, monitor=None):
"""
Runs all operations stored attached to this object. The monitor object
- is called after every iteration. The first argument of monitor is the
- operation object, the second argument is None or the exception that
- occurred.
+ is might be called before, during and after the execution of each
+ operation.
- The method returns True on success.
+ The method returns a tuple with the number of failed operations and
+ the total number of operations.
"""
- success = True
- for op in self.operations:
+ fail_count = 0
+ total_operations = 0
+
+ # set defaults
+ before_method = lambda l, o: None
+ log_method = lambda ls: None
+ after_method = lambda e=None: None
+
+ # overwrite defaults if monitor is given
+ if monitor is not None and isinstance(monitor, Monitor):
+ before_method = monitor.before_execute
+ log_method = monitor.log
+ after_method = monitor.after_execute
+
+ for line, op in self.operations:
+ before_method(line, op)
+ total_operations += 1
try:
- op.execute()
+ op.execute(log=log_method)
except TestException as e:
- success = False
- if callable(monitor):
- monitor(op, e)
- else:
- raise e
+ after_method(e)
+ fail_count += 1
+ else:
+ after_method()
+
+ return (fail_count, total_operations)
+
+
+class Monitor(metaclass=abc.ABCMeta):
+ """
+ This class is the default monitor which displays the results of
+ Document.run in a user friendly way. The monitor object also keeps track
+ of a fail and total counter.
+ """
+
+ def __init__(self, path, short=False):
+ """
+ Initializes the monitor. A new monitor should be used for each file.
+ The short argument, defines whether the standard output of operations
+ should be displayed.
+ """
+
+ self.full_path = os.path.abspath(path)
+ self.short = short
+
+ self.first_line = False
+
+ self.fail_count = 0
+ self.total_count = 0
+
+ def before_execute(self, line, operation):
+ """
+ This method is should before the operations execute method is called.
+ This method set the internal values and caches the lines and the
+ operation.
+ """
+ print("\033[2K\033[0G\033[33m%s:%-5d %s ... \033[39;49m" % (self.full_path, line, operation), end="")
+ self.pending_line_break = True
+ self.line = line
+ self.operation = operation
+
+ def after_execute(self, error=None):
+ """
+ This method should be called after the operations execute method is
+ called The optional parameter error is an exception, if one occurred
+ during execution. This method uses the cached values for line and
+ operation provided to before_execute.
+ """
+ self.total_count += 1
+ if error is None:
+ print("\033[2K\033[0G\033[32m%s:%-5d %s ... done\033[39;49m" % (self.full_path, self.line, self.operation))
+ else:
+ self.fail_count += 1
+ if self.pending_line_break:
+ print()
+ self.pending_line_break = False
+
+ if error.args is not None:
+ args = error.args
else:
- if callable(monitor):
- monitor(op, None)
- return success
+ args = [str(error)]
+ for error_line in args:
+ print("\033[31m%s" % str(error_line))
+ print("\033[31m%s:%-5d %s ... failed\033[39;49m" % (self.full_path, self.line, self.operation))
+
+ def log(self, lines):
+ """
+ This method should be called during the execution of a method. This
+ method can be called multiple times, or not at all. The given lines
+ should be printed to the terminal. This method might uses the cached
+ values for line and operation provided to before_execute.
+ """
+ if self.short:
+ return
+
+ if self.pending_line_break:
+ print()
+ self.pending_line_break = False
+
+ for line in lines:
+ print("--- %s" % line)
+
diff --git a/doxec/tests/document.py b/doxec/tests/document.py
index e0ac317e2be352e0d8d7175a168952d778ff9d3c..c825f2db8c705e69a73838d349679898af800e27 100644
--- a/doxec/tests/document.py
+++ b/doxec/tests/document.py
@@ -1,13 +1,13 @@
import os
import unittest
+from unittest.mock import MagicMock
from tempfile import NamedTemporaryFile
-from doxec import Document, OpWrite, OpConsoleOutput
+from doxec import Document, OpWrite, OpConsoleOutput, Markdown, Monitor
-toy_doc = """
-# Installation
+toy_doc = """# Installation
In order to run the code examples in this repository you need `python3` and
the
@@ -25,6 +25,7 @@ apt-get install python3 python3-numpy python3-scipy python3-matplotlib
As a first example we can compute the square root of 2. Create a
file named `example.py` with the following content.
+# This is line 19
<!-- write example.py -->
```python
import math
@@ -34,7 +35,7 @@ print("Square root of 2 = %g" % math.sqrt(2))
```
The file can be run with the `python3` interpreter.
-
+# This is line 29
<!-- console_output -->
```bash
$ python3 example.py
@@ -43,6 +44,24 @@ Square root of 2 = 1.41421
```
"""
+toy_doc2 = """
+Hello this is an other toy document.
+<!-- console_output -->
+```bash
+$ echo 123
+1234
+```
+"""
+
+toy_doc3 = """
+Hello this is an other toy document.
+<!-- console_output -->
+```bash
+$ echo 123
+123
+```
+"""
+
class DocumentTestCase(unittest.TestCase):
"""
This class tests the methods provided by the document class.
@@ -50,18 +69,24 @@ class DocumentTestCase(unittest.TestCase):
def test_parse(self):
"""
- Test whether the parse() correctly parses all operations in the toy
- file.
+ Test whether the init correctly parses all operations with correct
+ line numbers in the toy file.
"""
tmp = NamedTemporaryFile(delete=False)
tmp.write(toy_doc.encode('utf8'))
tmp.close()
- doc = Document(tmp.name)
+ doc = Document(tmp.name, Markdown)
self.assertEqual(len(doc.operations), 2)
- self.assertIsInstance(doc.operations[0], OpWrite)
- self.assertIsInstance(doc.operations[1], OpConsoleOutput)
+
+ line, op = doc.operations[0]
+ self.assertIsInstance(op, OpWrite)
+ self.assertEqual(line, 20)
+
+ line, op = doc.operations[1]
+ self.assertIsInstance(op, OpConsoleOutput)
+ self.assertEqual(line, 30)
os.remove(tmp.name)
@@ -73,9 +98,81 @@ class DocumentTestCase(unittest.TestCase):
tmp.write(toy_doc.encode('utf8'))
tmp.close()
- doc = Document(tmp.name)
+ doc = Document(tmp.name, Markdown)
- doc.run()
+ fail, total = doc.run()
+ self.assertEqual(fail, 0)
+ self.assertEqual(total, 2)
os.remove(tmp.name)
+
+ def test_run_monitor_with_error(self):
+ """
+ Check that run() calls the monitor.
+ """
+ tmp = NamedTemporaryFile(delete=False)
+ tmp.write(toy_doc2.encode('utf8'))
+ tmp.close()
+
+ monitor = Monitor(tmp.name)
+ monitor.before_execute = MagicMock()
+ monitor.log = MagicMock()
+ monitor.after_execute = MagicMock()
+
+ doc = Document(tmp.name, Markdown)
+
+ fail, total = doc.run(monitor=monitor)
+ self.assertEqual(fail, 1)
+ self.assertEqual(total, 1)
+
+ os.remove(tmp.name)
+
+
+ self.assertEqual(monitor.before_execute.call_count, 1)
+ self.assertEqual(monitor.before_execute.call_args[0][0], 3)
+ self.assertIsInstance(monitor.before_execute.call_args[0][1],
+ OpConsoleOutput)
+
+ # first call with $ echo 123
+ self.assertEqual(monitor.log.call_count, 2)
+ self.assertEqual(monitor.log.call_args[0][0], ["123"])
+
+ self.assertEqual(monitor.after_execute.call_count, 1)
+ self.assertIsInstance(monitor.after_execute.call_args[0][0], Exception)
+
+
+
+ def test_run_monitor_wo_error(self):
+ """
+ Check that run() calls the monitor.
+ """
+ tmp = NamedTemporaryFile(delete=False)
+ tmp.write(toy_doc3.encode('utf8'))
+ tmp.close()
+
+ monitor = Monitor(tmp.name)
+ monitor.before_execute = MagicMock()
+ monitor.log = MagicMock()
+ monitor.after_execute = MagicMock()
+
+ doc = Document(tmp.name, Markdown)
+
+ fail, total = doc.run(monitor=monitor)
+ self.assertEqual(fail, 0)
+ self.assertEqual(total, 1)
+
+ os.remove(tmp.name)
+
+ self.assertEqual(monitor.before_execute.call_count, 1)
+ self.assertEqual(monitor.before_execute.call_args[0][0], 3)
+ self.assertIsInstance(monitor.before_execute.call_args[0][1],
+ OpConsoleOutput)
+
+ # first call with $ echo 123
+ self.assertEqual(monitor.log.call_count, 2)
+ self.assertEqual(monitor.log.call_args[0][0], ["123"])
+
+ self.assertEqual(monitor.after_execute.call_count, 1)
+ self.assertEqual(monitor.after_execute.call_args[0], ())
+
diff --git a/doxec/tests/markdown.py b/doxec/tests/markdown.py
index 6e31b87bebbd4b3d3ff31cc6b250488e8909d5e4..84aa8ba5f33ae18b5e6915c2960d6814ba94377a 100644
--- a/doxec/tests/markdown.py
+++ b/doxec/tests/markdown.py
@@ -256,11 +256,12 @@ class MarkdownSyntaxTestCase(unittest.TestCase):
doc.append("```")
retval = Markdown.parse(doc)
- self.assertEqual(len(retval), 3)
- command, args, content = retval
+ self.assertEqual(len(retval), 4)
+ command, args, content, length = retval
self.assertEqual(command, "APPEND")
self.assertEqual(args, "/dev/null")
self.assertEqual(content, ["touch /tmp", "touch /home"])
+ self.assertEqual(length, 5)
def test_parse_valid_additional_lines(self):
"""
@@ -278,11 +279,12 @@ class MarkdownSyntaxTestCase(unittest.TestCase):
doc.append("This caused a seg fault?")
retval = Markdown.parse(doc)
- self.assertEqual(len(retval), 3)
- command, args, content = retval
+ self.assertEqual(len(retval), 4)
+ command, args, content, length = retval
self.assertEqual(command, "APPEND")
self.assertEqual(args, "/dev/null")
self.assertEqual(content, ["touch /tmp", "touch /home"])
+ self.assertEqual(length, 5)
self.assertEqual(doc, ["This caused a seg fault?"])
diff --git a/doxec/tests/operation.py b/doxec/tests/operation.py
index c1fa108b80c60683daa6ec2b38e823d0d114efac..3789d8951813c6135af9449b68b60bfc86508809 100644
--- a/doxec/tests/operation.py
+++ b/doxec/tests/operation.py
@@ -103,6 +103,13 @@ class OpWriteTestCase(unittest.TestCase):
os.remove(tmp_path)
+ def test_str(self):
+ """
+ Test string representation.
+ """
+ op = OpWrite("/path/to/file", ["content", "lines"])
+ self.assertEqual(str(op), "write(/path/to/file)")
+
class OpAppendTestCase(unittest.TestCase):
"""
@@ -131,6 +138,13 @@ class OpAppendTestCase(unittest.TestCase):
os.remove(tmp_path)
+ def test_str(self):
+ """
+ Test string representation.
+ """
+ op = OpAppend("/path/to/file", ["content", "lines"])
+ self.assertEqual(str(op), "append(/path/to/file)")
+
class OpConsoleTestCase(unittest.TestCase):
"""
Test the functionality of the console-operation. This test case focuses on
@@ -164,6 +178,46 @@ class OpConsoleTestCase(unittest.TestCase):
console = OpConsole(None, ["$ exit 1"])
self.assertRaises(TestException, console.execute)
+ def test_execute_ignore(self):
+ """
+ Create a OpConsole operation and check that it ignores lines without
+ $.
+ """
+ console = OpConsole(None, ["exit 1"])
+ console.execute()
+
+ def test_str(self):
+ """
+ Test string representation.
+ """
+ op = OpConsole(None, ["$ command", "ignore"])
+ self.assertEqual(str(op), "console")
+
+ def test_log(self):
+ """
+ Ensure that the console operation calls the log function, when output is
+ produced.
+ """
+ passed_lines = []
+ def log_function(lines):
+ passed_lines.extend(lines)
+
+ op = OpConsole(None, ["$ echo 'Hey\nYou!'", "$ echo 'Hello You!'"])
+ op.execute(log_function)
+
+ self.assertEqual(passed_lines,
+ ["$ echo 'Hey\nYou!'", "Hey", "You!", "$ echo 'Hello You!'", "Hello You!"])
+
+
+ def test_no_log(self):
+ """
+ Ensure that the console operation does not crash if the command
+ produces output, but no log function is present.
+ """
+ op = OpConsole(None, ["$ echo 'Hey\nYou!'", "$ echo 'Hello You!'"])
+ op.execute()
+
+
class OpConsoleOutputTestCase(unittest.TestCase):
"""
Test the functionality of the console-output-operation. This test case focuses on
@@ -195,3 +249,33 @@ class OpConsoleOutputTestCase(unittest.TestCase):
"""
console = OpConsoleOutput(None, ["heyho", "$ echo 'Hi'", "Hi"])
console.execute()
+
+ def test_str(self):
+ """
+ Test string representation.
+ """
+ op = OpConsoleOutput(None, ["$ command", "ignore"])
+ self.assertEqual(str(op), "console_output")
+
+ def test_log(self):
+ """
+ Ensure that the console_output operation calls the log function, when output is
+ produced.
+ """
+ passed_lines = []
+ def log_function(lines):
+ passed_lines.extend(lines)
+
+ block =["$ echo 'Hey\nYou!'", "Hey", "You!", "$ echo 'Hello You!'", "Hello You!"]
+ op = OpConsoleOutput(None, block)
+ op.execute(log_function)
+
+ self.assertEqual(passed_lines, block)
+
+ def test_no_log(self):
+ """
+ Ensure that the console operation does not crash if the command
+ produces output, but no log function is present.
+ """
+ op = OpConsole(None, ["$ echo 'Hey\nYou!'", "$ echo 'Hello You!'"])
+ op.execute()