Source code for grappa.test

# -*- coding: utf-8 -*-
from .log import log
from .empty import empty
from .base import BaseTest
from .engine import Engine
from .context import Context
from .resolver import OperatorResolver


[docs]class Test(BaseTest): """ Test represents the test definition in `grappa` with extensible and dynamic, runtime inferred DSL based on registered operators and third-party plugins. Arguments: subject (mixed): subject value to test. """ # Tracks context manager scopes _context = 0 # Tracks yielded value by context manager _context_subject = empty # Global flag, only used by global singleton instance _global = False def __init__(self, subject=empty): self._engine = Engine() self._ctx = Context() self._ctx.subjects = [] self._ctx.subject = subject self._ctx.chained = False self._ctx.style = 'should' @property def should(self): """ Alias name to self reference the current instance. Required for DSL API. """ return self @property def expect(self): """ Alias name to self reference the current instance. Required for DSL API. """ return self @property def _root(self): return test def __call__(self, subject, overload=False): """ Overloads function invokation of `Test` class instance. This is magical and widely used in `grappa` test execution by both developers and internal engine. Arguments: subject (mixed): test subject to use. overload (bool): `True` if the call if triggered via operator overloading invokation, otherise `False`. Returns: grappa.Test: new test instance with the given subject. """ self._ctx.subject = subject return self._trigger() if overload else Test(subject) def __getattr__(self, name): """ Overloads class attribute accessor proxying calls dynamically into assertion operators calls. This method is invoked by Python runtime engine, not by developers. """ # Return a new test instance if running as global if self._global: # If using context manager, use context defined subject subject = self._context_subject if self._context else empty # Create new test and proxy attribute call return Test(subject).__getattr__(name) # Resolve and register operator by name return OperatorResolver(self).resolve(name) def _trigger(self): """ Trigger assertions in the current test engine. Raises: AssertionError: in case of assertion error. Exception: in case of any other assertion error. """ log.debug('[test] trigger with context: {}'.format(self._ctx)) try: err = self._engine.run(self._ctx) except Exception as _err: err = _err finally: # Important: reset engine state to defaults self._engine.reset() self._root._engine.reset() # If error is present, raise it! if err: raise err return self def _clone(self): """ Clones the current `Test` instance. Returns: grappa.Test """ test = Test(self._ctx.subject) test._ctx = self._ctx.clone() test._engine = self._engine.clone() return test def _flush(self): """ Flushes the current test state, including test engine, assertions and current context. """ self.__init__() # Assertions composition
[docs] def all(self, *tests): """ Composes multiple tests and executes them, in series, once a subject is received. Conditional composition operator equivalent to `all` built-in Python function. Arguments: *tests (grappa.Test): test instances to run. """ def run_tests(subject): for test in tests: try: test(subject, overload=True) except Exception as err: return err return True self._engine.add_assertion(run_tests) return self
[docs] def any(self, *tests): """ Composes multiple tests and executes them, in series, once a subject is received. Conditional composition operator equivalent to `any` built-in Python function. Arguments: *tests (grappa.Test): test instances to run. """ def run_tests(subject): err = None for test in tests: try: test(subject, overload=True) except Exception as _err: err = _err else: return True return err self._engine.add_assertion(run_tests) return self
def __overload__(self, subject): """ Method triggered by magic methods executed via operator overloading. """ if isinstance(subject, Test): # Clone test instance to make it side-effects free fork = subject._clone() fork._ctx.chained = True fork._ctx.subject = self._ctx.subject # Trigger assertions return fork._trigger() # Otherwise invoke the test function with a subject return self.__call__(subject, overload=True) def __or__(self, value): """ Overloads ``|`` as from left-to-right operator precedence expression. """ return self.__overload__(value) def __ror__(self, value): """ Overloads ``|`` operator. """ return self.__overload__(value) def __gt__(self, value): """ Overloads ``>`` operator. """ return self.__overload__(value) def __enter__(self): """ Initializes context manager. """ log.debug('creates new test context manager: {}'.format(self._ctx)) test._context += 1 test._context_subject = self._ctx.subject def __exit__(self, etype, value, traceback): """ Exists context manager. """ log.debug('exists test context manager: {}'.format(value)) test._context -= 1 if test._context == 0: test._context_subject = empty
# Create global singleton instance test = Test() # This is black magic in order to deal with chainable states # and operator precedence. test._global = True