changeset 279:87bb649f5d3a

Add test for CommandLineWorker

--- a/tests/test_workers.py	Sun Sep 29 10:03:24 2019 +0100
+++ b/tests/test_workers.py	Sun Sep 29 10:20:52 2019 +0100
@@ -120,7 +120,7 @@
         self.assertEqual(mock_calls, expected_save_calls)
         self.assertEqual(len(tokenizer.detections), len(self.expected))
 
-        log_fmt = '[SAVE]: Detection {id} saved as "{filename}"'
+        log_fmt = "[SAVE]: Detection {id} saved as '{filename}'"
         for i, (det, exp, log_line) in enumerate(
             zip(tokenizer.detections, self.expected, log_lines), 1
         ):
@@ -134,6 +134,51 @@
             # remove timestamp part and strip new line
             self.assertEqual(log_line[28:].strip(), exp_log_line)
 
+    def test_CommandLineWorker(self):
+        command_format = "do nothing with"
+        with TemporaryDirectory() as tmpdir:
+            file = os.path.join(tmpdir, "file.log")
+            logger = make_logger(file=file, name="test_CommandLineWorker")
+            observers = [CommandLineWorker(command_format, logger=logger)]
+            tokenizer = TokenizerWorker(
+                self.reader,
+                logger=logger,
+                observers=observers,
+                min_dur=0.3,
+                max_dur=2,
+                max_silence=0.2,
+                drop_trailing_silence=False,
+                strict_min_dur=False,
+                eth=50,
+            )
+            with patch("auditok.workers.os.system") as patched_os_system:
+                tokenizer.start_all()
+                tokenizer.join()
+                tokenizer._observers[0].join()
+            # Get logged text
+            with open(file) as fp:
+                log_lines = [
+                    line
+                    for line in fp.readlines()
+                    if line.startswith("[COMMAND]")
+                ]
+
+        # Asser PrintWorker ran as expected
+        expected_save_calls = [call(command_format) for _ in self.expected]
+        self.assertEqual(patched_os_system.mock_calls, expected_save_calls)
+        self.assertEqual(len(tokenizer.detections), len(self.expected))
+
+        log_fmt = "[COMMAND]: Detection {id} command '{command}'"
+        for i, (det, exp, log_line) in enumerate(
+            zip(tokenizer.detections, self.expected, log_lines), 1
+        ):
+            start, end = exp
+            exp_log_line = log_fmt.format(i, command_format)
+            self.assertAlmostEqual(det.start, start)
+            self.assertAlmostEqual(det.end, end)
+            # remove timestamp part and strip new line
+            self.assertEqual(log_line[28:].strip(), exp_log_line)
+
     def test_PrintWorker(self):
         observers = [
             PrintWorker(print_format="[{id}] {start} {end}, dur: {duration}")