Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- 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
Add Comment
Please, Sign In to add comment