Make clean return non-zero exit status when a filter fails

Fixes #106.

opuscleaner/_util.py

@@ -1,9 +1,28 @@
 from typing import TypeVar, Optional
-
+from threading import Thread
 
 T = TypeVar("T")
 
 def none_throws(optional: Optional[T], message: str = "Unexpected `None`") -> T:
     if optional is None:
         raise AssertionError(message)
     return optional
+
+
+class RaisingThread(Thread):
+    """Thread that will raise any uncaught exceptions in the thread in the
+    parent once it joins again."""
+
+    exception: Optional[Exception]
+
+    def run(self):
+        self.exception = None
+        try:
+            super().run()
+        except Exception as exc:
+            self.exception = exc
+
+    def join(self, timeout:float=None):
+        super().join(timeout=timeout)
+        if self.exception is not None:
+            raise self.exception

opuscleaner/clean.py

@@ -26,7 +26,7 @@
 
 from opuscleaner.config import COL_PY, FILTER_PATH
 from opuscleaner.filters import list_filters, set_global_filters, filter_format_command, Filter, FilterStep, FilterPipeline
-from opuscleaner._util import none_throws
+from opuscleaner._util import none_throws, RaisingThread
 
 
 # Queue for printing lines to stdout or stderr. None means end of input.
@@ -399,7 +399,7 @@ def run_parallel(pipeline:Pipeline, stdin:BinaryIO, stdout:BinaryIO, *, parallel
     # Read `batch_queue` for batch filenames, and process them. Put output files
     # on `merge_queue`.
     runners = [
-        Thread(target=run_pipeline, args=[print_queue, batch_queue, merge_queue, pipeline, f'{n}/'])
+        RaisingThread(target=run_pipeline, args=[print_queue, batch_queue, merge_queue, pipeline, f'{n}/'])
         for n in range(parallel)
     ]
 
@@ -416,13 +416,13 @@ def run_parallel(pipeline:Pipeline, stdin:BinaryIO, stdout:BinaryIO, *, parallel
     # using a blocking size-limited Queue() to control the splitter's progress
     # but a SimpleQueue() that sends messages.
 
-    print_queue.put(f'[run.py] Waiting for splitter to finish\n'.encode())
-    splitter.join()
-
     print_queue.put(f'[run.py] Waiting for pipelines to finish\n'.encode())
     for runner in runners:
         runner.join()
 
+    print_queue.put(f'[run.py] Waiting for splitter to finish\n'.encode())
+    splitter.join()
+
     print_queue.put(f'[run.py] Waiting for merger to finish\n'.encode())
     merger.join()
 

test/deeper/filters/fail.json

@@ -0,0 +1,11 @@
+{
+    "description": "Crashes",
+    "type": "monolingual",
+    "command": "python3 -c 'import sys\nsys.exit(int(sys.argv[1]))' $EXITCODE",
+    "parameters": {
+        "EXITCODE": {
+            "type": "int",
+            "default": 1
+        }
+    }
+}

test/test_clean.py

@@ -0,0 +1,79 @@
+import sys
+import os
+import unittest
+import subprocess
+import json
+from typing import List
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+
+
+TEST_CWD = Path(os.path.join(os.path.dirname(__file__), 'deeper'))
+
+FILES = [
+	"bible-uedin-v1.de-en.de.gz",
+	"bible-uedin-v1.de-en.en.gz"
+]
+
+
+class TestClean(unittest.TestCase):
+	def _run(self, args:List[str]):
+		proc = subprocess.Popen(
+			args=[sys.executable, '-m', 'opuscleaner.clean'] + args,
+			cwd=TEST_CWD, # so it can find filters
+			env={
+				'PYTHONPATH': os.path.join(os.path.dirname(__file__), '..') # so it can find opuscleaner code
+			},
+			stdout=subprocess.PIPE,
+			stderr=subprocess.PIPE)
+
+		out, err = proc.communicate()
+		proc.wait()
+		return out, err, proc.returncode
+
+	def test_simple(self):
+		"""Test that clean runs"""
+		config = {
+			"version": 1,
+			"files": FILES,
+			"filters": [
+				{
+					"filter": "deescape_tsv",
+					"parameters": {},
+					"language": None
+				}
+			]
+		}
+		with NamedTemporaryFile(mode='w', dir=TEST_CWD / 'data/train-parts') as fh:
+			json.dump(config, fh)
+			fh.flush()
+			for mode in [[], ['--parallel', '1']]:
+				with self.subTest(mode=mode):
+					out, err, retval = self._run([*mode, fh.name])
+					self.assertEqual(out.count(b'\n'), 62195)
+					self.assertEqual(retval, 0)
+
+	def test_filter_fail(self):
+		"""Test that clean returns a non-zero exit code if a filter fails"""
+		config = {
+			"version": 1,
+				"files": FILES,
+				"filters": [
+					{
+						"filter": "fail",
+						"parameters": {
+							"EXITCODE": "42"
+						},
+						"language": "de"
+					}
+				]
+		}
+		with NamedTemporaryFile(mode='w', dir=TEST_CWD / 'data/train-parts') as fh:
+			json.dump(config, fh)
+			fh.flush()
+
+			for mode in [[], ['--parallel', '1']]:
+				with self.subTest(mode=mode):
+					out, err, retval = self._run([*mode, fh.name])
+					self.assertEqual(out.count(b'\n'), 0)
+					self.assertNotEqual(retval, 0)