Coverage for test/unittests/test_manager.py: 99%

1import unittest 

2from os.path import join, dirname 

3from unittest.mock import MagicMock, patch 

4 

5from ovos_bus_client.message import Message 

6 

7from ovos_core.skill_manager import SkillManager 

8 

9 

10class TestSkillManager(unittest.TestCase): 

11 

12 def setUp(self): 

13 self.bus = MagicMock() 

14 self.skill_manager = SkillManager(self.bus) 

15 

16 def test_blacklist_property(self): 

17 blacklist = self.skill_manager.blacklist 

18 self.assertIsInstance(blacklist, list) 

19 

20 @patch('ovos_core.skill_manager.LOG') 

21 def test_handle_settings_file_change(self, mock_log): 

22 path = '/some/path/skills/settings.json' 

23 self.skill_manager._handle_settings_file_change(path) 

24 self.bus.emit.assert_called_once_with(Message("ovos.skills.settings_changed", {"skill_id": "skills"})) 

25 mock_log.info.assert_called_once_with(f"skill settings.json change detected for skills") 

26 

27 @patch('ovos_core.skill_manager.find_skill_plugins', return_value={'mock_plugin': 'path/to/mock_plugin'}) 

28 def test_load_plugin_skills(self, mock_find_skill_plugins): 

29 self.skill_manager._load_plugin_skill = MagicMock(return_value=True) 

30 self.skill_manager.load_plugin_skills(network=True, internet=True) 

31 self.assertTrue(self.skill_manager._load_plugin_skill.called) 

32 mock_find_skill_plugins.assert_called_once() 

33 

34 @patch('ovos_core.skill_manager.is_gui_connected', return_value=True) 

35 def test_handle_gui_connected(self, mock_is_gui_connected): 

36 self.skill_manager._allow_state_reloads = True 

37 self.skill_manager._gui_event.clear() 

38 self.skill_manager._load_new_skills = MagicMock() 

39 self.skill_manager.handle_gui_connected(Message("", data={"permanent": False})) 

40 self.assertTrue(self.skill_manager._gui_event.is_set()) 

41 self.assertTrue(self.skill_manager._load_new_skills.called) 

42 

43 @patch('ovos_core.skill_manager.is_gui_connected', return_value=False) 

44 def test_handle_gui_disconnected(self, mock_is_gui_connected): 

45 self.skill_manager._allow_state_reloads = True 

46 self.skill_manager._gui_event.set() 

47 self.skill_manager._unload_on_gui_disconnect = MagicMock() 

48 self.skill_manager.handle_gui_disconnected(Message("")) 

49 self.assertFalse(self.skill_manager._gui_event.is_set()) 

50 self.assertTrue(self.skill_manager._unload_on_gui_disconnect.called) 

51 

52 @patch('ovos_core.skill_manager.is_connected_http', return_value=True) 

53 def test_handle_internet_connected(self, mock_is_connected): 

54 self.skill_manager._connected_event.clear() 

55 self.skill_manager._network_event.clear() 

56 self.skill_manager._network_loaded.set() 

57 self.skill_manager._load_on_internet = MagicMock() 

58 self.skill_manager.handle_internet_connected(Message("")) 

59 self.assertTrue(self.skill_manager._connected_event.is_set()) 

60 self.assertTrue(self.skill_manager._network_loaded.is_set()) 

61 self.assertTrue(self.skill_manager._load_on_internet.called) 

62 

63 @patch('ovos_core.skill_manager.is_connected_http', return_value=False) 

64 def test_handle_internet_disconnected(self, mock_is_connected): 

65 self.skill_manager._allow_state_reloads = True 

66 self.skill_manager._connected_event.set() 

67 self.skill_manager._internet_loaded.set() 

68 self.skill_manager._unload_on_internet_disconnect = MagicMock() 

69 self.skill_manager.handle_internet_disconnected(Message("")) 

70 self.assertFalse(self.skill_manager._connected_event.is_set()) 

71 self.assertTrue(self.skill_manager._unload_on_internet_disconnect.called) 

72 

73 @patch('ovos_core.skill_manager.is_connected_http', return_value=True) 

74 def test_handle_network_connected(self, mock_is_connected): 

75 self.skill_manager._network_event.clear() 

76 self.skill_manager._load_on_network = MagicMock() 

77 self.skill_manager.handle_network_connected(Message("")) 

78 self.assertTrue(self.skill_manager._network_event.is_set()) 

79 self.assertTrue(self.skill_manager._load_on_network.called) 

80 

81 @patch('ovos_core.skill_manager.is_connected_http', return_value=False) 

82 def test_handle_network_disconnected(self, mock_is_connected): 

83 self.skill_manager._allow_state_reloads = True 

84 self.skill_manager._network_event.set() 

85 self.skill_manager._unload_on_network_disconnect = MagicMock() 

86 self.skill_manager.handle_network_disconnected(Message("")) 

87 self.assertFalse(self.skill_manager._network_event.is_set()) 

88 self.assertTrue(self.skill_manager._unload_on_network_disconnect.called) 

89 

90 @patch('ovos_core.skill_manager.is_gui_connected', return_value=True) 

91 @patch('ovos_core.skill_manager.is_connected_http', return_value=True) 

92 def test_sync_skill_loading_state_no_phal_plugin(self, mock_is_connected, mock_is_gui_connected): 

93 self.bus.wait_for_response.return_value = None 

94 

95 self.skill_manager._gui_event.clear() 

96 self.skill_manager._connected_event.clear() 

97 self.skill_manager._network_event.clear() 

98 

99 self.skill_manager._sync_skill_loading_state() 

100 

101 self.assertTrue(self.skill_manager._gui_event.is_set()) 

102 self.assertTrue(self.bus.emit.called) 

103 self.assertEqual(self.bus.emit.call_args[0][0].msg_type, 'mycroft.internet.connected') 

104 

105 @patch('ovos_core.skill_manager.is_gui_connected', return_value=False) 

106 def test_sync_skill_loading_state_phal_plugin_no_gui(self, mock_is_gui_connected): 

107 self.bus.wait_for_response.return_value = Message("ovos.PHAL.internet_check", data={"internet_connected": True}) 

108 

109 self.skill_manager._gui_event.clear() 

110 self.skill_manager._connected_event.clear() 

111 self.skill_manager._network_event.clear() 

112 

113 self.skill_manager._sync_skill_loading_state() 

114 

115 self.assertFalse(self.skill_manager._gui_event.is_set()) 

116 self.assertTrue(self.bus.emit.called) 

117 self.assertEqual(self.bus.emit.call_args[0][0].msg_type, 'mycroft.internet.connected') 

118 

119 @patch('ovos_core.skill_manager.is_gui_connected', return_value=True) 

120 def test_sync_skill_loading_state_gui_no_internet_but_network(self, mock_is_gui_connected): 

121 self.bus.wait_for_response.return_value = Message("ovos.PHAL.internet_check", 

122 data={"internet_connected": False, 

123 "network_connected": True}) 

124 

125 self.skill_manager._gui_event.clear() 

126 self.skill_manager._connected_event.clear() 

127 self.skill_manager._network_event.clear() 

128 

129 self.skill_manager._sync_skill_loading_state() 

130 

131 self.assertTrue(self.skill_manager._gui_event.is_set()) 

132 self.assertTrue(self.bus.emit.called) 

133 self.assertEqual(self.bus.emit.call_args[0][0].msg_type, 'mycroft.network.connected') 

134 

135 @patch('ovos_core.skill_manager.MessageBusClient', autospec=True) 

136 def test_get_internal_skill_bus_shared_connection(self, mock_MessageBusClient): 

137 # Set the configuration to use shared_connection=True 

138 self.skill_manager.config = {'websocket': {'shared_connection': True}} 

139 

140 # Call the method under test 

141 result = self.skill_manager._get_internal_skill_bus() 

142 

143 # Ensure the shared connection is returned 

144 self.assertEqual(result, self.bus) 

145 # Ensure that MessageBusClient is not called since shared_connection=True 

146 self.assertFalse(mock_MessageBusClient.called) 

147 

148 @patch('ovos_core.skill_manager.MessageBusClient', autospec=True) 

149 def test_get_internal_skill_bus_not_shared_connection(self, mock_MessageBusClient): 

150 # Set the configuration to use shared_connection=False 

151 self.skill_manager.config = {'websocket': {'shared_connection': False}} 

152 

153 # Call the method under test 

154 result = self.skill_manager._get_internal_skill_bus() 

155 

156 # Ensure a new MessageBusClient is created and returned 

157 mock_MessageBusClient.assert_called_once_with(cache=True) 

158 self.assertTrue(result.run_in_thread.called) 

159 

160 

161if __name__ == '__main__': 

162 unittest.main()