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

1from copy import deepcopy 

2from unittest import TestCase 

3 

4from ovos_bus_client.message import Message 

5from ovos_bus_client.session import Session 

6from ovos_utils.log import LOG 

7 

8from ovoscope import End2EndTest, get_minicroft 

9 

10 

11class TestConverse(TestCase): 

12 

13 def setUp(self): 

14 LOG.set_level("DEBUG") 

15 self.skill_id = "ovos-skill-parrot.openvoiceos" 

16 self.minicroft = get_minicroft([self.skill_id]) # reuse for speed, but beware if skills keeping internal state 

17 

18 def tearDown(self): 

19 if self.minicroft: 

20 self.minicroft.stop() 

21 LOG.set_level("CRITICAL") 

22 

23 def test_parrot_mode(self): 

24 session = Session("123") 

25 session.lang = "en-US" 

26 session.pipeline = ["ovos-converse-pipeline-plugin", "ovos-padatious-pipeline-plugin-high"] 

27 

28 message1 = Message("recognizer_loop:utterance", 

29 {"utterances": ["start parrot mode"], "lang": session.lang}, 

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

31 # NOTE: we dont pass session after first message 

32 # End2EndTest will inject/update the session from message1 

33 message2 = Message("recognizer_loop:utterance", 

34 {"utterances": ["echo test"], "lang": session.lang}, 

35 {"source": "A", "destination": "B"}) 

36 message3 = Message("recognizer_loop:utterance", 

37 {"utterances": ["stop parrot"], "lang": session.lang}, 

38 {"source": "A", "destination": "B"}) 

39 message4 = Message("recognizer_loop:utterance", 

40 {"utterances": ["echo test"], "lang": session.lang}, 

41 {"source": "A", "destination": "B"}) 

42 

43 expected1 = [ 

44 message1, 

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

46 data={}, 

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

48 Message(f"{self.skill_id}:start_parrot.intent", 

49 data={"utterance": "start parrot mode", "lang": session.lang}, 

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

51 Message("mycroft.skill.handler.start", 

52 data={"name": "ParrotSkill.handle_start_parrot_intent"}, 

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

54 Message("speak", 

55 data={"expect_response": False, 

56 "meta": { 

57 "dialog": "parrot_start", 

58 "data": {}, 

59 "skill": self.skill_id 

60 }}, 

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

62 Message("mycroft.skill.handler.complete", 

63 data={"name": "ParrotSkill.handle_start_parrot_intent"}, 

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

65 Message("ovos.utterance.handled", 

66 data={}, 

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

68 ] 

69 expected2 = [ 

70 message2, 

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

72 data={"utterances": ["echo test"], "skill_id": self.skill_id}, 

73 context={}), 

74 Message("skill.converse.pong", 

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

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

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

78 data={}, 

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

80 Message("converse:skill", 

81 data={"utterances": ["echo test"], "lang": session.lang, "skill_id": self.skill_id}, 

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

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

84 data={"utterances": ["echo test"], "lang": session.lang}, 

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

86 Message("speak", 

87 data={"utterance": "echo test", 

88 "expect_response": False, 

89 "lang": session.lang, 

90 "meta": { 

91 "skill": self.skill_id 

92 }}, 

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

94 Message("skill.converse.response", 

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

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

97 Message("ovos.utterance.handled", 

98 data={}, 

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

100 ] 

101 expected3 = [ 

102 message3, 

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

104 data={"utterances": ["stop parrot"], "skill_id": self.skill_id}, 

105 context={}), 

106 Message("skill.converse.pong", 

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

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

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

110 data={}, 

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

112 

113 Message("converse:skill", 

114 data={"utterances": ["stop parrot"], "lang": session.lang, "skill_id": self.skill_id}, 

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

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

117 data={"utterances": ["stop parrot"], "lang": session.lang}, 

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

119 

120 Message("speak", 

121 data={"expect_response": False, 

122 "lang": session.lang, 

123 "meta": { 

124 "dialog": "parrot_stop", 

125 "data": {}, 

126 "skill": self.skill_id 

127 }}, 

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

129 Message("skill.converse.response", 

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

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

132 Message("ovos.utterance.handled", 

133 data={}, 

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

135 ] 

136 expected4 = [ 

137 message4, 

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

139 data={"utterances": ["echo test"], "skill_id": self.skill_id}, 

140 context={}), 

141 Message("skill.converse.pong", 

142 data={"can_handle": False, "skill_id": self.skill_id}, 

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

144 Message("mycroft.audio.play_sound", data={"uri": "snd/error.mp3"}), 

145 Message("complete_intent_failure"), 

146 Message("ovos.utterance.handled") 

147 ] 

148 

149 final_session = deepcopy(session) 

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

151 

152 test = End2EndTest( 

153 minicroft=self.minicroft, 

154 skill_ids=[self.skill_id], 

155 eof_msgs=["ovos.utterance.handled"], 

156 flip_points=["recognizer_loop:utterance"], 

157 final_session=final_session, 

158 source_message=[message1, message2, message3, message4], 

159 expected_messages=expected1 + expected2 + expected3 + expected4, 

160 activation_points=[f"{self.skill_id}:start_parrot.intent"], 

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

162 keep_original_src=[f"{self.skill_id}.converse.ping", 

163 f"{self.skill_id}.converse.request" 

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

165 ] 

166 ) 

167 test.execute(timeout=10)