How to use ServiceMethod method in autotest

Best Python code snippet using autotest_python Github


Full Screen

1import sys, inspect2from django.utils import simplejson3from exceptions import ServiceException4def servicemethod(*args, **kwargs):5 """ The Service method decorator.6 Decorate a function or method to expose it remotely7 via RPC (or other mechanism.)8 Arguments:9 name (optional):10 The name of this method as seen remotely.11 store (required if not decorating a bound Store method):12 A reference to the Store this method operates on.13 This is required if the method is a regular function,14 a staticmethod or otherwise defined outside a Store instance.15 (ie doesn't take a 'self' argument)16 store_arg (optional):17 Specifies whether this method should be passed the Store instance18 as the first argument (default is True so that servicemethods bound to19 a store instance can get a proper 'self' reference.)20 request_arg (optional):21 Specifies whether this method should be passed a reference to the current22 Request object. (Default is True)23 If both store_arg and request_arg are True, the the store will be passed first,24 then the request (to appease bound store methods that need a 'self' as the first arg)25 If only one is True then that one will be passed first. This is useful for using26 standard Django view functions as servicemethods since they require the 'request'27 as the first argument.28 """29 # Default options30 options = {'name': None, 'store': None, 'request_arg': True, 'store_arg': True}31 # Figure out if we were called with arguments32 # If we were called with args, ie:33 # @servicemethod(name='Foo')34 # Then the only argument here will be the pre-decorated function/method object.35 method = ( (len(args) == 1) and callable(args[0]) ) and args[0] or None36 if method is None:37 # We were called with args, (or @servicemethod() )38 # so figure out what they were ...39 # The method name should be either the first non-kwarg40 # or the kwarg 'name'41 # Example: @servicemethod('my_method', ...) or @servicemethod(name='my_method')42 options.update({43 'name': bool(args) and args[0] or kwargs.pop('name', None),44 'store': (len(args) >= 2) and args[1] or kwargs.pop('store', None),45 'request_arg': kwargs.pop('request_arg', True),46 'store_arg': kwargs.pop('store_arg', True),47 })48 else:49 options['name'] = method.__name__50 method.__servicemethod__ = options51 def method_with_args_wrapper(method):52 """ Wrapper for a method decorated with decorator arguments53 """54 if options['name'] is None:55 options['name'] = method.__name__56 method.__servicemethod__ = options57 if options['store'] is not None:58 options['store'].service.add_method(method)59 return method60 return method or method_with_args_wrapper61class BaseService(object):62 """ The base Service class that manages servicemethods and63 service method descriptions64 """65 def __init__(self):66 """ BaseService constructor67 """68 self.methods = {}69 self._store = None70 def _get_store(self):71 """ Property getter for the store this service is72 bound to73 """74 return self._store75 def _set_store(self, store):76 """ Property setter for the store this service is77 bound to. Automatically updates the store78 reference in all the __servicemethod__79 properties on servicemethods in this service80 """81 for method in self.methods.values():82 method.__servicemethod__['store'] = store83 self._store = store84 store = property(_get_store, _set_store)85 def _get_method_args(self, method, request, params):86 """ Decide if we should pass store_arg and/or request_arg87 to the servicemethod88 """89 idx = 090 if method.__servicemethod__['store_arg']:91 params.insert(idx, method.__servicemethod__['store'])92 idx += 193 if method.__servicemethod__['request_arg']:94 params.insert(idx, request)95 return params96 def add_method(self, method, name=None, request_arg=True, store_arg=True):97 """ Adds a method as a servicemethod to this service.98 """99 # Was this a decorated servicemethod?100 if hasattr(method, '__servicemethod__'):101 options = method.__servicemethod__102 else:103 options = {'name': name or method.__name__, 'store':,104 'request_arg': request_arg, 'store_arg': store_arg}105 method.__servicemethod__ = options106 self.methods[ options['name'] ] = method107 def get_method(self, name):108 """ Returns the servicemethod given by name109 """110 try:111 return self.methods[name]112 except KeyError:113 raise ServiceException('Service method "%s" not registered' % name)114 def list_methods(self):115 """ Returns a list of all servicemethod names116 """117 return self.methods.keys()118 def process_request(self, request):119 """ Processes a request object --120 This is generally the entry point for all121 servicemethod calls122 """123 raise NotImplementedError('process_request not implemented in BaseService')124 def process_response(self, id, result):125 """ Prepares a response from a servicemethod call126 """127 raise NotImplementedError('process_response not implemented in BaseService')128 def process_error(self, id, code, error):129 """ Prepares an error response from a servicemethod call130 """131 raise NotImplementedError('process_error not implemented in BaseService')132 def get_smd(self, url):133 """ Returns a service method description of all public servicemethods134 """135 raise NotImplementedError('get_smd not implemented in BaseService')136class JsonService(BaseService):137 """ Implements a JSON-RPC version 1.1 service138 """139 def __call__(self, request):140 """ JSON-RPC method calls come in as POSTs141 --142 Requests for the SMD come in as GETs143 """144 if request.method == 'POST':145 response = self.process_request(request)146 else:147 response = self.get_smd(request.get_full_path())148 return simplejson.dumps(response)149 def process_request(self, request):150 """ Handle the request151 """152 try:153 data = simplejson.loads(request.raw_post_data)154 id, method_name, params = data["id"], data["method"], data["params"]155 # Doing a blanket except here because God knows kind of crazy156 # POST data might come in.157 except:158 return self.process_error(0, 100, 'Invalid JSON-RPC request')159 try:160 method = self.get_method(method_name)161 except ServiceException:162 return self.process_error(id, 100, 'Unknown method: "%s"' % method_name)163 params = self._get_method_args(method, request, params)164 try:165 result = method(*params)166 return self.process_response(id, result)167 except BaseException:168 etype, eval, etb = sys.exc_info()169 return self.process_error(id, 100, '%s: %s' % (etype.__name__, eval) )170 except:171 etype, eval, etb = sys.exc_info()172 return self.process_error(id, 100, 'Exception %s: %s' % (etype, eval) )173 def process_response(self, id, result):174 """ Build a JSON-RPC 1.1 response dict175 """176 return {177 'version': '1.1',178 'id': id,179 'result': result,180 'error': None,181 }182 def process_error(self, id, code, error):183 """ Build a JSON-RPC 1.1 error dict184 """185 return {186 'id': id,187 'version': '1.1',188 'error': {189 'name': 'JSONRPCError',190 'code': code,191 'message': error,192 },193 }194 def get_smd(self, url):195 """ Generate a JSON-RPC 1.1 Service Method Description (SMD)196 """197 smd = {198 'serviceType': 'JSON-RPC',199 'serviceURL': url,200 'methods': []201 }202 for name, method in self.methods.items():203 # Figure out what params to report --204 # we don't want to report the 'store' and 'request'205 # params to the remote method.206 idx = 0207 idx += method.__servicemethod__['store_arg'] and 1 or 0208 idx += method.__servicemethod__['request_arg'] and 1 or 0209 sig = inspect.getargspec(method)210 smd['methods'].append({211 'name': name,212 'parameters': [ {'name': val} for val in sig.args[idx:] ]213 })...

Full Screen

Full Screen Github


Full Screen

1import requests2import json3class webServices:4 _url = ""5 6 def __init__(self, serviceMethod): 7 self.serviceMethod = serviceMethod8 9 @property10 def serviceMethod(self):11 return self._serviceMethod12 13 @serviceMethod.setter14 def serviceMethod(self, serviceMethod):15 self._serviceMethod = serviceMethod16 def __str__(self):17 return self._url +"/"+ self.serviceMethod18 19class doRequest(webServices):20 def __init__(self, serviceMethod, **params):21 super().__init__(serviceMethod)22 self.param = params23 24 @property25 def param(self):26 return self._param27 28 @param.setter29 def param(self, params):30 self._param = params 31 32 def _getMethodParam(self):33 if(self.serviceMethod == "fetchKMDetails"):34 return '{}/format.json'.format(self.param['serverName'])35 elif(self.serviceMethod == "fetchRecoveryKMDetails"):36 return '{}/format.json'.format(self.param['serverName'])37 elif(self.serviceMethod == "fetchUrlDetails"):38 return '{}/format.json'.format(self.param['serverName'])39 elif(self.serviceMethod == "fetchRecoveryUrlDetails"):40 return '{}/format.json'.format(self.param['serverName'])41 elif(self.serviceMethod == "updateDetails"):42 ''' This method would only update server flag.return code (0 -> Flag update successfully; 1 -> Not updated), check applied only for return code 0, not 1'''43 return '{}/format.json?{}'.format(self.param['serverName'], self.param['returnCode'])44 elif(self.serviceMethod == "addServer"):45 return '{}/{}/{}/{}/format.json? primaryXiServer={}&secondaryXiServer={}'.format(self.param['hostName'], self.param['agentName'], self.param['platform'], self.param['starVersion'], self.param['primaryXiServerName'], self.param['secondaryXiServerName'])46 elif(self.serviceMethod == "deleteServer"):47 return '{}/format.json'.format(self.param['hostName'])48 elif(self.serviceMethod == "addServerToServerGroup"):49 ''' Profile and server group should be exist beore server add 50 Server should already exist in ALDB server list. Use addServer method to add server to ALDB server list.51 '''52 return '{}/{}/{}/format.json'.format(self.param['profileName'],self.param['serverGroupName'],self.param['serverName'])53 elif(self.serviceMethod == "submitProfile"):54 '''This will submit the profile for monitoring'''55 return '{}/format.json'.format(self.param['profileName'])56 57 def _call(self):58 response=""59 try:60 response = requests.get(super().__str__() + "/" + self._getMethodParam())61 return response.json()62 except KeyError:63 print("Please validate parameters")64 except TypeError:65 print("Method does not exist")66 ...

Full Screen

Full Screen

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:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?