def test_replay_from_memory_message_store(self):
        """ We can store a message in FileMessageStore """

        store_factory = msgstore.MemoryMessageStoreFactory()

        chan = BaseChannel(name="test_channel10.5", loop=self.loop, message_store_factory=store_factory)

        n = TestNode()

        msg = generate_msg(with_context=True)
        msg2 = generate_msg(timestamp=(1982, 11, 27, 12, 35))


        chan.add(n)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))
        self.loop.run_until_complete(chan.handle(msg2))

        msg_stored = list(self.loop.run_until_complete(chan.message_store.search()))

        for msg in msg_stored:
            print(msg)

        self.assertEqual(len(msg_stored), 2, "Should be 2 messages in store!")

        self.loop.run_until_complete(chan.replay(msg_stored[0]['id']))

        msg_stored = list(self.loop.run_until_complete(chan.message_store.search()))
        self.assertEqual(len(msg_stored), 3, "Should be 3 messages in store!")

    def test_chan_node_mocking_in_test_mode(self):
        """ Whether node mocking input and output is working """

        chan = BaseChannel(name="test_channel_test_2", loop=self.loop)

        n = TestNode(name="testme")
        chan.add(n)

        msg_x = generate_msg(message_content="X")
        msg_a = generate_msg(message_content="A")
        msg_b = generate_msg(message_content="B")
        msg_c = generate_msg(message_content="C")
        msg_d = generate_msg(message_content="D")

        def concat_e(msg):
            msg.payload += "E"
            return msg

        def concat_f(msg):
            msg.payload += "F"
            return msg

        # Launch channel processing
        self.start_channels()

        # Mock input
        chan._reset_test()
        chan.get_node("testme").mock(input=msg_a)

        ret = chan.handle_and_wait(msg_x)

        self.assertEqual(n.processed, 1, "Channel in test mode not working")
        self.assertEqual(ret.payload, "A", "Mocking input broken")
        self.assertEqual(chan.get_node("testme").last_input().payload, "X", "Last input broken")

        # Mock output
        chan._reset_test()
        chan.get_node("testme").mock(output=msg_b)
        ret = chan.handle_and_wait(msg_x)

        self.assertEqual(n.processed, 1, "Channel in test mode not working")
        self.assertEqual(ret.payload, "B", "Mocking input broken")
        self.assertEqual(chan.get_node("testme").last_input().payload, "X", "Last input broken")

        # Mock both
        chan._reset_test()
        chan.get_node("testme").mock(input=msg_c, output=msg_d)
        ret = chan.handle_and_wait(msg_x)

        self.assertEqual(n.processed, 1, "Channel in test mode not working")
        self.assertEqual(ret.payload, "D", "Mocking both input and output broken")

        # Mock both functions
        chan._reset_test()
        chan.get_node("testme").mock(input=concat_e, output=concat_f)
        ret = chan.handle_and_wait(msg_x)

        self.assertEqual(n.processed, 1, "Channel in test mode not working")
        self.assertEqual(ret.payload, "XEF", "Mocking with function broken")

    def test_file_reader_node(self):
        """if FileReader are functionnal"""

        reader = nodes.FileReader(filepath='/filepath', filename='badname')
        channel = FakeChannel(self.loop)

        reader.channel = channel
        msg1 = generate_msg()

        with mock.patch("builtins.open", mock.mock_open(read_data="data")) as mock_file:
            result = self.loop.run_until_complete(reader.handle(msg1))

        mock_file.assert_called_once_with('/filepath', 'r')
        self.assertEqual(result.payload, "data", "FileReader not working")

        reader2 = nodes.FileReader()
        reader2.channel = channel
        msg2 = generate_msg()
        msg2.meta['filepath'] = '/filepath2'
        msg2.meta['filename'] = '/badpath'

        with mock.patch("builtins.open", mock.mock_open(read_data="data2")) as mock_file:
            result = self.loop.run_until_complete(reader2.handle(msg2))

        mock_file.assert_called_once_with('/filepath2', 'r')
        self.assertEqual(result.payload, "data2", "FileReader not working with meta")


        reader3 = nodes.FileReader(filepath=tstfct, filename='badname')
        reader3.channel = channel

        msg3 = generate_msg()
        msg3.meta['filepath'] = '/badpath'
        msg3.meta['filename'] = 'badname2'

        with mock.patch("builtins.open", mock.mock_open(read_data="data")) as mock_file:
            result = self.loop.run_until_complete(reader3.handle(msg3))

        mock_file.assert_called_once_with('/fctpath', 'r')

        reader4 = nodes.FileReader(filename=tstfct2)
        reader4.channel = channel
        msg4 = generate_msg()
        msg4.meta['filepath'] = '/filepath3/badname'
        msg4.meta['filename'] = 'badname'

        with mock.patch("builtins.open", mock.mock_open(read_data="data")) as mock_file:
            result = self.loop.run_until_complete(reader4.handle(msg4))

        mock_file.assert_called_once_with('/filepath3/fctname', 'r')

    def test_sub_channel_with_exception(self):
        """ Whether Sub Channel exception handling is working """

        chan = BaseChannel(name="test_channel4", loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="sub")
        n3 = ExceptNode(name="sub2")

        msg = generate_msg()

        chan.add(n1)
        sub = chan.fork(name="Hello")
        sub.add(n2, n3)

        # Launch channel processing
        self.start_channels()

        self.loop.run_until_complete(chan.handle(msg))

        self.assertEqual(n1.processed, 1, "Sub Channel not working")

        with self.assertRaises(TestException) as cm:
            self.clean_loop()

        self.assertEqual(n2.processed, 1, "Sub Channel not working")

    def test_channel_with_generator(self):
        """ Whether BaseChannel with generator is working """

        chan = BaseChannel(name="test_channel7.3", loop=self.loop)
        chan2 = BaseChannel(name="test_channel7.31", loop=self.loop)
        msg = generate_msg()
        msg2 = msg.copy()

        class TestIter(nodes.BaseNode):
            def process(self, msg):
                def iter():
                    for i in range(3):
                        yield msg
                return iter()

        final_node = nodes.Log()
        mid_node = nodes.Log()

        chan.add(TestIter(name="testiterr"), nodes.Log(), TestIter(name="testiterr"), final_node)
        chan2.add(TestIter(name="testiterr"), mid_node, nodes.Drop())

        # Launch channel processing
        self.start_channels()
        result = self.loop.run_until_complete(chan.handle(msg))

        self.assertEqual(result.payload, msg.payload, "Generator node not working")
        self.assertEqual(final_node.processed, 9, "Generator node not working")

        result = self.loop.run_until_complete(chan2.handle(msg2))

        self.assertEqual(mid_node.processed, 3, "Generator node not working with drop_node")

    def test_node_persistence(self):
        """ Whether BaseNode() data memory persistence works"""
        persistence._backend = None # noqa

        # Mock settings
        with mock.patch('pypeman.persistence.settings',
                        conf.Settings(module_name='pypeman.tests.settings.test_settings_persistence')):
            result = []

            class TestPers(nodes.BaseNode):
                """ Test node for persistence """

                async def process(self, msg):
                    await self.save_data('test', 'value')
                    result.append(await self.restore_data('test'))
                    result.append(await self.restore_data('titi', default='Yo'))
                    try:
                        result.append(await self.restore_data('titi'))
                    except KeyError:
                        result.append('yay')

                    return msg

            n = TestPers()

            n.channel = FakeChannel(self.loop)

            m = generate_msg(message_content='test')

            self.loop.run_until_complete(n.handle(m))

            self.assertEqual(result[0], 'value', "Can't persist data for node")
            self.assertEqual(result[1], 'Yo', "Default value not working")
            self.assertEqual(result[2], 'yay', "Exception on missing key not working")

    def test_cond_channel(self):
        """ Whether Conditionnal channel is working """

        chan = BaseChannel(name="test_channel5", loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="end_main")
        not_processed = TestNode(name="cond_notproc")
        processed = TestNode(name="cond_proc")

        msg = generate_msg()

        chan.add(n1)

        # Nodes in this channel should not be processed
        cond1 = chan.when(lambda x: False, name="Toto")
        # Nodes in this channel should be processed
        cond2 = chan.when(True, name="condchannel")

        chan.add(n2)

        cond1.add(not_processed)
        cond2.add(processed)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))

        self.assertFalse(not_processed.processed, "Cond Channel when condition == False not working")
        self.assertTrue(processed.processed, "Cond Channel when condition == True not working")
        self.assertFalse(n2.processed, "Cond Channel don't became the main path")
        self.assertEqual(cond2.name, "test_channel5.condchannel", "Condchannel name is incorrect")

    def test_sub_channel(self):
        """ Whether Sub Channel is working """

        chan = BaseChannel(name="test_channel3", loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="sub")
        n3 = TestNode(name="sub1")
        n4 = TestNode(name="sub2")

        msg = generate_msg()

        same_chan = chan.append(n1)

        self.assertEqual(chan, same_chan, "Append don't return channel.")

        sub = chan.fork(name="subchannel")
        sub.append(n2, n3, n4)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))

        self.assertTrue(n2.processed, "Sub Channel not working")
        self.assertTrue(n3.processed, "Sub Channel not working")
        self.assertTrue(n4.processed, "Sub Channel not working")
        self.assertEqual(sub.name, "test_channel3.subchannel", "Subchannel name is incorrect")

    def test_xml_nodes(self):
        """ if XML nodes are functional """
        try:
            import xmltodict  # noqa F401
        except ImportError:
            raise unittest.SkipTest("Missing dependency xmltodict.")

        n1 = nodes.XMLToPython()
        n2 = nodes.PythonToXML()

        channel = FakeChannel(self.loop)

        n1.channel = channel
        n2.channel = channel

        m = generate_msg()

        m.payload = '<?xml version="1.0" encoding="utf-8"?>\n<test>hello</test>'

        base = str(m.payload)

        ret = self.loop.run_until_complete(n1.handle(m))
        ext_new = self.loop.run_until_complete(n2.handle(ret))
        # Check return
        self.assertTrue(isinstance(ret, message.Message))
        self.assertEqual(base, ext_new.payload, "XML nodes not working !")

    def test_loop_slow2(self):
        """ slow tasks delay can be configured """
        import pypeman.debug
        pypeman.debug.enable_slow_log_stats()
        self.setup_evt_loop(slow_cb_duration=0.05)
        stats = pypeman.debug.stats

        # logger = self.logger # get default test logger
        # handler = self.loghandler # get test log handler

        tst_logger = logging.getLogger('tests.debug.main_loop.slow')
        print(tst_logger.handlers)

        loop = self.loop
        chan = BaseChannel(name="test_loop_slow2", loop=loop)
        n1 = SimpleTestNode(delay=0.03, async_delay=0, logger=tst_logger, loop=loop)
        n2 = SimpleTestNode(delay=0.06, logger=tst_logger, loop=loop)
        chan.add(n1)
        chan.add(n2)

        msg = generate_msg()
        # Launch channel processing
        self.loop.run_until_complete(chan.start())
        self.loop.run_until_complete(chan.handle(msg))
        # handler.show_entries()
        self.assertEqual(len(stats), 1, "should have 1 slow tasks, not %d" % len(stats))
        pypeman.debug.show_slow_log_stats()
        self.loop.close()

    def test_channel_events(self):
        """ Whether BaseChannel handling return a good result """

        chan = BaseChannel(name="test_channel7.5", loop=self.loop)
        msg = generate_msg()

        chan.add(nodes.JsonToPython(), nodes.PythonToJson())

        state_sequence = [chan.status]

        @events.channel_change_state.receiver
        def handle_change_state(channel=None, old_state=None, new_state=None):
            print(channel.name, old_state, new_state)
            state_sequence.append(new_state)

        @events.channel_change_state.receiver
        async def handle_change_state_async(channel=None, old_state=None, new_state=None):
            print(channel.name, old_state, new_state)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))
        self.loop.run_until_complete(chan.stop())

        print(state_sequence)

        valid_sequence = [BaseChannel.STOPPED, BaseChannel.STARTING, BaseChannel.WAITING,
                          BaseChannel.PROCESSING, BaseChannel.WAITING, BaseChannel.STOPPING, BaseChannel.STOPPED]
        self.assertEqual(state_sequence, valid_sequence, "Sequence state is not valid")

    def test_case_channel(self):
        """ Whether Conditionnal channel is working """

        chan = BaseChannel(name="test_channel6", loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="end_main")
        not_processed = TestNode(name="cond_notproc")
        processed = TestNode(name="cond_proc")
        not_processed2 = TestNode(name="cond_proc2")

        msg = generate_msg()

        chan.add(n1)

        cond1, cond2, cond3 = chan.case(lambda x: False, True, True, names=['first', 'second', 'third'])

        chan.add(n2)

        cond1.add(not_processed)
        cond2.add(processed)
        cond3.add(not_processed2)

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))

        self.assertFalse(not_processed.processed, "Case Channel when condition == False not working")
        self.assertFalse(not_processed2.processed, "Case Channel when condition == False not working")
        self.assertTrue(processed.processed, "Case Channel when condition == True not working")
        self.assertTrue(n2.processed, "Cond Channel don't became the main path")
        self.assertEqual(cond1.name, "test_channel6.first", "Casechannel name is incorrect")
        self.assertEqual(cond2.name, "test_channel6.second", "Casechannel name is incorrect")
        self.assertEqual(cond3.name, "test_channel6.third", "Casechannel name is incorrect")

    def test_no_node_base_channel(self):
        """ Whether BaseChannel handling is working even if there is no node """

        chan = BaseChannel(name="test_channel2", loop=self.loop)
        msg = generate_msg()

        # Launch channel processing
        self.start_channels()
        self.loop.run_until_complete(chan.handle(msg))

    def test_ftp_nodes(self):
        """ Whether FTP nodes are functional """

        channel = FakeChannel(self.loop)

        ftp_config = dict(host="fake", credentials=("fake", "fake"))

        fake_ftp = mock.MagicMock()
        fake_ftp.download_file = mock.Mock(return_value=b"new_content")

        fake_ftp_helper = mock.Mock(return_value=fake_ftp)

        with mock.patch('pypeman.contrib.ftp.FTPHelper', new=fake_ftp_helper):

            reader = nodes.FTPFileReader(filepath="test_read", **ftp_config)
            delete = nodes.FTPFileDeleter(filepath="test_delete", **ftp_config)

            writer = nodes.FTPFileWriter(filepath="test_write", **ftp_config)

            reader.channel = channel
            delete.channel = channel
            writer.channel = channel

            m1 = generate_msg(message_content="to_be_replaced")
            m1_delete = generate_msg(message_content="to_be_replaced")
            m2 = generate_msg(message_content="message_content")

            # Test reader
            result = self.loop.run_until_complete(reader.handle(m1))

            fake_ftp.download_file.assert_called_once_with('test_read')
            self.assertEqual(result.payload, b"new_content", "FTP reader not working")

            # Test reader with delete after
            result = self.loop.run_until_complete(delete.handle(m1_delete))

            fake_ftp.delete.assert_called_once_with('test_delete')

            # test writer
            result = self.loop.run_until_complete(writer.handle(m2))
            fake_ftp.upload_file.assert_called_once_with('test_write.part', 'message_content')
            fake_ftp.rename.assert_called_once_with('test_write.part', 'test_write')

    def test_log_node(self):
        """ whether Log() node functional """

        n = nodes.Log()
        n.channel = FakeChannel(self.loop)

        m = generate_msg()

        ret = self.loop.run_until_complete(n.handle(m))

        self.assertTrue(isinstance(ret, message.Message))

    def test_channel_exception(self):
        """ Whether BaseChannel handling return an exception in case of error in main branch """

        chan = BaseChannel(name="test_channel8", loop=self.loop)
        msg = generate_msg()

        chan.add(nodes.JsonToPython(), nodes.PythonToJson(), ExceptNode())

        # Launch channel processing
        self.start_channels()
        with self.assertRaises(TestException) as cm:
            self.loop.run_until_complete(chan.process(msg))

    def test_json_to_python_node(self):
        """ if JsonToPython() node functional """

        n = nodes.JsonToPython()
        n.channel = FakeChannel(self.loop)

        m = generate_msg(message_content='{"test":1}')

        ret = self.loop.run_until_complete(n.handle(m))
        # Check return
        self.assertTrue(isinstance(ret, message.Message))
        self.assertEqual(ret.payload, {"test": 1}, "JsonToPython node not working !")

    def test_channel_result(self):
        """ Whether BaseChannel handling return a good result """

        chan = BaseChannel(name="test_channel7", loop=self.loop)
        msg = generate_msg()

        chan.add(nodes.JsonToPython(), nodes.PythonToJson())

        # Launch channel processing
        self.start_channels()
        result = self.loop.run_until_complete(chan.handle(msg))

        self.assertEqual(result.payload, msg.payload, "Channel handle not working")

    def test_sleep_node(self):
        """ if Sleep() node functional """

        n = nodes.Sleep()
        n.channel = FakeChannel(self.loop)

        m = generate_msg(message_content='test')

        ret = self.loop.run_until_complete(n.handle(m))

        # Check return
        self.assertTrue(isinstance(ret, message.Message))
        self.assertEqual(ret.payload, 'test', "Sleep node not working !")

    def test_thread_node(self):
        """ if Thread node functional """

        # TODO test if another task can be executed in //

        n = LongNode()
        n.channel = FakeChannel(self.loop)

        m = generate_msg()

        ret = self.loop.run_until_complete(n.handle(m))
        # Check return
        self.assertTrue(isinstance(ret, message.Message))