Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

port/tests_from_core #125

Merged
merged 1 commit into from
Oct 15, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
176 changes: 176 additions & 0 deletions test/unittests/test_audioservice.py
Original file line number Diff line number Diff line change
@@ -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)
124 changes: 124 additions & 0 deletions test/unittests/test_event_scheduler.py
Original file line number Diff line number Diff line change
@@ -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)
Loading