add tests for lifo

Tests/brains/lifo_buffer_test_brain.yml

@@ -0,0 +1,63 @@
+- name: "synapse1"
+  signals:
+    - order: "enter in synapse 1"
+  neurons:
+    - say:
+        message: "question in synapse 1"
+    - neurotransmitter:
+        from_answer_link:
+          - synapse: "synapse2"
+            answers:
+              - "answer synapse1"
+          - synapse: "synapse3"
+            answers:
+              - "other answer synapse1"
+        default: "synapse4"
+    - say:
+        message: "last neuron in synapse 1"
+
+- name: "synapse2"
+  signals:
+    - order: "synapse2"
+  neurons:
+    - say:
+        message: "enter synapse 2"
+    - neurotransmitter:
+        from_answer_link:
+          - synapse: "synapse5"
+            answers:
+              - "synapse5"
+        default: "synapse4"
+
+- name: "synapse3"
+  signals:
+    - order: "synapse3"
+  neurons:
+    - say:
+        message: "I enter in synapse 3"
+
+- name: "synapse4"
+  signals:
+    - order: "synapse4"
+  neurons:
+    - say:
+        message: "not understood"
+
+
+- name: "synapse5"
+  signals:
+    - order: "synapse5"
+  neurons:
+    - say:
+        message: "execution of synapse 5"
+
+- name: "synapse6"
+  signals:
+    - order: "synapse6"
+  neurons:
+    - neurotransmitter:
+        from_answer_link:
+          - synapse: "synapse5"
+            answers:
+              - "synapse 6 answer"
+        default: "synapse4"

Tests/test_lifo_buffer.py

@@ -0,0 +1,242 @@
+import os
+import unittest
+
+import mock
+
+from kalliope.core import LIFOBuffer
+from kalliope.core.ConfigurationManager import BrainLoader
+from kalliope.core.LIFOBuffer import Serialize, SynapseListAddedToLIFO
+
+from kalliope.core.Models import Singleton
+from kalliope.core.Models.MatchedSynapse import MatchedSynapse
+
+
+class TestLIFOBuffer(unittest.TestCase):
+
+    def setUp(self):
+        # be sure the brain haven't been instantiated before
+        Singleton._instances = dict()
+
+        if "/Tests" in os.getcwd():
+            self.brain_to_test = os.getcwd() + os.sep + "brains/lifo_buffer_test_brain.yml"
+        else:
+            self.brain_to_test = os.getcwd() + os.sep + "Tests/brains/lifo_buffer_test_brain.yml"
+
+        BrainLoader(file_path=self.brain_to_test)
+        # create a new lifo buffer
+        self.lifo_buffer = LIFOBuffer()
+        self.lifo_buffer.clean()
+
+    def test_execute(self):
+        """
+        In this test the brain contains a neurotransmitter
+        :return: 
+        """
+        # --------------------------------------
+        # Test 1. The user answers correctly to all neurotransmitter
+        # --------------------------------------
+        # we suppose that the first synapse has matched the first synapse
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
+        order = "enter in synapse 1"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+        list_matched_synapse = list()
+        list_matched_synapse.append(matched_synapse)
+        self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
+        self.lifo_buffer.api_response.user_order = order
+
+        with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
+
+            response = self.lifo_buffer.execute(is_api_call=True)
+
+            expected_result = {'status': 'waiting_for_answer', 'matched_synapses': [
+                {'matched_order': 'enter in synapse 1', 'neuron_module_list':
+                    [{'neuron_name': 'Say', 'generated_message': 'question in synapse 1'}],
+                 'synapse_name': 'synapse1'}], 'user_order': 'enter in synapse 1'}
+
+            self.assertEqual(response, expected_result)
+
+            # give an answer
+            answer = "answer synapse1"
+            response = self.lifo_buffer.execute(answer=answer,
+                                                is_api_call=True)
+            expected_result = {'status': 'waiting_for_answer', 'matched_synapses': [
+                {'matched_order': 'enter in synapse 1',
+                 'neuron_module_list': [{'neuron_name': 'Say',
+                                         'generated_message': 'question in synapse 1'},
+                                        {'neuron_name': 'Neurotransmitter', 'generated_message': None}],
+                 'synapse_name': 'synapse1'},
+                {'matched_order': 'answer synapse1', 'neuron_module_list': [
+                    {'neuron_name': 'Say', 'generated_message': 'enter synapse 2'}],
+                 'synapse_name': 'synapse2'}], 'user_order': None}
+            self.assertEqual(response, expected_result)
+
+            # give the last answer
+            answer = "synapse5"
+            response = self.lifo_buffer.execute(answer=answer,
+                                                is_api_call=True)
+            expected_result = {'status': 'complete', 'matched_synapses': [
+                {'matched_order': 'answer synapse1', 'neuron_module_list': [
+                    {'neuron_name': 'Say', 'generated_message': 'enter synapse 2'},
+                    {'neuron_name': 'Neurotransmitter', 'generated_message': None}],
+                 'synapse_name': 'synapse2'}, {'matched_order': 'synapse5', 'neuron_module_list': [
+                    {'neuron_name': 'Say', 'generated_message': 'execution of synapse 5'}],
+                                               'synapse_name': 'synapse5'},
+                {'matched_order': 'enter in synapse 1', 'neuron_module_list':
+                    [{'neuron_name': 'Say', 'generated_message': 'question in synapse 1'},
+                     {'neuron_name': 'Neurotransmitter', 'generated_message': None},
+                     {'neuron_name': 'Say', 'generated_message': 'last neuron in synapse 1'}],
+                 'synapse_name': 'synapse1'}], 'user_order': None}
+
+            self.assertEqual(response, expected_result)
+
+        # --------------------------------------
+        # Test 2. The user doesn't answered correctly to the first neurotransmitter
+        # --------------------------------------
+        # we suppose that the first synapse has matched the first synapse
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
+        order = "enter in synapse 1"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+        list_matched_synapse = list()
+        list_matched_synapse.append(matched_synapse)
+        self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
+        self.lifo_buffer.api_response.user_order = order
+
+        with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
+            # fist call to enter in the neurotransmitter
+            self.lifo_buffer.execute(is_api_call=True)
+
+            wrong_answer = "wrong answer"
+            response = self.lifo_buffer.execute(answer=wrong_answer, is_api_call=True)
+
+            expected_result = {'status': 'complete',
+                               'matched_synapses': [
+                                   {'matched_order': 'enter in synapse 1',
+                                    'neuron_module_list': [
+                                        {'neuron_name': 'Say', 'generated_message': 'question in synapse 1'},
+                                        {'neuron_name': 'Neurotransmitter', 'generated_message': None},
+                                        {'neuron_name': 'Say', 'generated_message': 'last neuron in synapse 1'}],
+                                    'synapse_name': 'synapse1'},
+                                   {'matched_order': None, 'neuron_module_list': [
+                                       {'neuron_name': 'Say',
+                                        'generated_message': 'not understood'}],
+                                    'synapse_name': 'synapse4'}], 'user_order': None}
+
+            self.assertEqual(response, expected_result)
+
+        # --------------------------------------
+        # Test 3. No synapse matched, we still execute the list
+        # --------------------------------------
+        list_matched_synapse = list()
+        self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
+        self.lifo_buffer.api_response.user_order = "this is an order"
+
+        with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
+            # fist call to enter in the neurotransmitter
+            response = self.lifo_buffer.execute(is_api_call=True)
+
+            expected_result = {'status': None, 'matched_synapses': [], 'user_order': 'this is an order'}
+
+            self.assertEqual(response, expected_result)
+
+    def test_add_synapse_list_to_lifo(self):
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
+        order = "enter in synapse 1"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+        list_matched_synapse = list()
+        list_matched_synapse.append(matched_synapse)
+        self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
+
+        self.assertEqual(self.lifo_buffer.lifo_list, [list_matched_synapse])
+
+    def test_clean(self):
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
+        order = "enter in synapse 1"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+        list_matched_synapse = list()
+        list_matched_synapse.append(matched_synapse)
+        self.lifo_buffer.add_synapse_list_to_lifo(list_matched_synapse)
+
+        self.lifo_buffer.clean()
+        self.assertEqual(0, len(self.lifo_buffer.lifo_list))
+
+    def test_return_serialized_api_response(self):
+        self.lifo_buffer.clean()
+        self.lifo_buffer.execute(is_api_call=True)
+        expected_result = {'status': None, 'matched_synapses': [], 'user_order': None}
+        response = self.lifo_buffer._return_serialized_api_response()
+        self.assertEqual(expected_result, response)
+
+    def test_process_synapse_list(self):
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse1")
+        order = "enter in synapse 1"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+        list_matched_synapse = list()
+        list_matched_synapse.append(matched_synapse)
+
+        with mock.patch("kalliope.core.LIFOBuffer._process_neuron_list"):
+            self.lifo_buffer._process_synapse_list(list_matched_synapse)
+            expected_response = {'status': None, 'matched_synapses': [
+                {'matched_order': 'enter in synapse 1',
+                 'neuron_module_list': [], 'synapse_name': 'synapse1'}],
+                                 'user_order': None}
+            self.assertEqual(expected_response, self.lifo_buffer.api_response.serialize())
+            self.assertEqual(0, len(self.lifo_buffer.lifo_list))
+
+    def test_process_neuron_list(self):
+        # Test with a neuron that doesn't wait for an answer
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse5")
+        order = "synapse5"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+
+        with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
+            self.lifo_buffer.set_api_call(True)
+            self.lifo_buffer._process_neuron_list(matched_synapse=matched_synapse)
+            self.assertEqual("complete", self.lifo_buffer.api_response.status)
+
+        # test with neuron that wait for an answer
+        self.lifo_buffer.clean()
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse6")
+        order = "synapse6"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+
+        self.lifo_buffer.set_api_call(True)
+        with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
+            with self.assertRaises(Serialize):
+                self.lifo_buffer._process_neuron_list(matched_synapse=matched_synapse)
+
+        # test with a neuron that want to add a synapse list to the LIFO
+        self.lifo_buffer.clean()
+        synapse = BrainLoader().brain.get_synapse_by_name("synapse6")
+        order = "synapse6"
+        matched_synapse = MatchedSynapse(matched_synapse=synapse,
+                                         user_order=order,
+                                         matched_order=order)
+
+        self.lifo_buffer.set_api_call(True)
+        self.lifo_buffer.set_answer("synapse 6 answer")
+        with mock.patch("kalliope.core.TTS.TTSModule.generate_and_play"):
+            with self.assertRaises(SynapseListAddedToLIFO):
+                self.lifo_buffer._process_neuron_list(matched_synapse=matched_synapse)
+
+
+if __name__ == '__main__':
+    unittest.main()
+
+    # suite = unittest.TestSuite()
+    # suite.addTest(TestLIFOBuffer("test_process_neuron_list"))
+    # runner = unittest.TextTestRunner()
+    # runner.run(suite)

Tests/test_synapse_launcher.py

@@ -1,7 +1,6 @@
 import unittest
 
 import mock
-from kalliope.core.ConfigurationManager import BrainLoader
 
 from kalliope.core import LIFOBuffer
 from kalliope.core.Models import Brain
@@ -118,7 +117,7 @@ class TestSynapseLauncher(unittest.TestCase):
         LIFOBuffer.lifo_list = list()
         with mock.patch("kalliope.core.LIFOBuffer.execute"):
             order_to_match = "not existing sentence"
-            new_settings= Settings()
+            new_settings = Settings()
             expected_result = [[]]
             SynapseLauncher.run_matching_synapse_from_order(order_to_match,
                                                             brain=self.brain_test,

kalliope/core/LIFOBuffer.py

@@ -40,6 +40,14 @@ class LIFOBuffer(object):
     answer = None
     is_api_call = False
 
+    @classmethod
+    def set_answer(cls, value):
+        cls.answer = value
+
+    @classmethod
+    def set_api_call(cls, value):
+        cls.is_api_call = value
+
     @classmethod
     def add_synapse_list_to_lifo(cls, matched_synapse_list):
         """
@@ -56,6 +64,7 @@ class LIFOBuffer(object):
         Clean the LIFO by creating a new list
         """
         cls.lifo_list = list()
+        cls.api_response = APIResponse()
 
     @classmethod
     def _return_serialized_api_response(cls):
@@ -104,7 +113,7 @@ class LIFOBuffer(object):
             raise Serialize
 
         except Serialize:
-            cls._return_serialized_api_response()
+            return cls._return_serialized_api_response()
 
     @classmethod
     def _process_synapse_list(cls, synapse_list):
@@ -124,9 +133,8 @@ class LIFOBuffer(object):
             # in the response
             if matched_synapse not in cls.api_response.list_processed_matched_synapse:
                 cls.api_response.list_processed_matched_synapse.append(matched_synapse)
-            # while we have synapse to process in the list of synapse
-            while matched_synapse.neuron_fifo_list:
-                cls._process_neuron_list(matched_synapse=matched_synapse)
+
+            cls._process_neuron_list(matched_synapse=matched_synapse)
 
             # The synapse has been processed we can remove it from the list.
             synapse_list.remove(matched_synapse)
@@ -147,37 +155,40 @@ class LIFOBuffer(object):
         :return: 
         """
         logger.debug("[LIFOBuffer] number of neuron to process: %s" % len(matched_synapse.neuron_fifo_list))
-        # get the first neuron in the FIFO neuron list
-        neuron = matched_synapse.neuron_fifo_list[0]
-        # from here, we are back into the last neuron we were processing.
-        if cls.answer is not None:  # we give the answer if exist to the first neuron
-            neuron.parameters["answer"] = cls.answer
-            # the next neuron should not get this answer
-            cls.answer = None
-        # todo fix this when we have a full client/server call. The client would be the voice or api call
-        neuron.parameters["is_api_call"] = cls.is_api_call
-        # execute the neuron
-        instantiated_neuron = NeuronLauncher.start_neuron(neuron=neuron,
-                                                          parameters_dict=matched_synapse.parameters)
-
-        # the status of an execution is "complete" if no neuron are waiting for an answer
-        cls.api_response.status = "complete"
-        if instantiated_neuron.is_waiting_for_answer:  # the neuron is waiting for an answer
-            logger.debug("[LIFOBuffer] Wait for answer mode")
-            cls.api_response.status = "waiting_for_answer"
-            raise Serialize
-        else:
-            logger.debug("[LIFOBuffer] complete mode")
-            # we add the instantiated neuron to the neuron_module_list.
-            # This one contains info about generated text
-            matched_synapse.neuron_module_list.append(instantiated_neuron)
-            # the neuron is fully processed we can remove it from the list
-            matched_synapse.neuron_fifo_list.remove(neuron)
-
-        if instantiated_neuron.pending_synapse:  # the last executed neuron want to run a synapse
-            logger.debug("[LIFOBuffer] Last executed neuron want to run a synapse. Restart the LIFO")
-            # add the synapse to the lifo (inside a list as expected by the lifo)
-            cls.add_synapse_list_to_lifo([instantiated_neuron.pending_synapse])
-            # we have added a list of synapse to the LIFO ! this one must start over.
-            # break all while loop until the execution is back to the LIFO loop
-            raise SynapseListAddedToLIFO
+        # while we have synapse to process in the list of synapse
+        while matched_synapse.neuron_fifo_list:
+            # get the first neuron in the FIFO neuron list
+            neuron = matched_synapse.neuron_fifo_list[0]
+            # from here, we are back into the last neuron we were processing.
+            if cls.answer is not None:  # we give the answer if exist to the first neuron
+                neuron.parameters["answer"] = cls.answer
+                # the next neuron should not get this answer
+                cls.answer = None
+            # todo fix this when we have a full client/server call. The client would be the voice or api call
+            neuron.parameters["is_api_call"] = cls.is_api_call
+            logger.debug("[LIFOBuffer] process_neuron_list: is_api_call: %s" % cls.is_api_call)
+            # execute the neuron
+            instantiated_neuron = NeuronLauncher.start_neuron(neuron=neuron,
+                                                              parameters_dict=matched_synapse.parameters)
+
+            # the status of an execution is "complete" if no neuron are waiting for an answer
+            cls.api_response.status = "complete"
+            if instantiated_neuron.is_waiting_for_answer:  # the neuron is waiting for an answer
+                logger.debug("[LIFOBuffer] Wait for answer mode")
+                cls.api_response.status = "waiting_for_answer"
+                raise Serialize
+            else:
+                logger.debug("[LIFOBuffer] complete mode")
+                # we add the instantiated neuron to the neuron_module_list.
+                # This one contains info about generated text
+                matched_synapse.neuron_module_list.append(instantiated_neuron)
+                # the neuron is fully processed we can remove it from the list
+                matched_synapse.neuron_fifo_list.remove(neuron)
+
+            if instantiated_neuron.pending_synapse:  # the last executed neuron want to run a synapse
+                logger.debug("[LIFOBuffer] Last executed neuron want to run a synapse. Restart the LIFO")
+                # add the synapse to the lifo (inside a list as expected by the lifo)
+                cls.add_synapse_list_to_lifo([instantiated_neuron.pending_synapse])
+                # we have added a list of synapse to the LIFO ! this one must start over.
+                # break all while loop until the execution is back to the LIFO loop
+                raise SynapseListAddedToLIFO

kalliope/core/TTS/__init__.py

@@ -0,0 +1 @@
+from TTSModule import TTSModule