KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use iscoroutinefunction method in Behave

Best Python code snippet using behave

coroutines.py

Source:coroutines.py

...160    """Decorator to mark coroutines.161    If the coroutine is not yielded from before it is destroyed,162    an error message is logged.163    """164    if _inspect_iscoroutinefunction(func):165        # In Python 3.5 that's all we need to do for coroutines166        # defined with "async def".167        # Wrapping in CoroWrapper will happen via168        # 'sys.set_coroutine_wrapper' function.169        return func170    if inspect.isgeneratorfunction(func):171        coro = func172    else:173        @functools.wraps(func)174        def coro(*args, **kw):175            res = func(*args, **kw)176            if (base_futures.isfuture(res) or inspect.isgenerator(res) or177                isinstance(res, CoroWrapper)):178                res = yield from res179            elif _AwaitableABC is not None:180                # If 'func' returns an Awaitable (new in 3.5) we181                # want to run it.182                try:183                    await_meth = res.__await__184                except AttributeError:185                    pass186                else:187                    if isinstance(res, _AwaitableABC):188                        res = yield from await_meth()189            return res190    if not _DEBUG:191        if _types_coroutine is None:192            wrapper = coro193        else:194            wrapper = _types_coroutine(coro)195    else:196        @functools.wraps(func)197        def wrapper(*args, **kwds):198            w = CoroWrapper(coro(*args, **kwds), func=func)199            if w._source_traceback:200                del w._source_traceback[-1]201            # Python < 3.5 does not implement __qualname__202            # on generator objects, so we set it manually.203            # We use getattr as some callables (such as204            # functools.partial may lack __qualname__).205            w.__name__ = getattr(func, '__name__', None)206            w.__qualname__ = getattr(func, '__qualname__', None)207            return w208    wrapper._is_coroutine = _is_coroutine  # For iscoroutinefunction().209    return wrapper210# A marker for iscoroutinefunction.211_is_coroutine = object()212def iscoroutinefunction(func):213    """Return True if func is a decorated coroutine function."""214    return (getattr(func, '_is_coroutine', None) is _is_coroutine or215            _inspect_iscoroutinefunction(func))216_COROUTINE_TYPES = (types.GeneratorType, CoroWrapper)217if _CoroutineABC is not None:218    _COROUTINE_TYPES += (_CoroutineABC,)219if _types_CoroutineType is not None:220    # Prioritize native coroutine check to speed-up221    # asyncio.iscoroutine.222    _COROUTINE_TYPES = (_types_CoroutineType,) + _COROUTINE_TYPES223def iscoroutine(obj):224    """Return True if obj is a coroutine object."""225    return isinstance(obj, _COROUTINE_TYPES)226def _format_coroutine(coro):227    assert iscoroutine(coro)228    if not hasattr(coro, 'cr_code') and not hasattr(coro, 'gi_code'):229        # Most likely a built-in type or a Cython coroutine....

LOADER.py

Source:LOADER.py

...36            self._help[name] = f()37        return deco38    def command(self, name, **options):39        def deco(f):40            if not asyncio.iscoroutinefunction(f):41                print('{0.__name__} command must be a coroutine'.format(f))42            else:43                #setattr(self, coro.__name__, coro)44                self._role[name] = ROLES.EVERYONE45                if "role" in options:46                    self._role[name] = options["role"]47                if "no_error" in options:48                    self._noerr[name] = options["no_error"]49                self.Command[name] = f50                print('\tLOAD_OK: {0.__name__}: on_command @ {1}: {2}'.format(f, name, self._role[name]))51        return deco52    def message(self, name, **options):53        def deco(f):54            if not asyncio.iscoroutinefunction(f):55                print('{0.__name__} command must be a coroutine'.format(f))56            else:57                #setattr(self, coro.__name__, coro)58                self._role[name] = ROLES.EVERYONE59                if "role" in options:60                    self._role[name] = options["role"]61                self.Message[name] = f62                print('\tLOAD_OK: {0.__name__}: on_message @ {1}: {2}'.format(f, name, self._role[name]))63        return deco64    def mention(self, name, **options):65        def deco(f):66            if not asyncio.iscoroutinefunction(f):67                print('{0.__name__} command must be a coroutine'.format(f))68            else:69                #setattr(self, coro.__name__, coro)70                self._role[name] = ROLES.EVERYONE71                if "role" in options:72                    self._role[name] = options["role"]73                self.Mention[name] = f74                print('\tLOAD_OK: {0.__name__}: on_mention @ {1} with role {2}'.format(f, name, self._role[name]))75        return deco76    def typing(self, name, **options):77        def deco(f):78            if not asyncio.iscoroutinefunction(f):79                print('{0.__name__} command must be a coroutine'.format(f))80            else:81                self.Typing[name] = f82                print('\tLOAD_OK: {0.__name__}: on_typing @ {1}'.format(f, name))83        return deco84    def delete(self, name, **options):85        def deco(f):86            if not asyncio.iscoroutinefunction(f):87                print('{0.__name__} command must be a coroutine'.format(f))88            else:89                self.Delete[name] = f90                print('\tLOAD_OK: {0.__name__}: on_delete @ {1}.'.format(f, name))91        return deco92    def edit(self, name, **options):93        def deco(f):94            if not asyncio.iscoroutinefunction(f):95                print('{0.__name__} command must be a coroutine'.format(f))96            else:97                98                self.Edit[name] = f99                print('\tLOAD_OK: {0.__name__}: on_edit @ {1}'.format(f, name))100        return deco101    def ready(self, name, **options):102        def deco(f):103            if not asyncio.iscoroutinefunction(f):104                print('{0.__name__} command must be a coroutine'.format(f))105            else:106                107                self.Ready[name] = f108                print('\tLOAD_OK: {0.__name__}: on_ready @ {1}'.format(f, name))109        return deco110    def reactadd(self, name, **options):111        def deco(f):112            if not asyncio.iscoroutinefunction(f):113                print('{0.__name__} command must be a coroutine'.format(f))114            else:115                116                self.ReactAdd[name] = f117                print('\tLOAD_OK: {0.__name__}: react_add @ {1}'.format(f, name))118        return deco119    def reactremove(self, name, **options):120        def deco(f):121            if not asyncio.iscoroutinefunction(f):122                print('{0.__name__} command must be a coroutine'.format(f))123            else:124                125                self.ReactRemove[name] = f126                print('\tLOAD_OK: {0.__name__}: react_remove @ {1}'.format(f, name))127        return deco128    def reactclear(self, name, **options):129        def deco(f):130            if not asyncio.iscoroutinefunction(f):131                print('{0.__name__} command must be a coroutine'.format(f))132            else:133                134                self.ReactClear[name] = f135                print('\tLOAD_OK: {0.__name__}: react_clear @ {1}'.format(f, name))136        return deco137    def channeladd(self, name, **options):138        def deco(f):139            if not asyncio.iscoroutinefunction(f):140                print('{0.__name__} command must be a coroutine'.format(f))141            else:142                143                self.ChannelAdd[name] = f144                print('\tLOAD_OK: {0.__name__}: channel_add @ {1}'.format(f, name))145        return deco146    def channelremove(self, name, **options):147        def deco(f):148            if not asyncio.iscoroutinefunction(f):149                print('{0.__name__} command must be a coroutine'.format(f))150            else:151                152                self.ChannelRemove[name] = f153                print('\tLOAD_OK: {0.__name__}: channel_remove @ {1}'.format(f, name))154        return deco155    def memberjoin(self, name, **options):156        def deco(f):157            if not asyncio.iscoroutinefunction(f):158                print('{0.__name__} command must be a coroutine'.format(f))159            else:160                161                self.MemberJoin[name] = f162                print('\tLOAD_OK: {0.__name__}: member_join @ {1}'.format(f, name))163        return deco164    def memberleave(self, name, **options):165        def deco(f):166            if not asyncio.iscoroutinefunction(f):167                print('{0.__name__} command must be a coroutine'.format(f))168            else:169                self.MemberLeave[name] = f170                print('\tLOAD_OK: {0.__name__}: member_leave @ {1}'.format(f, name))171        return deco172    def memberupdate(self, name, **options):173        def deco(f):174            if not asyncio.iscoroutinefunction(f):175                print('{0.__name__} command must be a coroutine'.format(f))176            else:177                self.MemberUpdate[name] = f178                print('\tLOAD_OK: {0.__name__}: member_update @ {1}'.format(f, name))179        return deco180    def ban(self, name, **options):181        def deco(f):182            if not asyncio.iscoroutinefunction(f):183                print('{0.__name__} command must be a coroutine'.format(f))184            else:185                self.MemberBan[name] = f186                print('\tLOAD_OK: {0.__name__}: member_ban @ {1}'.format(f, name))187        return deco188    def unban(self, name, **options):189        def deco(f):190            if not asyncio.iscoroutinefunction(f):191                print('{0.__name__} command must be a coroutine'.format(f))192            else:193                self.MemberUnban[name] = f194                print('\tLOAD_OK: {0.__name__}: member_unban @ {1}'.format(f, name))195        return deco196    def botjoin(self, name, **options):197        def deco(f):198            if not asyncio.iscoroutinefunction(f):199                print('{0.__name__} command must be a coroutine'.format(f))200            else:201                self.BotJoin[name] = f202                print('\tLOAD_OK: {0.__name__}: bot_join @ {1}'.format(f, name))203        return deco204    def serverenable(self, name, **options):205        def deco(f):206            if not asyncio.iscoroutinefunction(f):207                print('{0.__name__} command must be a coroutine'.format(f))208            else:209                self.ServerEnable[name] = f210                print('\tLOAD_OK: {0.__name__}: server_enable @ {1}'.format(f, name))211        return deco212    def serverdisable(self, name, **options):213        def deco(f):214            if not asyncio.iscoroutinefunction(f):215                print('{0.__name__} command must be a coroutine'.format(f))216            else:217                self.ServerDisable[name] = f218                print('\tLOAD_OK: {0.__name__}: server_disable @ {1}'.format(f, name))219        return deco220    def serverchange(self, name, **options):221        def deco(f):222            if not asyncio.iscoroutinefunction(f):223                print('{0.__name__} command must be a coroutine'.format(f))224            else:225                self.ServerChange[name] = f226                print('\tLOAD_OK: {0.__name__}: server_change @ {1}'.format(f, name))227        return deco228    def roleadd(self, name, **options):229        def deco(f):230            if not asyncio.iscoroutinefunction(f):231                print('{0.__name__} command must be a coroutine'.format(f))232            else:233                self.RoleAdd[name] = f234                print('\tLOAD_OK: {0.__name__}: roleadd @ {1}'.format(f, name))235        return deco236    def roleremove(self, name, **options):237        def deco(f):238            if not asyncio.iscoroutinefunction(f):239                print('{0.__name__} command must be a coroutine'.format(f))240            else:241                self.RoleRemove[name] = f242                print('\tLOAD_OK: {0.__name__}: roleremove @ {1}'.format(f, name))243        return deco244    def emojichange(self, name, **options):245        def deco(f):246            if not asyncio.iscoroutinefunction(f):247                print('{0.__name__} command must be a coroutine'.format(f))248            else:249                self.EmojiUpdate[name] = f250                print('\tLOAD_OK: {0.__name__}: emojichange @ {1}'.format(f, name))251        return deco252    def voicechange(self, name, **options):253        def deco(f):254            if not asyncio.iscoroutinefunction(f):255                print('{0.__name__} command must be a coroutine'.format(f))256            else:257                self.VoiceUpdate[name] = f258                print('\tLOAD_OK: {0.__name__}: voicechange @ {1}'.format(f, name))...

base_decorator.py

Source:base_decorator.py

...17        self.func = func18    @abstractmethod19    def __call__(self, func, instance, *call_args, **call_kwargs):20        func = self.func if self.func else call_args[0]21        # iscoroutinefunction = inspect.iscoroutinefunction(func)22        @functools.wraps(func)23        def wrapped_function(*args, **kwargs):24            ## execute the func25            if self.func:26                res = self.call_func(func=func, *call_args, **call_kwargs)27            else:28                res = self.call_func(func=func, *args, **kwargs)29            return res30        return wrapped_function() if self.func else wrapped_function31    def __get__(self, instance, owner):32        """33        So, what happens is that in python 3,34        for method declarations work as methods,35        when they are just defined as functions inside the class body,36        what happens is that the language makes use of the "descriptor protocol".37        And to put it simply, an ordinary method is just a function,38        until it is retrieved from an instance:39        since the function has a __get__ method,40        they are recognized as descriptors,41        and the __get__ method is the one responsible to return a "partial function"42        which is the "bound method", and will insert the self parameter upon being called.43        Without a __get__ method, the instance of SomeWrapper when retrieved from an instance,44        has no information on the instance.45        :param instance: the instance of the class where the decorated method is.46        :param owner: owner argument is the owner class47        :return:48        """49        if self.func and instance:50            # if isinstance(self.func, functools.partial) and self.func.args[0] is instance:51            if isinstance(self.func, functools.partial): ## todo: how to indentify a wrapper with another?52                return functools.partial(53                    self.__call__,54                    self.func,55                    instance56                )57            else:58                partial_func = functools.partial(self.func, instance)59                partial_func.__name__ = self.func.__name__60                self.func = partial_func61                return functools.partial(62                    self.__call__,63                    partial_func,64                    instance65                )66        else:67            return self68    def call_func(self, func, *args, **kwargs):69        iscoroutinefunction = self.is_coroutine_funciton(obj=func)70        if iscoroutinefunction:71            res = self.call_async(func, *args, **kwargs)72        else:73            res = self.call_sync(func, *args, **kwargs)74        return res75    @staticmethod76    def call_async(func, *args, **kwargs):77        loop = asyncio.get_event_loop()78        if loop.is_running():79            res = asyncio.ensure_future(func(*args, **kwargs), loop=loop)80        else:81            res = loop.run_until_complete(func(*args, **kwargs))82        return res83    @staticmethod84    def call_sync(func, *args, **kwargs):85        res = func(*args, **kwargs)86        return res87    @staticmethod88    def is_coroutine_funciton(obj):89        if isinstance(obj, functools.partial):90            while isinstance(obj, functools.partial):91                obj = obj.func92            return inspect.iscoroutinefunction(obj)93        else:...

decorators.py

Source:decorators.py

...16            except JSONDecodeError as exc:17                raise HTTPException(400, repr(exc))18            except ValidationError as exc:19                raise BodyValidationException(exc)20            if iscoroutinefunction(func):21                return await func(request=request, **{name: data}, **kwargs)22            else:23                return func(request=request, **{name: data}, **kwargs)24        return wrapper25    return decorator26def db():27    def decorator(func):28        session = Session()29        @wraps(func)30        async def wrapper(request, **kwargs):31            if iscoroutinefunction(func):32                response = await func(request=request, db_session=session, **kwargs)33            else:34                response = func(request=request, db_session=session, **kwargs)35            return response36        return wrapper37    return decorator38def with_user(detail=False, teacher=False, student=False):39    def decorator(func):40        @wraps(func)41        async def wrapper(request, **kwargs):42            user = request.user43            if (44                user.is_authenticated45                and (teacher is False or user.is_teacher == teacher)46                and (student is False or user.is_student == student)47            ):48                if detail:49                    user = (50                        Session.query(User).filter(User.username == user.username).one()51                    )52                if iscoroutinefunction(func):53                    return await func(request=request, user=user, **kwargs)54                else:55                    return func(request=request, user=user, **kwargs)56            else:57                raise UnauthorizedException58        return wrapper59    return decorator60def query(*args):61    def decorator(func):62        @wraps(func)63        async def wrapper(request, **kwargs):64            query_params = {}65            for key in args:66                matched = re.match(r"(int):(.+)", key)67                if matched:68                    key = matched[2]69                if key in request.query_params:70                    if matched and request.query_params[key].isdecimal():71                        query_params[key] = int(request.query_params[key])72                    elif not matched:73                        query_params[key] = request.query_params[key]74            if iscoroutinefunction(func):75                return await func(request=request, **query_params, **kwargs)76            else:77                return func(request=request, **query_params, **kwargs)78        return wrapper79    return decorator80def path_param():81    def decorator(func):82        @wraps(func)83        async def wrapper(request, **kwargs):84            if iscoroutinefunction(func):85                return await func(request=request, **request.path_params, **kwargs)86            else:87                return func(request=request, **request.path_params, **kwargs)88        return wrapper...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.