<commit_before>import unittest
import asyncio

from pypeman.channels import BaseChannel
from pypeman import message
from pypeman import nodes

message_content = """{"test":1}"""

class TestNode(nodes.BaseNode):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Used tu test if node is processed during test
        self.processed = False

    def process(self, msg):
        print("Process %s" % self.name)
        self.processed = True
        return msg


class ExceptNode(nodes.BaseNode):
    # This node raise an exception
    def process(self, msg):
        raise Exception()


def generate_msg():
    # Default message
    m = message.Message()
    m.payload = message_content

    return m


class ChannelsTests(unittest.TestCase):
    def setUp(self):
        # Create class event loop used for tests to avoid failing
        # previous tests to impact next test ? (Not shure)
        self.loop = asyncio.new_event_loop()
        # Remove thread event loop to be sure we are not using
        # another event loop somewhere
        asyncio.set_event_loop(None)

    def test_base_channel(self):
        """ if BaseChannel handling is working """

        chan = BaseChannel(loop=self.loop)
        n = TestNode()
        msg = generate_msg()

        chan.add(n)

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

        self.assertTrue(n.processed, "Channel handle not working")

    def test_sub_channel(self):
        """ if BaseChannel handling is working """

        chan = BaseChannel(loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="sub")

        msg = generate_msg()

        chan.add(n1)
        sub = chan.fork()
        sub.add(n2)

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

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

    def test_cond_channel(self):
        """ if BaseChannel handling is working """

        chan = BaseChannel(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()<commit_msg>Add test for non node bug<commit_after>import unittest
import asyncio

from pypeman.channels import BaseChannel
from pypeman import message
from pypeman import nodes

message_content = """{"test":1}"""

class TestNode(nodes.BaseNode):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Used tu test if node is processed during test
        self.processed = False

    def process(self, msg):
        print("Process %s" % self.name)
        self.processed = True
        return msg


class ExceptNode(nodes.BaseNode):
    # This node raise an exception
    def process(self, msg):
        raise Exception()


def generate_msg():
    # Default message
    m = message.Message()
    m.payload = message_content

    return m


class ChannelsTests(unittest.TestCase):
    def setUp(self):
        # Create class event loop used for tests to avoid failing
        # previous tests to impact next test ? (Not shure)
        self.loop = asyncio.new_event_loop()
        # Remove thread event loop to be sure we are not using
        # another event loop somewhere
        asyncio.set_event_loop(None)

    def test_base_channel(self):
        """ if BaseChannel handling is working """

        chan = BaseChannel(loop=self.loop)
        n = TestNode()
        msg = generate_msg()

        chan.add(n)

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

        self.assertTrue(n.processed, "Channel handle not working")

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

        chan = BaseChannel(loop=self.loop)
        msg = generate_msg()

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


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

        chan = BaseChannel(loop=self.loop)
        n1 = TestNode(name="main")
        n2 = TestNode(name="sub")

        msg = generate_msg()

        chan.add(n1)
        sub = chan.fork()
        sub.add(n2)

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

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

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

        chan = BaseChannel(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()
{'fraction_non_alphanumeric': 0.10184821077467558, 'fraction_numerical': 0.0031458906802988595, 'mean_word_length': 2.6232193732193734, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4201871', 'n_tokens_mistral': 1501, 'n_tokens_neox': 1432, 'n_words': 388}