Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use has_internal_errors method in Slash

Best Python code snippet using slash

admin.py

Source:admin.py

1import asyncio2import inspect3import logging4import traceback5from typing import Mapping6from aiohttp import web7import graphene8from graphql.execution.executors.asyncio import AsyncioExecutor9from graphql.error.located_error import GraphQLLocatedError10import simplejson as json11from .exceptions import InvalidAPIParameters, BackendError12from .auth import auth_required13from ..manager.models.base import DataLoaderManager14from ..manager.models import (15    Agent,16    KeyPair, CreateKeyPair, ModifyKeyPair, DeleteKeyPair,17    ComputeSession, ComputeWorker, KernelStatus,18    VirtualFolder,19)20log = logging.getLogger('ai.backend.gateway.admin')21@auth_required22async def handle_gql(request: web.Request) -> web.Response:23    executor = request.app['admin.gql_executor']24    if request['is_admin']:25        schema = request.app['admin.gql_schema_admin']26    else:27        schema = request.app['admin.gql_schema_user']28    try:29        body = await request.json(loads=json.loads)30    except (asyncio.TimeoutError, json.decoder.JSONDecodeError):31        raise InvalidAPIParameters('Malformed request body.')32    try:33        assert 'query' in body, \34               'The request must have "query" JSON field.'35        assert isinstance(body['query'], str), \36               'The "query" field must be a JSON string.'37        if 'variables' in body:38            assert (body['variables'] is None or39                    isinstance(body['variables'], Mapping)), \40                   'The "variables" field must be an JSON object or null.'41        else:42            body['variables'] = None43    except AssertionError as e:44        raise InvalidAPIParameters(e.args[0])45    dlmanager = DataLoaderManager(request.app['dbpool'])46    result = schema.execute(47        body['query'], executor,48        variable_values=body['variables'],49        context_value={50            'dlmgr': dlmanager,51            'access_key': request['keypair']['access_key'],52            'dbpool': request.app['dbpool'],53            'redis_stat_pool': request.app['redis_stat_pool'],54        },55        return_promise=True)56    if inspect.isawaitable(result):57        result = await result58    if result.errors:59        has_internal_errors = False60        for e in result.errors:61            if isinstance(e, GraphQLLocatedError):62                exc_info = (type(e.original_error),63                            e.original_error,64                            e.original_error.__traceback__)65                tb_text = ''.join(traceback.format_exception(*exc_info))66                log.error(f'GraphQL located error:\n{tb_text}')67                request.app['sentry'].captureException(exc_info)68                has_internal_errors = True69        if has_internal_errors:70            raise BackendError(str(result.errors[0]))71        raise InvalidAPIParameters(str(result.errors[0]))72    else:73        return web.json_response(result.data, status=200, dumps=json.dumps)74class MutationForAdmin(graphene.ObjectType):75    create_keypair = CreateKeyPair.Field()76    modify_keypair = ModifyKeyPair.Field()77    delete_keypair = DeleteKeyPair.Field()78# Nothing yet!79# class MutationForUser(graphene.ObjectType):80#     pass81class QueryForAdmin(graphene.ObjectType):82    '''83    Available GraphQL queries for the admin privilege.84    It allows use of any access keys regardless of the one specified in the85    authorization header as well as querying the keypair information of all86    users.87    '''88    agent = graphene.Field(89        Agent,90        agent_id=graphene.String())91    agents = graphene.List(92        Agent,93        status=graphene.String())94    keypair = graphene.Field(95        KeyPair,96        access_key=graphene.String())97    keypairs = graphene.List(98        KeyPair,99        user_id=graphene.Int(required=True),100        is_active=graphene.Boolean())101    vfolders = graphene.List(102        VirtualFolder,103        access_key=graphene.String())104    compute_sessions = graphene.List(105        ComputeSession,106        access_key=graphene.String(),107        status=graphene.String())108    compute_workers = graphene.List(109        ComputeWorker,110        sess_id=graphene.String(required=True),111        status=graphene.String())112    @staticmethod113    async def resolve_agent(executor, info, agent_id):114        manager = info.context['dlmgr']115        loader = manager.get_loader('Agent', status=None)116        return await loader.load(agent_id)117    @staticmethod118    async def resolve_agents(executor, info, status=None):119        dbpool = info.context['dbpool']120        return await Agent.load_all(dbpool, status=status)121    @staticmethod122    async def resolve_keypair(executor, info, access_key=None):123        manager = info.context['dlmgr']124        if access_key is None:125            access_key = info.context['access_key']126        loader = manager.get_loader('KeyPair.by_ak')127        return await loader.load(access_key)128    @staticmethod129    async def resolve_keypairs(executor, info, user_id, is_active=None):130        manager = info.context['dlmgr']131        loader = manager.get_loader('KeyPair.by_uid', is_active=is_active)132        return await loader.load(user_id)133    @staticmethod134    async def resolve_vfolders(executor, info, access_key=None):135        manager = info.context['dlmgr']136        if access_key is None:137            access_key = info.context['access_key']138        loader = manager.get_loader('VirtualFolder')139        return await loader.load(access_key)140    @staticmethod141    async def resolve_compute_sessions(executor, info, access_key=None, status=None):142        manager = info.context['dlmgr']143        # TODO: make status a proper graphene.Enum type144        #       (https://github.com/graphql-python/graphene/issues/544)145        if access_key is None:146            access_key = info.context['access_key']147        if status is not None:148            status = KernelStatus[status]149        loader = manager.get_loader('ComputeSession', status=status)150        return await loader.load(access_key)151    @staticmethod152    async def resolve_compute_workers(executor, info, sess_id, status=None):153        manager = info.context['dlmgr']154        if status is not None:155            status = KernelStatus[status]156        loader = manager.get_loader('ComputeWorker', status=status)157        return await loader.load(sess_id)158class QueryForUser(graphene.ObjectType):159    '''160    Available GraphQL queries for the user priveilege.161    It only allows use of the access key specified in the authorization header.162    '''163    keypair = graphene.Field(lambda: KeyPair)164    vfolders = graphene.List(VirtualFolder)165    compute_sessions = graphene.List(166        ComputeSession,167        status=graphene.String())168    compute_workers = graphene.List(169        ComputeWorker,170        sess_id=graphene.String(required=True),171        status=graphene.String())172    @staticmethod173    async def resolve_keypair(executor, info):174        manager = info.context['dlmgr']175        access_key = info.context['access_key']176        loader = manager.get_loader('KeyPair.by_ak')177        return await loader.load(access_key)178    @staticmethod179    async def resolve_vfolders(executor, info):180        manager = info.context['dlmgr']181        access_key = info.context['access_key']182        loader = manager.get_loader('VirtualFolder')183        return await loader.load(access_key)184    @staticmethod185    async def resolve_compute_sessions(executor, info, status=None):186        manager = info.context['dlmgr']187        access_key = info.context['access_key']188        # TODO: make status a proper graphene.Enum type189        #       (https://github.com/graphql-python/graphene/issues/544)190        if status is not None:191            status = KernelStatus[status]192        loader = manager.get_loader('ComputeSession', status=status)193        return await loader.load(access_key)194    @staticmethod195    async def resolve_compute_workers(executor, info, sess_id, status=None):196        manager = info.context['dlmgr']197        access_key = info.context['access_key']198        if status is not None:199            status = KernelStatus[status]200        loader = manager.get_loader(201            'ComputeWorker', status=status, access_key=access_key)202        return await loader.load(sess_id)203async def init(app):204    loop = asyncio.get_event_loop()205    app.router.add_route('POST', r'/v{version:\d+}/admin/graphql', handle_gql)206    app['admin.gql_executor'] = AsyncioExecutor(loop=loop)207    app['admin.gql_schema_admin'] = graphene.Schema(208        query=QueryForAdmin,209        mutation=MutationForAdmin,210        auto_camelcase=False)211    app['admin.gql_schema_user'] = graphene.Schema(212        query=QueryForUser,213        mutation=None,214        auto_camelcase=False)215async def shutdown(app):216    pass217if __name__ == '__main__':218    # If executed as a main program, print all GraphQL schemas.219    # (graphene transforms our object model into a textual representation)220    # This is useful for writing documentation!221    admin_schema = graphene.Schema(222        query=QueryForAdmin,223        mutation=MutationForAdmin,224        auto_camelcase=False)225    user_schema = graphene.Schema(226        query=QueryForUser,227        mutation=None,228        auto_camelcase=False)229    print('======== Admin Schema ========')230    print(str(admin_schema))231    print('======== User Schema ========')...

session.py

Source:session.py

...48        self.cleanups = CleanupManager()49        self._skip_exc_types = (exceptions.SkipTest,)50    def notify_internal_error(self):51        self._has_internal_errors = True52    def has_internal_errors(self):53        return self._has_internal_errors54    def register_skip_exception(self, exc_type):55        self._skip_exc_types += (exc_type,)56    def get_skip_exception_types(self):57        return self._skip_exc_types58    def has_children(self):59        return not self.parallel_manager is None60    @property61    def started(self):62        return self._started63    def activate(self):64        assert not self._active, 'Attempted to activate an already-active session'65        with handling_exceptions():66            ctx.push_context()...

suite_builder.py

Source:suite_builder.py

...32        self.path = path33        os.makedirs(path)34    def run(self, *args):35        app = slash_run(munch.Munch(argv=[self.path] + list(args), cmd="run"))36        assert not app.session.has_internal_errors(), 'Session has internal errors!'37        return SuiteBuilderSuiteResult(app)38class SuiteBuilderSuiteResult(object):39    def __init__(self, slash_app):40        self.slash_app = slash_app41    def assert_success(self, num_tests):42        return self.assert_all(num_tests).success()43    def assert_all(self, num_tests):44        assert len(self.slash_app.session.results) == num_tests45        return AssertAllHelper(self)46    def assert_results(self, num_results):47        results = list(self.slash_app.session.results.iter_test_results())48        assert len(results) == num_results49        return results50    def assert_session_error(self, error_substring):...

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.