test_lifo_buffer.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242
  1. import os
  2. import unittest
  3. import mock
  4. from kalliope.core import LIFOBuffer
  5. from kalliope.core.ConfigurationManager import BrainLoader
  6. from kalliope.core.LIFOBuffer import Serialize, SynapseListAddedToLIFO
  7. from kalliope.core.Models import Singleton
  8. from kalliope.core.Models.MatchedSynapse import MatchedSynapse
  9. class TestLIFOBuffer(unittest.TestCase):
  10. def setUp(self):
  11. # be sure the brain haven't been instantiated before
  12. Singleton._instances = dict()
  13. if "/Tests" in os.getcwd():
  14. self.brain_to_test = os.getcwd() + os.sep + "brains/lifo_buffer_test_brain.yml"
  15. else:
  16. self.brain_to_test = os.getcwd() + os.sep + "Tests/brains/lifo_buffer_test_brain.yml"
  17. BrainLoader(file_path=self.brain_to_test)
  18. # create a new lifo buffer
  19. self.lifo_buffer = LIFOBuffer()
  20. self.lifo_buffer.clean()
  21. def test_execute(self):
  22. """
  23. In this test the brain contains a neurotransmitter
  24. :return:
  25. """
  26. # --------------------------------------
  27. # Test 1. The user answers correctly to all neurotransmitter
  28. # --------------------------------------
  29. # we suppose that the first synapse has matched the first synapse
  30. synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
  31. order = "enter in synapse 1"
  32. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  33. user_order=order,
  34. matched_order=order)
  35. list_matched_synapse = list()
  36. list_matched_synapse.append(matched_synapse)
  37. self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
  38. self.lifo_buffer.api_response.user_order = order
  39. with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
  40. response = self.lifo_buffer.execute(is_api_call=True)
  41. expected_result = {'status': 'waiting_for_answer', 'matched_synapses': [
  42. {'matched_order': 'enter in synapse 1', 'neuron_module_list':
  43. [{'neuron_name': 'Say', 'generated_message': 'question in synapse 1'}],
  44. 'synapse_name': 'synapse1'}], 'user_order': 'enter in synapse 1'}
  45. self.assertEqual(response, expected_result)
  46. # give an answer
  47. answer = "answer synapse1"
  48. response = self.lifo_buffer.execute(answer=answer,
  49. is_api_call=True)
  50. expected_result = {'status': 'waiting_for_answer', 'matched_synapses': [
  51. {'matched_order': 'enter in synapse 1',
  52. 'neuron_module_list': [{'neuron_name': 'Say',
  53. 'generated_message': 'question in synapse 1'},
  54. {'neuron_name': 'Neurotransmitter', 'generated_message': None}],
  55. 'synapse_name': 'synapse1'},
  56. {'matched_order': 'answer synapse1', 'neuron_module_list': [
  57. {'neuron_name': 'Say', 'generated_message': 'enter synapse 2'}],
  58. 'synapse_name': 'synapse2'}], 'user_order': None}
  59. self.assertEqual(response, expected_result)
  60. # give the last answer
  61. answer = "synapse5"
  62. response = self.lifo_buffer.execute(answer=answer,
  63. is_api_call=True)
  64. expected_result = {'status': 'complete', 'matched_synapses': [
  65. {'matched_order': 'answer synapse1', 'neuron_module_list': [
  66. {'neuron_name': 'Say', 'generated_message': 'enter synapse 2'},
  67. {'neuron_name': 'Neurotransmitter', 'generated_message': None}],
  68. 'synapse_name': 'synapse2'}, {'matched_order': 'synapse5', 'neuron_module_list': [
  69. {'neuron_name': 'Say', 'generated_message': 'execution of synapse 5'}],
  70. 'synapse_name': 'synapse5'},
  71. {'matched_order': 'enter in synapse 1', 'neuron_module_list':
  72. [{'neuron_name': 'Say', 'generated_message': 'question in synapse 1'},
  73. {'neuron_name': 'Neurotransmitter', 'generated_message': None},
  74. {'neuron_name': 'Say', 'generated_message': 'last neuron in synapse 1'}],
  75. 'synapse_name': 'synapse1'}], 'user_order': None}
  76. self.assertEqual(response, expected_result)
  77. # --------------------------------------
  78. # Test 2. The user doesn't answered correctly to the first neurotransmitter
  79. # --------------------------------------
  80. # we suppose that the first synapse has matched the first synapse
  81. synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
  82. order = "enter in synapse 1"
  83. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  84. user_order=order,
  85. matched_order=order)
  86. list_matched_synapse = list()
  87. list_matched_synapse.append(matched_synapse)
  88. self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
  89. self.lifo_buffer.api_response.user_order = order
  90. with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
  91. # fist call to enter in the neurotransmitter
  92. self.lifo_buffer.execute(is_api_call=True)
  93. wrong_answer = "wrong answer"
  94. response = self.lifo_buffer.execute(answer=wrong_answer, is_api_call=True)
  95. expected_result = {'status': 'complete',
  96. 'matched_synapses': [
  97. {'matched_order': 'enter in synapse 1',
  98. 'neuron_module_list': [
  99. {'neuron_name': 'Say', 'generated_message': 'question in synapse 1'},
  100. {'neuron_name': 'Neurotransmitter', 'generated_message': None},
  101. {'neuron_name': 'Say', 'generated_message': 'last neuron in synapse 1'}],
  102. 'synapse_name': 'synapse1'},
  103. {'matched_order': None, 'neuron_module_list': [
  104. {'neuron_name': 'Say',
  105. 'generated_message': 'not understood'}],
  106. 'synapse_name': 'synapse4'}], 'user_order': None}
  107. self.assertEqual(response, expected_result)
  108. # --------------------------------------
  109. # Test 3. No synapse matched, we still execute the list
  110. # --------------------------------------
  111. list_matched_synapse = list()
  112. self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
  113. self.lifo_buffer.api_response.user_order = "this is an order"
  114. with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
  115. # fist call to enter in the neurotransmitter
  116. response = self.lifo_buffer.execute(is_api_call=True)
  117. expected_result = {'status': None, 'matched_synapses': [], 'user_order': 'this is an order'}
  118. self.assertEqual(response, expected_result)
  119. def test_add_synapse_list_to_lifo(self):
  120. synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
  121. order = "enter in synapse 1"
  122. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  123. user_order=order,
  124. matched_order=order)
  125. list_matched_synapse = list()
  126. list_matched_synapse.append(matched_synapse)
  127. self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
  128. self.assertEqual(self.lifo_buffer.lifo_list, [list_matched_synapse])
  129. def test_clean(self):
  130. synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
  131. order = "enter in synapse 1"
  132. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  133. user_order=order,
  134. matched_order=order)
  135. list_matched_synapse = list()
  136. list_matched_synapse.append(matched_synapse)
  137. self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
  138. self.lifo_buffer.clean()
  139. self.assertEqual(0, len(self.lifo_buffer.lifo_list))
  140. def test_return_serialized_api_response(self):
  141. self.lifo_buffer.clean()
  142. self.lifo_buffer.execute(is_api_call=True)
  143. expected_result = {'status': None, 'matched_synapses': [], 'user_order': None}
  144. response = self.lifo_buffer._return_serialized_api_response()
  145. self.assertEqual(expected_result, response)
  146. def test_process_synapse_list(self):
  147. synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
  148. order = "enter in synapse 1"
  149. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  150. user_order=order,
  151. matched_order=order)
  152. list_matched_synapse = list()
  153. list_matched_synapse.append(matched_synapse)
  154. with mock.patch("kalliope.core.LIFOBuffer._process_neuron_list"):
  155. self.lifo_buffer._process_synapse_list(list_matched_synapse)
  156. expected_response = {'status': None, 'matched_synapses': [
  157. {'matched_order': 'enter in synapse 1',
  158. 'neuron_module_list': [], 'synapse_name': 'synapse1'}],
  159. 'user_order': None}
  160. self.assertEqual(expected_response, self.lifo_buffer.api_response.serialize())
  161. self.assertEqual(0, len(self.lifo_buffer.lifo_list))
  162. def test_process_neuron_list(self):
  163. # Test with a neuron that doesn't wait for an answer
  164. synapse = BrainLoader().brain.get_synapse_by_name("synapse5")
  165. order = "synapse5"
  166. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  167. user_order=order,
  168. matched_order=order)
  169. with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
  170. self.lifo_buffer.set_api_call(True)
  171. self.lifo_buffer._process_neuron_list(matched_synapse=matched_synapse)
  172. self.assertEqual("complete", self.lifo_buffer.api_response.status)
  173. # test with neuron that wait for an answer
  174. self.lifo_buffer.clean()
  175. synapse = BrainLoader().brain.get_synapse_by_name("synapse6")
  176. order = "synapse6"
  177. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  178. user_order=order,
  179. matched_order=order)
  180. self.lifo_buffer.set_api_call(True)
  181. with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
  182. with self.assertRaises(Serialize):
  183. self.lifo_buffer._process_neuron_list(matched_synapse=matched_synapse)
  184. # test with a neuron that want to add a synapse list to the LIFO
  185. self.lifo_buffer.clean()
  186. synapse = BrainLoader().brain.get_synapse_by_name("synapse6")
  187. order = "synapse6"
  188. matched_synapse = MatchedSynapse(matched_synapse=synapse,
  189. user_order=order,
  190. matched_order=order)
  191. self.lifo_buffer.set_api_call(True)
  192. self.lifo_buffer.set_answer("synapse 6 answer")
  193. with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
  194. with self.assertRaises(SynapseListAddedToLIFO):
  195. self.lifo_buffer._process_neuron_list(matched_synapse=matched_synapse)
  196. if __name__ == '__main__':
  197. unittest.main()
  198. # suite = unittest.TestSuite()
  199. # suite.addTest(TestLIFOBuffer("test_process_neuron_list"))
  200. # runner = unittest.TextTestRunner()
  201. # runner.run(suite)