Add tests for output for context.output.changed. Writer is 100% teste…

…d. Test coverage is 92% (232 missing out of 2776).

lena/output/writer.py

@@ -211,6 +211,10 @@ def should_be_written(data, context):
                     # size = maximum size (bytes) to read
                     # negative or omitted means whole content
                     size = -1
+                    # security warning: say an adversary creates
+                    # a huge file and adds its path to context (!).
+                    # Then the program will crash
+                    # trying to read that in memory.
                     existing_data = fil.read(size)
                 if data != existing_data:
                     self._write_data(filepath, data)
@@ -236,10 +240,11 @@ def _write_data(self, filepath, data):
         # write output to filesystem
         # todo: allow binary files
         with open(filepath, "w") as fil:
-            try:
-                fil.write(data)
-            except TypeError:
-                raise lena.core.LenaTypeError(
-                    "can't write data {} to file {}"
-                    .format(data, filepath)
-                )
+            fil.write(data)
+            # we write only strings, so no TypeError is expected.
+            # If one occurs, the user will note that.
+            # except TypeError:
+            #     raise lena.core.LenaTypeError(
+            #         "can't write data {} to file {}"
+            #         .format(data, filepath)
+            #     )

tests/output/test_pdf_to_png.py

@@ -10,9 +10,9 @@
 
 
 def test_pdf_to_png(mocker):
-    mocker.patch('subprocess.Popen.communicate', return_value=("stdout", "stderr"))
-    mocker.patch('subprocess.Popen.returncode', return_value=True, create=True)
-    mocker.patch('subprocess.Popen', return_value=subprocess.Popen)
+    mocker.patch("subprocess.Popen.communicate", return_value=("stdout", "stderr"))
+    mocker.patch("subprocess.Popen.returncode", return_value=True, create=True)
+    mocker.patch("subprocess.Popen", return_value=subprocess.Popen)
 
     pdf_to_png = PDFToPNG()
     data = [
@@ -23,10 +23,26 @@ def test_pdf_to_png(mocker):
     assert res ==  [
         ('output/file.csv', {'output': {'filename': 'y', 'filetype': 'csv'}}),
         # since no png file exists,
-        # mocker imitates creation of a new one, thus changed=True
+        # mocker imitates creation of a new one, thus changed is True
         ('output/file.png', {'output': {'changed': True,
                                         'filename': 'y', 'filetype': 'png'}})
     ]
 
     command = ['pdftoppm', 'output/file.pdf', 'output/file', '-png', '-singlefile']
     subprocess.Popen.assert_called_once_with(command)
+
+    # test "existing" png
+    mocker.patch("subprocess.Popen", return_value=subprocess.Popen)
+    def _os_path_exists(filepath):
+        return filepath == "output/file.png"
+    mocker.patch("os.path.exists", _os_path_exists)
+    pdf_data = [("output/file.pdf", {"output": {"filename": "y", "filetype": "pdf"}})]
+    assert list(pdf_to_png.run(pdf_data)) == [
+        ('output/file.png',
+         {'output': 
+             {'changed': False,
+              'filename': 'y',
+              'filetype': 'png'}})
+    ]
+    # command was not called
+    assert not subprocess.Popen.called

tests/output/test_writer.py

@@ -1,8 +1,10 @@
 from __future__ import print_function
 
+import copy
 import os
 import pytest
 import sys
+import warnings
 
 import lena.core
 from lena.output import Writer
@@ -36,14 +38,20 @@ def test_writer():
         Writer(1)
     with pytest.raises(lena.core.LenaTypeError):
         Writer(output_filename=1)
+    with pytest.raises(lena.core.LenaValueError):
+        Writer(existing_unchanged=True, overwrite=True)
     w1 = Writer()
     # unwritten values pass unchanged
     indata = [1, 2, 
               (3, {}),
+              # note that context.output is not a dict!
               (4, {"output": ("writer", True)}),
-              (5, {"output": {"writer": False}}),
+              ("5", {"output": {"writer": False}}),
+              # only here data part is checked
+              (6, {"output": {"writer": True}}),
              ]
     assert list(w1.run(indata)) == indata
+    # empty filename is prohibited
     with pytest.raises(lena.core.LenaRuntimeError):
         list(w1.run([("0", {"output": {"filename": ""}})]))
 
@@ -54,7 +62,9 @@ def test_writer_writes(mocker):
         mocker.patch("__builtin__.open", m)
     else:
         mocker.patch("builtins.open", m)
+    # otherwise the current directory "" would be absent
     mocker.patch("os.path.exists", lambda val: val == "")
+
     w1 = Writer()
     data = [("0", {"output": {"filename": "y", "filetype": "csv"}})]
     res = list(w1.run(data))
@@ -73,3 +83,135 @@ def test_writer_writes(mocker):
     w2 = Writer("output")
     res = list(w2.run(data))
     assert makedirs.mock_calls == [call("output")]
+
+    # test one line where file name is not joined with file extension
+    data_empty_ext = [("0", {"output": {"filename": "y", "fileext": ""}})]
+    assert list(w2.run(data_empty_ext)) == [
+        ('output/y',
+         {'output': {'fileext': '', 'filename': 'y', 'filepath': 'output/y'}})
+    ]
+
+    # absolute paths raise warnings
+    data_absolute_path = [("0", {"output": {"filename": "y", "dirname": "/tmp"}})]
+    with pytest.warns(RuntimeWarning):
+        assert list(w2.run(data_absolute_path)) == [
+            ('output/tmp/y.txt',
+             {'output': {'dirname': '/tmp',
+                         'fileext': 'txt',
+                         'filename': 'y',
+                         'filepath': 'output/tmp/y.txt'}})
+        ]
+
+
+def test_writer_changed(mocker):
+    m = mocker.mock_open()
+    if sys.version[0] == "2":
+        mocker.patch("__builtin__.open", m)
+    else:
+        mocker.patch("builtins.open", m)
+    mocker.patch("os.path.exists", lambda _: True)
+
+    # overwrite set to True always overwrites
+    data = [("1", {"output": {"filename": "x"}})]
+    w_overwrite = Writer(overwrite=True)
+    assert list(w_overwrite.run(data)) == [
+        ('x.txt',
+         {'output': {'changed': True,
+                     'fileext': 'txt',
+                     'filename': 'x',
+                     'filepath': 'x.txt'}}),
+    ]
+    # data is actually "written"
+    call = mocker.call
+    assert m.mock_calls == [
+        call("x.txt", "w"), call().__enter__(),
+        call().write("1"), call().__exit__(None, None, None)
+    ]
+    # context is updated in place
+    assert data[0][1]["output"]["changed"] is True
+
+    # existing_unchanged skips all existing files
+    data_2 = [("2", {"output": {"filename": "x"}})]
+    w_unchanged = Writer(existing_unchanged=True)
+    # clear the list of calls
+    m.reset_mock()
+    assert list(w_unchanged.run(copy.deepcopy(data_2))) == [
+        ('x.txt',
+         {'output': {'changed': False,
+                     'fileext': 'txt',
+                     'filename': 'x',
+                     'filepath': 'x.txt'}}),
+    ]
+    # nothing was actually written
+    assert m.mock_calls == []
+
+
+def test_writer_cached(mocker):
+    m = mocker.mock_open(read_data="1")
+    if sys.version[0] == "2":
+        mocker.patch("__builtin__.open", m)
+    else:
+        mocker.patch("builtins.open", m)
+    mocker.patch("os.path.exists", lambda _: True)
+    data_1 = [("1", {"output": {"filename": "x"}})]
+
+    # unchanged data is not written
+    assert list(Writer().run(copy.deepcopy(data_1))) == [
+        ('x.txt',
+         {'output': {'changed': False,
+                     'fileext': 'txt',
+                     'filename': 'x',
+                     'filepath': 'x.txt'}})
+    ]
+    call = mocker.call
+    assert m.mock_calls == [
+        call("x.txt"), call().__enter__(),
+        call().read(-1), call().__exit__(None, None, None)
+    ]
+    m.reset_mock()
+
+    # even if file is unchanged, context.output.changed remains the same
+    data_changed = [("1", {"output": {"filename": "x", "changed": True}})]
+    assert list(Writer().run(copy.deepcopy(data_changed))) == [
+        ('x.txt',
+         {'output': {'changed': True,
+                     'fileext': 'txt',
+                     'filename': 'x',
+                     'filepath': 'x.txt'}})
+    ]
+    # no writes were done, as before
+    assert m.mock_calls == [
+        call("x.txt"), call().__enter__(),
+        call().read(-1), call().__exit__(None, None, None)
+    ]
+
+
+def test_writer_changed_data(mocker):
+    m = mocker.mock_open(read_data="pi")
+    if sys.version[0] == "2":
+        mocker.patch("__builtin__.open", m)
+    else:
+        mocker.patch("builtins.open", m)
+    mocker.patch("os.path.exists", lambda _: True)
+
+    # changed data is written
+    data_changed = [("1", {"output": {"filename": "x_changed"}})]
+    # changed is True
+    assert list(Writer().run(copy.deepcopy(data_changed))) == [
+        ('x_changed.txt',
+         {'output': {'changed': True,
+                     'fileext': 'txt',
+                     'filename': 'x_changed',
+                     'filepath': 'x_changed.txt'}})
+    ]
+    call = mocker.call
+    # file is read and then written with 1
+    assert m.mock_calls == [
+        call("x_changed.txt"), call().__enter__(),
+        call().read(-1),
+        call().__exit__(None, None, None),
+        call('x_changed.txt', 'w'),
+        call().__enter__(),
+        call().write('1'),
+        call().__exit__(None, None, None)
+    ]