diff --git a/test/unittests/test_audioservice.py b/test/unittests/test_audioservice.py new file mode 100644 index 0000000..898df9c --- /dev/null +++ b/test/unittests/test_audioservice.py @@ -0,0 +1,176 @@ +from unittest import TestCase, mock + +from ovos_bus_client.message import Message +from ovos_bus_client.apis.ocp import ClassicAudioServiceInterface + + +class TestAudioServiceControls(TestCase): + def assertLastMessageTypeEqual(self, bus, msg_type): + message = bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, msg_type) + + def setUp(self): + self.bus = mock.Mock(name='bus') + self.audioservice = ClassicAudioServiceInterface(self.bus) + + def test_pause(self): + self.audioservice.pause() + self.assertLastMessageTypeEqual(self.bus, + 'mycroft.audio.service.pause') + + def test_resume(self): + self.audioservice.resume() + self.assertLastMessageTypeEqual(self.bus, + 'mycroft.audio.service.resume') + + def test_next(self): + self.audioservice.next() + self.assertLastMessageTypeEqual(self.bus, 'mycroft.audio.service.next') + + def test_prev(self): + self.audioservice.prev() + self.assertLastMessageTypeEqual(self.bus, 'mycroft.audio.service.prev') + + def test_stop(self): + self.audioservice.stop() + self.assertLastMessageTypeEqual(self.bus, 'mycroft.audio.service.stop') + + def test_seek(self): + self.audioservice.seek() + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, + 'mycroft.audio.service.seek_forward') + self.assertEqual(message.data['seconds'], 1) + self.audioservice.seek(5) + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, + 'mycroft.audio.service.seek_forward') + self.assertEqual(message.data['seconds'], 5) + self.audioservice.seek(-5) + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, + 'mycroft.audio.service.seek_backward') + self.assertEqual(message.data['seconds'], 5) + + +class TestAudioServicePlay(TestCase): + def setUp(self): + self.bus = mock.Mock(name='bus') + self.audioservice = ClassicAudioServiceInterface(self.bus) + + def test_proper_uri(self): + self.audioservice.play('file:///hello_nasty.mp3') + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.play') + self.assertEqual(message.data['tracks'], ['file:///hello_nasty.mp3']) + self.assertEqual(message.data['repeat'], False) + + def test_path(self): + self.audioservice.play('/hello_nasty.mp3') + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.play') + self.assertEqual(message.data['tracks'], ['file:///hello_nasty.mp3']) + self.assertEqual(message.data['repeat'], False) + + def test_tuple(self): + """Test path together with mimetype.""" + self.audioservice.play(('/hello_nasty.mp3', 'audio/mp3')) + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.play') + self.assertEqual(message.data['tracks'], + [('file:///hello_nasty.mp3', 'audio/mp3')]) + self.assertEqual(message.data['repeat'], False) + + def test_invalid(self): + """Test play request with invalid type.""" + with self.assertRaises(ValueError): + self.audioservice.play(12) + + def test_extra_arguments(self): + """Test sending along utterance and setting repeat.""" + self.audioservice.play('/hello_nasty.mp3', 'on vlc', True) + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.play') + self.assertEqual(message.data['tracks'], ['file:///hello_nasty.mp3']) + self.assertEqual(message.data['repeat'], True) + self.assertEqual(message.data['utterance'], 'on vlc') + + +class TestAudioServiceQueue(TestCase): + def setUp(self): + self.bus = mock.Mock(name='bus') + self.audioservice = ClassicAudioServiceInterface(self.bus) + + def test_uri(self): + self.audioservice.queue('file:///hello_nasty.mp3') + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.queue') + self.assertEqual(message.data['tracks'], ['file:///hello_nasty.mp3']) + + def test_path(self): + self.audioservice.queue('/hello_nasty.mp3') + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.queue') + self.assertEqual(message.data['tracks'], ['file:///hello_nasty.mp3']) + + def test_tuple(self): + self.audioservice.queue(('/hello_nasty.mp3', 'audio/mp3')) + message = self.bus.emit.call_args_list[-1][0][0] + self.assertEqual(message.msg_type, 'mycroft.audio.service.queue') + self.assertEqual(message.data['tracks'], + [('file:///hello_nasty.mp3', 'audio/mp3')]) + + def test_invalid(self): + with self.assertRaises(ValueError): + self.audioservice.queue(12) + + +class TestAudioServiceMisc(TestCase): + def test_lifecycle(self): + bus = mock.Mock(name='bus') + audioservice = ClassicAudioServiceInterface(bus) + self.assertEqual(audioservice.bus, bus) + + def test_available_backends(self): + bus = mock.Mock(name='bus') + audioservice = ClassicAudioServiceInterface(bus) + + available_backends = { + 'simple': { + 'suported_uris': ['http', 'file'], + 'default': True, + 'remote': False + } + } + bus.wait_for_response.return_value = Message('test_msg', + available_backends) + response = audioservice.available_backends() + self.assertEqual(available_backends, response) + # Check no response behaviour + bus.wait_for_response.return_value = None + response = audioservice.available_backends() + self.assertEqual({}, response) + + def test_track_info(self): + """Test is_playing property.""" + bus = mock.Mock(name='bus') + audioservice = ClassicAudioServiceInterface(bus) + info = {'album': 'Hello Nasty', + 'artist': 'Beastie Boys', + 'name': 'Intergalactic' + } + bus.wait_for_response.return_value = Message('test_msg', info) + self.assertEqual(audioservice.track_info(), info) + bus.wait_for_response.return_value = None + self.assertEqual(audioservice.track_info(), {}) + + def test_is_playing(self): + """Test is_playing property.""" + bus = mock.Mock(name='bus') + audioservice = ClassicAudioServiceInterface(bus) + audioservice.track_info = mock.Mock() + + audioservice.track_info.return_value = {'track': 'one cool song'} + self.assertTrue(audioservice.is_playing) + audioservice.track_info.return_value = {} + self.assertFalse(audioservice.is_playing) diff --git a/test/unittests/test_event_scheduler.py b/test/unittests/test_event_scheduler.py new file mode 100644 index 0000000..8c26803 --- /dev/null +++ b/test/unittests/test_event_scheduler.py @@ -0,0 +1,124 @@ +""" + Test cases regarding the event scheduler. +""" + +import unittest +import time +from pyee import ExecutorEventEmitter + +from unittest.mock import MagicMock, patch +from ovos_utils.messagebus import FakeBus +from ovos_bus_client.util.scheduler import EventScheduler, EventSchedulerInterface + + +# TODO - move to ovos-bus-client +class TestEventScheduler(unittest.TestCase): + @patch('threading.Thread') + @patch('json.load') + @patch('json.dump') + @patch('builtins.open') + def test_create(self, mock_open, mock_json_dump, mock_load, mock_thread): + """ + Test creating and shutting down event_scheduler. + """ + mock_load.return_value = '' + mock_open.return_value = MagicMock() + emitter = MagicMock() + es = EventScheduler(emitter) + es.shutdown() + self.assertEqual(mock_json_dump.call_args[0][0], {}) + + @patch('threading.Thread') + @patch('json.load') + @patch('json.dump') + @patch('builtins.open') + def test_add_remove(self, mock_open, mock_json_dump, + mock_load, mock_thread): + """ + Test add an event and then remove it. + """ + # Thread start is mocked so will not actually run the thread loop + mock_load.return_value = '' + mock_open.return_value = MagicMock() + emitter = MagicMock() + es = EventScheduler(emitter) + + # 900000000000 should be in the future for a long time + es.schedule_event('test', 90000000000, None) + es.schedule_event('test-2', 90000000000, None) + + es.check_state() # run one cycle + self.assertTrue('test' in es.events) + self.assertTrue('test-2' in es.events) + + es.remove_event('test') + es.check_state() # run one cycle + self.assertTrue('test' not in es.events) + self.assertTrue('test-2' in es.events) + es.shutdown() + + @patch('threading.Thread') + @patch('json.load') + @patch('json.dump') + @patch('builtins.open') + def test_save(self, mock_open, mock_dump, mock_load, mock_thread): + """ + Test save functionality. + """ + mock_load.return_value = '' + mock_open.return_value = MagicMock() + emitter = MagicMock() + es = EventScheduler(emitter) + + # 900000000000 should be in the future for a long time + es.schedule_event('test', 900000000000, None) + es.schedule_event('test-repeat', 910000000000, 60) + es.check_state() + + es.shutdown() + + # Make sure the dump method wasn't called with test-repeat + self.assertEqual(mock_dump.call_args[0][0], + {'test': [(900000000000, None, {}, None)]}) + + @patch('threading.Thread') + @patch('json.load') + @patch('json.dump') + @patch('builtins.open') + def test_send_event(self, mock_open, mock_dump, mock_load, mock_thread): + """ + Test save functionality. + """ + mock_load.return_value = '' + mock_open.return_value = MagicMock() + emitter = MagicMock() + es = EventScheduler(emitter) + + # 0 should be in the future for a long time + es.schedule_event('test', time.time(), None) + + es.check_state() + self.assertEqual(emitter.emit.call_args[0][0].msg_type, 'test') + self.assertEqual(emitter.emit.call_args[0][0].data, {}) + es.shutdown() + + +class TestEventSchedulerInterface(unittest.TestCase): + def test_shutdown(self): + def f(message): + print('TEST FUNC') + + bus = ExecutorEventEmitter() + + es = EventSchedulerInterface('tester') + es.set_bus(FakeBus()) + es.set_id('id') + + # Schedule a repeating event + es.schedule_repeating_event(f, None, 10, name='f') + self.assertTrue(len(es.bus.ee._events['id:f']) == 1) + + es.shutdown() + # Check that the reference to the function has been removed from the + # bus emitter + self.assertTrue(len(bus._events['id:f']) == 0)