test/utils/MockedHandler.py

#// Copyright (c) 2000-2019 Ericsson Telecom AB                                                         //
#// All rights reserved. This program and the accompanying materials are made available under the terms //
#// of the Eclipse Public License v2.0 which accompanies this distribution, and is available at         //
#// https://www.eclipse.org/org/documents/epl-2.0/EPL-2.0.html                                                           //
#/////////////////////////////////////////////////////////////////////////////////////////////////////////
import copy, json
from Common.DsRestAPI import DsRestAPI
from Utils import *

class MockedDsRestApiHandler:

    SOURCE_ID = 'Dummy_DS'

    def __init__(self):
        self.elements = {
            "int": {"node": {"val": "1", "tp": 1}},
            "string": {"node": {"val": "str", "tp": 4}},
            "true": {"node": {"val": "true", "tp": 3}},
            "false": {"node": {"val": "false", "tp": 3}},
            "list": {"list": [{"node": {"val": str(i), "tp": 8}} for i in range(3)]},
            "stringOfList": {
                "0": {"node": {"val": "string_0", "tp": 4}},
                "1": {"node": {"val": "string_1", "tp": 4}},
                "2": {"node": {"val": "string_2", "tp": 4}}
            },
            "boolOfList": {
                "0": {"node": {"val": "true", "tp": 3}},
                "1": {"node": {"val": "true", "tp": 3}},
                "2": {"node": {"val": "false", "tp": 3}}
            },
            "boolOfParam": {
                "1": {"node": {"val": "true", "tp": 3}},
                "str": {"node": {"val": "false", "tp": 3}}
            },
            "userError": {"node": {"val": "User Error", "tp": -4}},
            "filteredItem": {"node": {"val": "", "tp": 0}},
            "help": {"node": {"val": json.dumps({'sources': [{'source': self.SOURCE_ID, 'dataElements': []}]}).encode('hex'), "tp": 5}}
        }

        self._dsRestAPI = DsRestAPI(self.mockedGetDataHandler, self.mockedSetDataHandler)

    def mockedGetDataHandler(self, request, userCredentials):
        if userCredentials['username'] == ADMIN['username'] and userCredentials['password'] == ADMIN['password'] and len(set(userCredentials['groups']) ^ set(ADMIN['groups'])) == 0:
            element = request['element']
            params = request['params']
            params.reverse()
        else:
            element = 'userError'
            params = []

        try:
            response = self.elements[element]
            while len(params) > 0:
                response = response[params.pop()['paramValue']]
            return copy.deepcopy(response)
        except:
            pass

    def mockedSetDataHandler(self, request, userCredentials):
        element = request['element']
        params = request['params']
        content = request['content']
        params.reverse()

        try:
            if userCredentials['username'] == ADMIN['username'] and userCredentials['password'] == ADMIN['password'] and len(set(userCredentials['groups']) ^ set(ADMIN['groups'])) == 0:
                response = self.elements[element]
                while len(params) > 0:
                    response = response[params.pop()['paramValue']]
                if len(request['indxsInList']) > 0:
                    response = response['list'][request['indxsInList'][0]]
                response['node']['val'] = content
                return response
        except:
            pass

    def handleMessage(self, method, path, headers, body, userCredentials, response):
        response['body'] = json.dumps(self._dsRestAPI.parseRequest(json.loads(body), userCredentials))
        response['headers']['Content-Type'] = 'application/json'

    def getDataSourceHandlers(self):
        return {
            self.SOURCE_ID: {
                'getDataHandler': self.mockedGetDataHandler,
                'setDataHandler': self.mockedSetDataHandler
            }
        }

    def close(self):
        for processId in self._processes:
            if self._processes[processId]['Process'].poll() is None:
                os.killpg(os.getpgid(self._processes[processId]['Process'].pid), signal.SIGTERM)
                self._logger.info('Stopped TitanSim instance:', processId)