Coverage for test/end2end/test_activate.py: 100%

1from unittest import TestCase 

2 

3from ovos_bus_client.message import Message 

4from ovos_bus_client.session import Session 

5from ovos_utils.log import LOG 

6 

7from ovos_workshop.skills.converse import ConversationalSkill 

8from ovoscope import End2EndTest, get_minicroft 

9 

10 

11class TestSkill(ConversationalSkill): 

12 

13 def initialize(self): 

14 self.add_event("test_activate", self.handle_activate_test) 

15 self.add_event("test_deactivate", self.handle_deactivate_test) 

16 

17 def handle_activate_test(self, message: Message): 

18 self.activate() 

19 

20 def handle_deactivate_test(self, message: Message): 

21 self.deactivate() 

22 

23 def can_converse(self, message: Message) -> bool: 

24 return True 

25 

26 def converse(self, message: Message): 

27 self.log.debug("I dont wanna converse anymore") 

28 self.deactivate() 

29 

30 

31class TestDeactivate(TestCase): 

32 

33 def setUp(self): 

34 LOG.set_level("DEBUG") 

35 self.skill_id = "test_activation.openvoiceos" 

36 self.minicroft = get_minicroft([self.skill_id], 

37 extra_skills={self.skill_id: TestSkill}) 

38 

39 def tearDown(self): 

40 if self.minicroft: 

41 self.minicroft.stop() 

42 LOG.set_level("CRITICAL") 

43 

44 def test_activate(self): 

45 session = Session("123") 

46 session.lang = "en-US" 

47 session.deactivate_skill(self.skill_id) # start with skill inactive 

48 

49 message = Message("test_activate", 

50 context={"session": session.serialize(), 

51 "source": "A", "destination": "B"}) 

52 

53 final_session = Session("123") 

54 final_session.lang = "en-US" 

55 final_session.active_skills = [(self.skill_id, 0.0)] 

56 

57 test = End2EndTest( 

58 minicroft=self.minicroft, 

59 skill_ids=[self.skill_id], 

60 source_message=message, 

61 deactivation_points=[message.msg_type], 

62 final_session=final_session, 

63 activation_points=["intent.service.skills.activated"], 

64 # messages internal to ovos-core, i.e. would not be sent to clients such as hivemind 

65 keep_original_src=[ 

66 #"intent.service.skills.activate", # TODO 

67 #f"{self.skill_id}.activate", # TODO 

68 ], 

69 expected_messages=[ 

70 message, 

71 # handler code 

72 Message("intent.service.skills.activate", 

73 data={"skill_id": self.skill_id}, 

74 context={"skill_id": self.skill_id}), 

75 Message("intent.service.skills.activated", 

76 data={"skill_id": self.skill_id}, 

77 context={"skill_id": self.skill_id}), 

78 Message(f"{self.skill_id}.activate", 

79 data={}, 

80 context={"skill_id": self.skill_id}), 

81 ] 

82 ) 

83 

84 test.execute(timeout=10) 

85 

86 def test_deactivate(self): 

87 session = Session("123") 

88 session.lang = "en-US" 

89 session.activate_skill(self.skill_id) # start with skill active 

90 

91 message = Message("test_deactivate", 

92 context={"session": session.serialize(), 

93 "source": "A", "destination": "B"}) 

94 

95 final_session = Session("123") 

96 final_session.lang = "en-US" 

97 final_session.active_skills = [] 

98 

99 test = End2EndTest( 

100 minicroft=self.minicroft, 

101 skill_ids=[self.skill_id], 

102 source_message=message, 

103 final_session=final_session, 

104 activation_points=[message.msg_type], # starts activated 

105 deactivation_points=["intent.service.skills.deactivated"], 

106 # messages internal to ovos-core, i.e. would not be sent to clients such as hivemind 

107 keep_original_src=[ 

108 #"intent.service.skills.deactivate", # TODO 

109 #f"{self.skill_id}.deactivate", # TODO 

110 #f"{self.skill_id}.activate", # TODO 

111 ], 

112 expected_messages=[ 

113 message, 

114 # handler code 

115 Message("intent.service.skills.deactivate", 

116 data={"skill_id": self.skill_id}, 

117 context={"skill_id": self.skill_id}), 

118 Message("intent.service.skills.deactivated", 

119 data={"skill_id": self.skill_id}, 

120 context={"skill_id": self.skill_id}), 

121 Message(f"{self.skill_id}.deactivate", 

122 data={}, 

123 context={"skill_id": self.skill_id}), 

124 ] 

125 ) 

126 

127 test.execute(timeout=10) 

128 

129 def test_deactivate_inside_converse(self): 

130 session = Session("123") 

131 session.lang = "en-US" 

132 session.activate_skill(self.skill_id) # start with skill active 

133 

134 message = Message("recognizer_loop:utterance", 

135 {"utterances": ["deactivate skill from within converse"], "lang": session.lang}, 

136 {"session": session.serialize(), "source": "A", "destination": "B"}) 

137 

138 final_session = Session("123") 

139 final_session.lang = "en-US" 

140 final_session.active_skills = [] 

141 

142 test = End2EndTest( 

143 minicroft=self.minicroft, 

144 skill_ids=[self.skill_id], 

145 source_message=message, 

146 final_session=final_session, 

147 activation_points=[message.msg_type], # starts activated 

148 deactivation_points=["intent.service.skills.deactivated"], 

149 # messages internal to ovos-core, i.e. would not be sent to clients such as hivemind 

150 keep_original_src=[ 

151 f"{self.skill_id}.converse.ping", 

152 f"{self.skill_id}.converse.request", 

153 #"intent.service.skills.deactivate", # TODO 

154 #f"{self.skill_id}.deactivate", # TODO 

155 #f"{self.skill_id}.activate", # TODO 

156 ], 

157 expected_messages=[ 

158 message, 

159 Message(f"{self.skill_id}.converse.ping", 

160 data={"utterances": ["deactivate skill from within converse"], "skill_id": self.skill_id}, 

161 context={}), 

162 Message("skill.converse.pong", 

163 data={"can_handle": True, "skill_id": self.skill_id}, 

164 context={"skill_id": self.skill_id}), 

165 Message(f"{self.skill_id}.activate", 

166 data={}, 

167 context={"skill_id": self.skill_id}), 

168 Message("converse:skill", 

169 data={"utterances": ["deactivate skill from within converse"], "lang": session.lang, 

170 "skill_id": self.skill_id}, 

171 context={"skill_id": self.skill_id}), 

172 Message(f"{self.skill_id}.converse.request", 

173 data={"utterances": ["deactivate skill from within converse"], "lang": session.lang}, 

174 context={"skill_id": self.skill_id}), 

175 # converse handler code 

176 Message("intent.service.skills.deactivate", 

177 data={"skill_id": self.skill_id}, 

178 context={"skill_id": self.skill_id}), 

179 Message("intent.service.skills.deactivated", 

180 data={"skill_id": self.skill_id}, 

181 context={"skill_id": self.skill_id}), 

182 Message(f"{self.skill_id}.deactivate", 

183 data={}, 

184 context={"skill_id": self.skill_id}), 

185 # post converse handler 

186 Message("skill.converse.response", 

187 data={"skill_id": self.skill_id}, 

188 context={"skill_id": self.skill_id}), 

189 Message("ovos.utterance.handled", 

190 data={}, 

191 context={"skill_id": self.skill_id}) 

192 

193 ] 

194 ) 

195 

196 test.execute(timeout=10)