<commit_before># Test suite for kaa.notifier classes

import kaa
from kaa import *

def test(result, expected):
    if result != expected:
        raise ValueError, "expected: %s - got: %s" % (str(expected), str(result))

def cb_func(*args, **kwargs):
    return args, kwargs

class Cls(object):
    def meth(self, *args, **kwargs):
        return args, kwargs

##############################################################
# Callbacks
#######################
print "Callbacks ..." 

cb = Callback(cb_func)
test( cb(42), ((42,), {}) )

cb = Callback(cb_func, 42)
test( cb(), ((42,), {}) )

cb = Callback(cb_func)
test( cb(42, "foo", bar="baz"), ((42,"foo"), {"bar":"baz"}) )


cb_meth = Cls().meth
cb = Callback(cb_meth)
test( cb(42), ((42,), {}) )

cb = Callback(cb_meth)
del cb_meth
test( cb(42), ((42,), {}) )



##############################################################
# Weak callbacks
#######################
print "Weak Callbacks ..." 

def cb_func2(*args, **kwargs):
    return args, kwargs

cb = WeakCallback(cb_func)
test( cb(42), ((42,), {}) )

# Lambdas are not weakref'd
cb = WeakCallback(lambda arg: arg)
test( cb(42), 42)

# Functions are weakref'd
cb = WeakCallback(cb_func2)
del cb_func2
test( cb(42), None )

cb_meth = Cls().meth
cb = WeakCallback(cb_meth)
del cb_meth
test( cb(42), None )


##############################################################
# Timers
#######################
print "Timers ..." 

def test_OneShotTimer(arg):
    result.append(arg)

def test_Timer(arg):
    result.append(arg)

class Cls(object):
    def meth(self, arg):
        result.append(arg)

result = []
OneShotTimer(test_OneShotTimer, 42).start(0)
kaa.main.step()
test(result, [42])

result = []
OneShotTimer(Cls().meth, 42).start(0)
kaa.main.step()
test(result, [42])

result = []
WeakOneShotTimer(Cls().meth, 42).start(0)
kaa.main.step()
test(result, [])

result = []
cb = Cls().meth
WeakOneShotTimer(cb, 42).start(0)
kaa.main.step()
test(result, [42])

result = []
timer = Timer(Cls().meth, 42)
timer.start(0)
for i in range(5):
    kaa.main.step()
timer.stop()
test(result, [42, 42, 42, 42, 42])

result = []
timer = WeakTimer(Cls().meth, 42)
timer.start(0)
for i in range(5):
    kaa.main.step()
test(result, [])
test(timer.active(), False)


##############################################################
# Signals 
#######################
print "Signals ..."<commit_msg>Adjust number of times we step in the test for weak timer destruction since we could be waiting a while.  (More details in comments.)<commit_after># Test suite for kaa.notifier classes

import kaa
from kaa import *

def test(result, expected):
    if result != expected:
        raise ValueError, "expected: %s - got: %s" % (str(expected), str(result))

def cb_func(*args, **kwargs):
    return args, kwargs

class Cls(object):
    def meth(self, *args, **kwargs):
        return args, kwargs

##############################################################
# Callbacks
#######################
print "Callbacks ..." 

cb = Callback(cb_func)
test( cb(42), ((42,), {}) )

cb = Callback(cb_func, 42)
test( cb(), ((42,), {}) )

cb = Callback(cb_func)
test( cb(42, "foo", bar="baz"), ((42,"foo"), {"bar":"baz"}) )


cb_meth = Cls().meth
cb = Callback(cb_meth)
test( cb(42), ((42,), {}) )

cb = Callback(cb_meth)
del cb_meth
test( cb(42), ((42,), {}) )



##############################################################
# Weak callbacks
#######################
print "Weak Callbacks ..." 

def cb_func2(*args, **kwargs):
    return args, kwargs

cb = WeakCallback(cb_func)
test( cb(42), ((42,), {}) )

# Lambdas are not weakref'd
cb = WeakCallback(lambda arg: arg)
test( cb(42), 42)

# Functions are weakref'd
cb = WeakCallback(cb_func2)
del cb_func2
test( cb(42), None )

cb_meth = Cls().meth
cb = WeakCallback(cb_meth)
del cb_meth
test( cb(42), None )


##############################################################
# Timers
#######################
print "Timers ..." 

def test_OneShotTimer(arg):
    result.append(arg)

def test_Timer(arg):
    result.append(arg)

class Cls(object):
    def meth(self, arg):
        result.append(arg)

result = []
OneShotTimer(test_OneShotTimer, 42).start(0)
kaa.main.step()
test(result, [42])

result = []
OneShotTimer(Cls().meth, 42).start(0)
kaa.main.step()
test(result, [42])

result = []
WeakOneShotTimer(Cls().meth, 42).start(0)
kaa.main.step()
test(result, [])

result = []
cb = Cls().meth
WeakOneShotTimer(cb, 42).start(0)
kaa.main.step()
test(result, [42])

result = []
timer = Timer(Cls().meth, 42)
timer.start(0)
for i in range(5):
    kaa.main.step()
timer.stop()
test(result, [42, 42, 42, 42, 42])

result = []
# Tests proper destruction of a weak timer.  We hold a weak ref to Cls().meth
# and since there are no other refs, the weak timer never gets activated.
timer = WeakTimer(Cls().meth, 42)
timer.start(0)
# Notifier technically has nothing to do, so it may sleep for 30 seconds.
print '(This could take 30 seconds or so.)'
for i in range(2):
    kaa.main.step()
test(result, [])
test(timer.active(), False)


##############################################################
# Signals 
#######################
print "Signals ..."
{'fraction_non_alphanumeric': 0.28881145954316684, 'fraction_numerical': 0.028261711188540456, 'mean_word_length': 4.88496583143508, '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': '6155093', 'n_tokens_mistral': 1941, 'n_tokens_neox': 1769, 'n_words': 461}