Mercurial > hg > auditok
changeset 271:dbb12345a19f
Add tests for initialize_workers
| author | Amine Sehili <amine.sehili@gmail.com> |
|---|---|
| date | Sun, 15 Sep 2019 18:13:07 +0100 |
| parents | 2b981910b1cd |
| children | eb77a08a608a |
| files | auditok/workers.py tests/test_cmdline_util.py |
| diffstat | 2 files changed, 154 insertions(+), 2 deletions(-) [+] |
line wrap: on
line diff
--- a/auditok/workers.py Sun Sep 15 17:07:27 2019 +0100 +++ b/auditok/workers.py Sun Sep 15 18:13:07 2019 +0100 @@ -409,6 +409,12 @@ self._log(message) +class CommandLineWorker(Worker): + def __init__(self, command, timeout=0.2, logger=None): + self._command = command + Worker.__init__(self, timeout=timeout, logger=logger) + + class PrintWorker(Worker): def __init__( self,
--- a/tests/test_cmdline_util.py Sun Sep 15 17:07:27 2019 +0100 +++ b/tests/test_cmdline_util.py Sun Sep 15 18:13:07 2019 +0100 @@ -1,5 +1,6 @@ import os from unittest import TestCase +from unittest.mock import patch from tempfile import TemporaryDirectory from collections import namedtuple from genty import genty, genty_dataset @@ -12,7 +13,13 @@ initialize_workers, KeywordArguments, ) -from auditok.workers import StreamSaverWorker +from auditok.workers import ( + StreamSaverWorker, + RegionSaverWorker, + PlayerWorker, + CommandLineWorker, + PrintWorker, +) from auditok.exceptions import TimeFormatError _ArgsNamespece = namedtuple( @@ -186,4 +193,143 @@ def test_make_logger_None(self): logger = make_logger(debug_stdout=False, debug_file=None) self.assertIsNone(logger) - del logger + + def test_initialize_workers_all(self): + with patch("auditok.cmdline_util.player_for") as patched_player_for: + reader, observers = initialize_workers( + input="tests/data/test_16KHZ_mono_400Hz.wav", + save_stream="output.wav", + export_format="wave", + save_detections_as="{id}.wav", + echo=True, + progress_bar=False, + command="some command", + quiet=False, + printf="abcd", + time_format="%S", + timestamp_format="%h:%M:%S", + ) + reader.stop() + self.assertTrue(patched_player_for.called) + self.assertIsInstance(reader, StreamSaverWorker) + for obs, cls in zip( + observers, + [ + RegionSaverWorker, + PlayerWorker, + CommandLineWorker, + PrintWorker, + ], + ): + self.assertIsInstance(obs, cls) + + def test_initialize_workers_no_RegionSaverWorker(self): + with patch("auditok.cmdline_util.player_for") as patched_player_for: + reader, observers = initialize_workers( + input="tests/data/test_16KHZ_mono_400Hz.wav", + save_stream="output.wav", + export_format="wave", + save_detections_as=None, + echo=True, + progress_bar=False, + command="some command", + quiet=False, + printf="abcd", + time_format="%S", + timestamp_format="%h:%M:%S", + ) + reader.stop() + self.assertTrue(patched_player_for.called) + self.assertIsInstance(reader, StreamSaverWorker) + for obs, cls in zip( + observers, [PlayerWorker, CommandLineWorker, PrintWorker] + ): + self.assertIsInstance(obs, cls) + + def test_initialize_workers_no_PlayerWorker(self): + with patch("auditok.cmdline_util.player_for") as patched_player_for: + reader, observers = initialize_workers( + input="tests/data/test_16KHZ_mono_400Hz.wav", + save_stream="output.wav", + export_format="wave", + save_detections_as="{id}.wav", + echo=False, + progress_bar=False, + command="some command", + quiet=False, + printf="abcd", + time_format="%S", + timestamp_format="%h:%M:%S", + ) + reader.stop() + self.assertFalse(patched_player_for.called) + self.assertIsInstance(reader, StreamSaverWorker) + for obs, cls in zip( + observers, [RegionSaverWorker, CommandLineWorker, PrintWorker] + ): + self.assertIsInstance(obs, cls) + + def test_initialize_workers_no_CommandLineWorker(self): + with patch("auditok.cmdline_util.player_for") as patched_player_for: + reader, observers = initialize_workers( + input="tests/data/test_16KHZ_mono_400Hz.wav", + save_stream="output.wav", + export_format="wave", + save_detections_as="{id}.wav", + echo=True, + progress_bar=False, + command=None, + quiet=False, + printf="abcd", + time_format="%S", + timestamp_format="%h:%M:%S", + ) + reader.stop() + self.assertTrue(patched_player_for.called) + self.assertIsInstance(reader, StreamSaverWorker) + for obs, cls in zip( + observers, [RegionSaverWorker, PlayerWorker, PrintWorker] + ): + self.assertIsInstance(obs, cls) + + def test_initialize_workers_no_PrintWorker(self): + with patch("auditok.cmdline_util.player_for") as patched_player_for: + reader, observers = initialize_workers( + input="tests/data/test_16KHZ_mono_400Hz.wav", + save_stream="output.wav", + export_format="wave", + save_detections_as="{id}.wav", + echo=True, + progress_bar=False, + command="some command", + quiet=True, + printf="abcd", + time_format="%S", + timestamp_format="%h:%M:%S", + ) + reader.stop() + self.assertTrue(patched_player_for.called) + self.assertIsInstance(reader, StreamSaverWorker) + for obs, cls in zip( + observers, [RegionSaverWorker, PlayerWorker, CommandLineWorker] + ): + self.assertIsInstance(obs, cls) + + def test_initialize_workers_no_observers(self): + with patch("auditok.cmdline_util.player_for") as patched_player_for: + reader, observers = initialize_workers( + input="tests/data/test_16KHZ_mono_400Hz.wav", + save_stream=None, + export_format="wave", + save_detections_as=None, + echo=True, + progress_bar=False, + command=None, + quiet=True, + printf="abcd", + time_format="%S", + timestamp_format="%h:%M:%S", + ) + self.assertTrue(patched_player_for.called) + self.assertFalse(isinstance(reader, StreamSaverWorker)) + self.assertTrue(len(observers), 0)