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

How to use ServiceMethod method in autotest

Best Python code snippet using autotest_python

services.py

Source:services.py

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': self.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            })...

webServices.py

Source:webServices.py

1import requests2import json3class webServices:4	_url = "http://aldbnagioscd.dev.att.com:8080/ALDB-NAG/NagiosService"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		...

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.