factory.py

import collections
import inspect

EMPTY = object()

# Goals:
# - Make it easier to write tests:
#   - Reduce the number of variables a test needs to manage by introducing hidden state
#   - Automate duplicated tasks:
#     - Creating topologies
#   - Easy to control the messages sent/received:
#     - Simulate concurrency by interleaving state changes for differents transfers
#     - Simulate networks problems by delaying/replaying/dropping state changes
#     - Simulate attacks by producing invalid state changes (with or without valid signatures)
#   - Easy to test multiple nodes:
#     - To test both ends of a channel and simulate a network
#   - Easy to assert on valid states:
#     - Needs to take into account difference of views of both nodes:
#       - assertions that are valid globally
#       - assertions that are valid locally
#       - assertions that are valid after a given state change


def get_defaults(argspec):
    """ Returns a dictionary with the default arguments of a function. """
    # This assumes the language forbids duplicate names for function arguments
    if argspec.kwonlydefaults is not None:
        args = dict(argspec.kwonlydefaults)
    else:
        args = dict()

    if argspec.defaults is not None:
        positional_with_defaults = argspec.args[-len(argspec.defaults):]
        args.update(zip(positional_with_defaults, argspec.defaults))

    return args


def make_chainmap(defaults):
    if defaults is not None and not isinstance(defaults, collections.ChainMap):
        variables = collections.ChainMap(defaults)
    elif defaults is None:
        variables = collections.ChainMap()
    else:
        variables = defaults

    return variables


class Builder:
    def __init__(self, defaults: dict = None, factories: dict = None):
        variables = make_chainmap(defaults)
        self.variables = variables
        self.factories = factories or dict()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        pass

    def __getitem__(self, key):
        return self.variables[key]

    def __setitem__(self, key, value):
        self.variables[key] = value

    def __delitem__(self, key):
        del self.variables[key]

    def overwrite(self, **overwrites):
        return Builder(self.variables.new_child(overwrites))

    def builder_for(self, func):
        """ Createa a builder which has all the variables to call `func`. """
        missing = self.missing_callargs(func)
        return self.overwrite(**missing)

    def create(self, what):
        """ Create an instance of `what`.

        If there is a factory associated with `what`, this factory function is
        called, and the arguments passed to it are the ones in this builder.

        Values currently in the builder always have priority. If there is no
        value for the given argument but `func` has a default argument, that is
        used instead of instantiating a new object. Otherwise a new object is
        created.

        Note:
            varargs is ignored, it cannot be called by keyword
        """
        factory = self.factories.get(what, what)

        if not callable(factory):
            raise ValueError(
                f'Cant create an instance of {repr(what)} because there is no '
                f'factory associated to it.',
            )

        try:
            callargs = self.callargs_for(factory)
            instance = factory(**callargs)
        except TypeError:
            # builtins raise type error on inspect, assume no arguments
            instance = factory()

        return instance

    def missing_callargs(self, func):
        """ Return the a dictionary with the values which are missing in this
        builder to call func, defaults are take into account.

        Raises:
            TypeError if func cannot be inspected.
        """
        argspec = inspect.getfullargspec(func)
        callargs = get_defaults(argspec)
        argument_names = argspec.args + argspec.kwonlyargs
        missing = dict()

        # Do not try to force all names to have a value, otherwise this won't
        # work with methods (self should not be assigned)
        for name in argument_names:
            value = self.variables.get(name, EMPTY)
            annotation = argspec.annotations.get(name)

            if value is not EMPTY:
                callargs[name] = value
            elif name not in callargs and annotation:
                instance = self.create(annotation)
                missing[name] = instance

        return missing

    def callargs_for(self, func):
        """ Return the arguments necessary to call func.

        Raises:
            TypeError if func cannot be inspected.
        """
        argspec = inspect.getfullargspec(func)
        callargs = get_defaults(argspec)
        argument_names = argspec.args + argspec.kwonlyargs

        # Do not try to force all names to have a value, otherwise this won't
        # work with methods (self should not be assigned)
        for name in argument_names:
            value = self.variables.get(name, EMPTY)
            annotation = argspec.annotations.get(name)

            if value is not EMPTY:
                callargs[name] = value
            elif name not in callargs and annotation:
                instance = self.create(annotation)
                callargs[name] = instance

        return callargs