KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Slash automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
