<commit_before># Module:   core
# Date:     23rd June 2006
# Author:   <NAME>, prologic at shortcircuit dot net dot au

"""Core Test Suite"""

import unittest
from os import kill
from signal import SIGHUP, SIGINT, SIGTERM

from circuits.core import HAS_MULTIPROCESSING
from circuits.core import handler, Manager, Component, Event

if HAS_MULTIPROCESSING == 2:
    from multiprocessing import Value
else:
    from processing import Value


class Test(Event):
    """Test(Event) -> Test Event"""

class FilterComponent(Component):

    @handler("foo", filter=True)
    def onFOO(self, msg=""):
        return True

    @handler("bar", filter=True)
    def onBAR(self, msg=""):
        if msg.lower() == "hello world":
            return True

class ListenerComponent(Component):

    def foo(self, test, msg=""):
        if msg.lower() == "start":
            self.push(Test(msg="foo"), "foo")

    def bar(self, event, test, msg=""):
        if msg.lower() == "test":
            self.push(Test(msg="hello world"), event._channel)

class Foo(Component):

    flag = False
    gotbar = False

    @handler("test")
    def onTEST(self, event, *args, **kwargs):
        self.flag = True

    @handler("foo")
    def onFOO(self, event):
        self.send(Test(), "bar")

    @handler("gotbar")
    def onGOTBAR(self, event, *args):
        self.gotbar = True


class Bar(Component):

    @handler("bar")
    def onBAR(self, event, *args, **kwargs):
        self.send(Test(), "gotbar")

if HAS_MULTIPROCESSING:

    class TestSignals(unittest.TestCase):
        """Test Signal Handling

        Test that the OS signals SIGHUP, SIGINT and SIGTERM are handled
        and Signal events are fired for each.
        """

        def runTest(self):

            class SignalTester(Component):

                def __init__(self):
                    super(SignalTester, self).__init__()

                    self.signal = Value("I", 0)

                @handler("signa", filter=True)
                def signal(self, signal, stack):
                    self.signal = signal
                    return True

            tester = SignalTester()
            tester.start(process=True)

            pid = tester._task.pid

            kill(pid, SIGHUP)
            self.assertEquals(tester.signal.value, SIGHUP)

            #kill(pid, SIGINT)
            #self.assertEquals(tester.signal.value, SIGINT)

            #kill(pid, SIGTERM)
            #self.assertEquals(tester.signal.value, SIGTERM)

            tester.stop()

class TestAllChannels(unittest.TestCase):
    """Test All Channels

    Test that Events can be sent to all channels.
    """

    def runTest(self):

        class A(Component):

            channel = "A"

            flag = False

            @handler("foo")
            def onFOO(self, event, *args, **kwargs):
                self.flag = True

        class B(Component):

            channel = "B"

            flag = False<commit_msg>core.tests.test_core: Removed SignalTest - turns out there is no good way to test this :/<commit_after># Module:   core
# Date:     23rd June 2006
# Author:   <NAME>, prologic at shortcircuit dot net dot au

"""Core Test Suite"""

import unittest

from circuits.core import HAS_MULTIPROCESSING
from circuits.core import handler, Manager, Component, Event

class Test(Event):
    """Test(Event) -> Test Event"""

class FilterComponent(Component):

    @handler("foo", filter=True)
    def onFOO(self, msg=""):
        return True

    @handler("bar", filter=True)
    def onBAR(self, msg=""):
        if msg.lower() == "hello world":
            return True

class ListenerComponent(Component):

    def foo(self, test, msg=""):
        if msg.lower() == "start":
            self.push(Test(msg="foo"), "foo")

    def bar(self, event, test, msg=""):
        if msg.lower() == "test":
            self.push(Test(msg="hello world"), event._channel)

class Foo(Component):

    flag = False
    gotbar = False

    @handler("test")
    def onTEST(self, event, *args, **kwargs):
        self.flag = True

    @handler("foo")
    def onFOO(self, event):
        self.send(Test(), "bar")

    @handler("gotbar")
    def onGOTBAR(self, event, *args):
        self.gotbar = True


class Bar(Component):

    @handler("bar")
    def onBAR(self, event, *args, **kwargs):
        self.send(Test(), "gotbar")

class TestAllChannels(unittest.TestCase):
    """Test All Channels

    Test that Events can be sent to all channels.
    """

    def runTest(self):

        class A(Component):

            channel = "A"

            flag = False

            @handler("foo")
            def onFOO(self, event, *args, **kwargs):
                self.flag = True

        class B(Component):

            channel = "B"

            flag = False
{'fraction_non_alphanumeric': 0.12745918794474675, 'fraction_numerical': 0.0029300962745918793, 'mean_word_length': 2.24881033310673, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 7, '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': '6517622', 'n_tokens_mistral': 1484, 'n_tokens_neox': 1384, 'n_words': 402}