<commit_before>import warnings
import sys
import traceback
from contextlib import contextmanager
from functools import wraps

from twisted.internet.task import Clock
from twisted.internet.defer import Deferred, succeed

from spinoff.util.async import CancelledError
from spinoff.actor import BaseActor
from spinoff.util.pattern_matching import match, ANY, IS_INSTANCE, NOT
from spinoff.util.python import combomethod
from spinoff.actor.comm import ActorRef


__all__ = ['deferred', 'assert_raises', 'assert_not_raises', 'MockFunction', 'assert_num_warnings', 'assert_no_warnings', 'assert_one_warning']


def deferred(f):
    """This is almost exactly the same as nose.twistedtools.deferred, except it allows one to have simulated time in
    their test code by passing in a `Clock` instance. That `Clock` instance will be `advance`d as long as there are
    any deferred calls bound to it.

    """
    @wraps(f)
    def ret():
        error = [None]

        clock = Clock()

        d = f(clock)
        @d.addErrback
        def on_error(f):
            error[0] = sys.exc_info()

        while True:
            time_to_wait = max([0] + [call.getTime() - clock.seconds() for call in clock.getDelayedCalls()])
            if time_to_wait == 0:
                break
            else:
                clock.advance(time_to_wait)

        if error[0]:
            exc_info = error[0]
            raise exc_info[0], exc_info[1], exc_info[2]

    return ret


def immediate(d):
    assert d.called
    return d


def deferred_result(d):
    ret = [None]
    exc = [None]
    if isinstance(d, Deferred):
        d = immediate(d)
        d.addCallback(lambda result: ret.__setitem__(0, result))
        d.addErrback(lambda f: exc.__setitem__(0, f))
        if exc[0]:
            exc[0].raiseException()
        return ret[0]
    else:
        return d


def cancel_deferred(d):
    d.addErrback(lambda f: f.trap(CancelledError))
    d.cancel()


@contextmanager
def assert_not_raises(exc_class=Exception, message=None):
    if isinstance(exc_class, basestring):
        message, exc_class = exc_class, Exception
    try:
        yield
    except exc_class as e:
        raise AssertionError(message or "No exception should have been raised but instead %s was raised" % repr(e))


@contextmanager
def assert_raises(exc_class=Exception, message=None):
    if isinstance(exc_class, basestring):
        message, exc_class = exc_class, Exception
    try:
        yield
    except exc_class:
        pass
    else:
        raise AssertionError(message or "An exception should have been raised")


@contextmanager
def assert_num_warnings(n):
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        yield
        assert len(w) == n, "expected %s warnings but found %s: %s" % (n, len(w), ', '.join(map(str, w)))


def assert_no_warnings():
    return assert_num_warnings(0)


def assert_one_warning():
    return assert_num_warnings(1)


class MockFunction(object):

    called = property(lambda self: len(self.argses))

    def __init__(self, return_values=None):
        self.reset()<commit_msg>Added the optional 'message' parameter to assert_*_warnings in util.testing<commit_after>import warnings
import sys
import traceback
from contextlib import contextmanager
from functools import wraps

from twisted.internet.task import Clock
from twisted.internet.defer import Deferred, succeed

from spinoff.util.async import CancelledError
from spinoff.actor import BaseActor
from spinoff.util.pattern_matching import match, ANY, IS_INSTANCE, NOT
from spinoff.util.python import combomethod
from spinoff.actor.comm import ActorRef


__all__ = ['deferred', 'assert_raises', 'assert_not_raises', 'MockFunction', 'assert_num_warnings', 'assert_no_warnings', 'assert_one_warning']


def deferred(f):
    """This is almost exactly the same as nose.twistedtools.deferred, except it allows one to have simulated time in
    their test code by passing in a `Clock` instance. That `Clock` instance will be `advance`d as long as there are
    any deferred calls bound to it.

    """
    @wraps(f)
    def ret():
        error = [None]

        clock = Clock()

        d = f(clock)
        @d.addErrback
        def on_error(f):
            error[0] = sys.exc_info()

        while True:
            time_to_wait = max([0] + [call.getTime() - clock.seconds() for call in clock.getDelayedCalls()])
            if time_to_wait == 0:
                break
            else:
                clock.advance(time_to_wait)

        if error[0]:
            exc_info = error[0]
            raise exc_info[0], exc_info[1], exc_info[2]

    return ret


def immediate(d):
    assert d.called
    return d


def deferred_result(d):
    ret = [None]
    exc = [None]
    if isinstance(d, Deferred):
        d = immediate(d)
        d.addCallback(lambda result: ret.__setitem__(0, result))
        d.addErrback(lambda f: exc.__setitem__(0, f))
        if exc[0]:
            exc[0].raiseException()
        return ret[0]
    else:
        return d


def cancel_deferred(d):
    d.addErrback(lambda f: f.trap(CancelledError))
    d.cancel()


@contextmanager
def assert_not_raises(exc_class=Exception, message=None):
    if isinstance(exc_class, basestring):
        message, exc_class = exc_class, Exception
    try:
        yield
    except exc_class as e:
        raise AssertionError(message or "No exception should have been raised but instead %s was raised" % repr(e))


@contextmanager
def assert_raises(exc_class=Exception, message=None):
    if isinstance(exc_class, basestring):
        message, exc_class = exc_class, Exception
    try:
        yield
    except exc_class:
        pass
    else:
        raise AssertionError(message or "An exception should have been raised")


@contextmanager
def assert_num_warnings(n, message=None):
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        yield
        assert len(w) == n, message or "expected %s warnings but found %s: %s" % (n, len(w), ', '.join(map(str, w)))


def assert_no_warnings(message=None):
    return assert_num_warnings(0, message)


def assert_one_warning(message=None):
    return assert_num_warnings(1, message)


class MockFunction(object):

    called = property(lambda self: len(self.argses))

    def __init__(self, return_values=None):
        self.reset()
{'fraction_non_alphanumeric': 0.09635538870639762, 'fraction_numerical': 0.004692632566870014, 'mean_word_length': 3.0442757748260596, 'pattern_counts': {'":': 0, '<': 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': '626508', 'n_tokens_mistral': 1991, 'n_tokens_neox': 1942, 'n_words': 566}