<commit_before>import unittest
import sys
import os
sys.path.insert(0, os.path.abspath('..'))
import observed
from observed import observable_function, observable_method


def get_caller_name(caller):
    """Find the name of a calling (i.e. observed) object.

    Args:
        caller: The observed object which is calling an observer.

    Returns:
        The name of the caller.
    """
    if hasattr(caller, "__self__"):
        # caller is a Foo instance
        name = caller.__self__.name
    else:
        # caller is a function.
        name = caller.__name__
    return name


def clear_list(l):
    """Remove all entries from a list.

    Args:
        l: The list to be cleared.
    """
    while True:
        try:
            l.pop(0)
        except IndexError:
            break


class Foo(object):
    """A generic class with observable methods.

    Attributes:
        name - string: A string naming the instance.
        buf - list: A buffer for collecting a record of called methods. My
            methods are set up to append a string to buf when they are called.
            The strings are formatted in one of two ways. If the method being
            called accepts the observed object as a first argument, the string
            is:
            "%s%s%s"%(self.name, method_name, caller_name)
            where method_name is the name of the method being called on me and
            caller_name is the name of the calling Foo instance or the name of
            the calling function. Note that the name of the calling instance is
            NOT the same thing as the name of the calling method, which doesn't
            exist. If the method being called does not accept the caller as a
            first argument, the string written to buf is:
            "%s%s"%(self.name, method_name).
    """

    def __init__(self, name, buf):
        """Initialize a Foo object.

        Args:
            name: A name for this instance.
            buf: A buffer (list) into which I write data when my methods are
                called. See the class docstring for details.
        """
        self.name = name
        self.buf = buf
    
    @observable_method
    def bar(self):
        self.buf.append("%sbar"%(self.name,))
    
    def baz(self):
        self.buf.append("%sbaz"%(self.name,))
    
    @observable_method
    def milton(self, caller):
        caller_name = get_caller_name(caller)
        self.buf.append("%smilton%s"%(self.name, caller_name))
    
    def waldo(self, caller):
        caller_name = get_caller_name(caller)
        self.buf.append("%swaldo%s"%(self.name, caller_name))

    def observable_methods(self):
        return [self.bar, self.milton]

    def method_info(self):
        return [(self.bar, False),
                (self.baz, False),
                (self.milton, True),
                (self.waldo, True)]


class Goo(Foo):
    """A class using the descriptor strategy for observable methods.

    I am entirely similar to Foo except that my observable methods use the
    descriptor persistence strategy. See the docstring for observable_method
    for a detailed explanation.
    """

    def bar(self):
        self.buf.append("%sbar"%(self.name,))
    bar = observable_method(bar, strategy='descriptor')

    def milton(self, caller):
        caller_name = get_caller_name(caller)
        self.buf.append("%smilton%s"%(self.name, caller_name))
    milton = observable_method(milton, strategy='descriptor')


# Not currently used

def get_observables(*objs):
    """Get a list observables from some objects.

    For each object, if it's a Foo or subclass, we get all of the object's
    observable methods. If it's a function
    """

    observables = []
    for obj in objs:
        if isinstance(obj, Foo):
            observables.extend(obj.observable_methods())
        elif isinstance(obj, observed.ObservableFunction):
            observables.append(obj)
        else:
            raise TypeError("Object of type %s not observable"%(type(obj),))
    return observables


def get_observers(*objs):
    observer_sets = []
    single_observers = []
    for obj in objs:
        if isinstance(obj, Foo):
            single_observers.extend(obj.method_info())
        else:
            single_observers.append((obj[0], obj[1]))
    for num_observers in range(len(single_observers)):
        for comb in itertools.combinations(single_observers, num_observers):
            observer_sets.append(comb)
    return observer_sets

# End not currently used


def make_observed_dict(*objects):
    """Construct a table of observable objects keyed by unique string names.

    In test scripts, it is convenient to able to iterate over combinations of
    observed and observing objects. However, we may wish to define these
    combinations before the objects are constructed. Therefore, we define the
    combinations as a set of string identifiers and then _after_ the objects
    are created we use this function to build a table of objects keyed by
    those identifiers.

    Args:
        objects: Collection of objects to be formed into the table.

    Returns:
        A table of observable objects keyed by unique string identifiers. A Foo
        instance with .name="my_foo" produces two entries in the table:
            "my_foo.bar": my_foo.bar
            "my_foo.milton": my_foo.milton
        and a function func produces one entry:<commit_msg>Added a function to generate observed/observer combinations.<commit_after>import unittest
import sys
import os
import itertools
sys.path.insert(0, os.path.abspath('..'))
import observed
from observed import observable_function, observable_method


def get_caller_name(caller):
    """Find the name of a calling (i.e. observed) object.

    Args:
        caller: The observed object which is calling an observer.

    Returns:
        The name of the caller.
    """
    if hasattr(caller, "__self__"):
        # caller is a Foo instance
        name = caller.__self__.name
    else:
        # caller is a function.
        name = caller.__name__
    return name


def clear_list(l):
    """Remove all entries from a list.

    Args:
        l: The list to be cleared.
    """
    while True:
        try:
            l.pop(0)
        except IndexError:
            break


class Foo(object):
    """A generic class with observable methods.

    Attributes:
        name - string: A string naming the instance.
        buf - list: A buffer for collecting a record of called methods. My
            methods are set up to append a string to buf when they are called.
            The strings are formatted in one of two ways. If the method being
            called accepts the observed object as a first argument, the string
            is:
            "%s%s%s"%(self.name, method_name, caller_name)
            where method_name is the name of the method being called on me and
            caller_name is the name of the calling Foo instance or the name of
            the calling function. Note that the name of the calling instance is
            NOT the same thing as the name of the calling method, which doesn't
            exist. If the method being called does not accept the caller as a
            first argument, the string written to buf is:
            "%s%s"%(self.name, method_name).
    """

    def __init__(self, name, buf):
        """Initialize a Foo object.

        Args:
            name: A name for this instance.
            buf: A buffer (list) into which I write data when my methods are
                called. See the class docstring for details.
        """
        self.name = name
        self.buf = buf
    
    @observable_method
    def bar(self):
        self.buf.append("%sbar"%(self.name,))
    
    def baz(self):
        self.buf.append("%sbaz"%(self.name,))
    
    @observable_method
    def milton(self, caller):
        caller_name = get_caller_name(caller)
        self.buf.append("%smilton%s"%(self.name, caller_name))
    
    def waldo(self, caller):
        caller_name = get_caller_name(caller)
        self.buf.append("%swaldo%s"%(self.name, caller_name))

    def observable_methods(self):
        return [self.bar, self.milton]

    def method_info(self):
        return [(self.bar, False),
                (self.baz, False),
                (self.milton, True),
                (self.waldo, True)]


class Goo(Foo):
    """A class using the descriptor strategy for observable methods.

    I am entirely similar to Foo except that my observable methods use the
    descriptor persistence strategy. See the docstring for observable_method
    for a detailed explanation.
    """

    def bar(self):
        self.buf.append("%sbar"%(self.name,))
    bar = observable_method(bar, strategy='descriptor')

    def milton(self, caller):
        caller_name = get_caller_name(caller)
        self.buf.append("%smilton%s"%(self.name, caller_name))
    milton = observable_method(milton, strategy='descriptor')


# Not currently used

def get_observables(*objs):
    """Get a list observables from some objects.

    For each object, if it's a Foo or subclass, we get all of the object's
    observable methods. If it's a function
    """

    observables = []
    for obj in objs:
        if isinstance(obj, Foo):
            observables.extend(obj.observable_methods())
        elif isinstance(obj, observed.ObservableFunction):
            observables.append(obj)
        else:
            raise TypeError("Object of type %s not observable"%(type(obj),))
    return observables


def get_observer_sets(*objs):
    observer_sets = []
    single_observers = []
    for obj in objs:
        if isinstance(obj, Foo):
            single_observers.extend(obj.method_info())
        else:
            single_observers.append((obj[0], obj[1]))
    # for num_observers in range(len(single_observers)):
    for comb in itertools.combinations(single_observers, 3):
        observer_sets.append(comb)
    return observer_sets


def get_items(observables, observer_sets):

    def get_buff_data(observable, observer, identify_observed):
        """Get the buffer data an object will write."""
        if hasattr(observer, '__self__'):
            expected = observer.__self__.name+observer.__name__
        else:
            expected = observer.__name__
        if identify_observed:
            expected = expected + get_caller_name(observable)
        return expected

    items = []
    for observable, observer_set in itertools.product(observables, observer_sets):
        # Don't include this combination if it would cause infinite recursion.
        recursion = False
        for observer, _ in observer_set:
            if type(observer) == type(observable):
                if observer == observable:
                    recursion = True
        if recursion:
            continue
        buff_data = []
        if isinstance(observable, observed.ObservableBoundMethod):
            final = observable.__self__.name + observable.__name__
        elif isinstance(observable, observed.ObservableFunction):
            final = observale.__name__
        for observer, caller_id in observer_set:
            buff_data.append(get_buff_data(observable, observer, caller_id))
        buff_data.insert(0, final)
        buff_data.sort()
        items.append((observable, observer_set, buff_data, final))
    return items

# End not currently used


def make_observed_dict(*objects):
    """Construct a table of observable objects keyed by unique string names.

    In test scripts, it is convenient to able to iterate over combinations of
    observed and observing objects. However, we may wish to define these
    combinations before the objects are constructed. Therefore, we define the
    combinations as a set of string identifiers and then _after_ the objects
    are created we use this function to build a table of objects keyed by
    those identifiers.

    Args:
        objects: Collection of objects to be formed into the table.

    Returns:
        A table of observable objects keyed by unique string identifiers. A Foo
        instance with .name="my_foo" produces two entries in the table:
            "my_foo.bar": my_foo.bar
            "my_foo.milton": my_foo.milton
        and a function func produces one entry:
{'fraction_non_alphanumeric': 0.08045883501464367, 'fraction_numerical': 0.0008135372600065083, 'mean_word_length': 2.4328399888299357, 'pattern_counts': {'":': 4, '<': 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': '463075', 'n_tokens_mistral': 3394, 'n_tokens_neox': 3277, 'n_words': 1224}