| #// Copyright (c) 2000-2018 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 unittest, json |
| import Microservices.DataSource |
| from utils.MockedHandler import * |
| from test.utils.Utils import * |
| |
| class DataSourceTest(unittest.TestCase): |
| |
| def setUp(self): |
| # the "microservice" under test |
| self.handler = Microservices.DataSource.createHandler('src/Microservices/DataSource') |
| # a mocked microservice |
| self.mockedHandler = MockedDsRestApiHandler() |
| # the datasource handlers of the microservice |
| self.datasourceHandlers = self.handler.getDataSourceHandlers() |
| # setting the handlers of DataSource |
| handlers = {} |
| handlers.update(self.mockedHandler.getDataSourceHandlers()) |
| handlers.update(self.datasourceHandlers) |
| self.handler.setHandlers(handlers) |
| |
| def tearDown(self): |
| # close tested "microservice" |
| self.handler.close() |
| |
| def test_builtInElements(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_builtInElements |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - DataSource |
| Requirement: DataSource built in elements |
| Action_To_Be_taken: |
| check source id |
| check request response pairs |
| Expected_Result: pass |
| ''' |
| |
| # check source id |
| self.assertEqual('DataSource', self.handler.SOURCE_ID) |
| |
| # check request response pairs: expected, element, param1, param2 |
| self.checkElement('true', 'not', 'false') |
| self.checkElement('false', 'not', 'true') |
| |
| self.checkElement('true', '==', '10.0', '10.0') |
| self.checkElement('false', '==', '20.0', '10.0') |
| |
| self.checkElement('false', '!=', '10.0', '10.0') |
| self.checkElement('true', '!=', '321.0', '10.0') |
| |
| self.checkElement('false', '>', '10.0', '20.0') |
| self.checkElement('true', '>', '20.0', '10.0') |
| self.checkElement('false', '>', '10.0', '10.0') |
| |
| self.checkElement('false', '>=', '10.0', '20.0') |
| self.checkElement('true', '>=', '20.0', '10.0') |
| self.checkElement('true', '>=', '10.0', '10.0') |
| |
| self.checkElement('true', '<', '10.0', '20.0') |
| self.checkElement('false', '<', '20.0', '10.0') |
| self.checkElement('false', '<', '10.0', '10.0') |
| |
| self.checkElement('true', '<=', '10.0', '20.0') |
| self.checkElement('false', '<=', '20.0', '10.0') |
| self.checkElement('true', '<=', '10.0', '10.0') |
| |
| self.checkElement('true', 'and', 'true', 'true') |
| self.checkElement('false', 'and', 'true', 'false') |
| self.checkElement('false', 'and', 'false', 'true') |
| self.checkElement('false', 'and', 'false', 'false') |
| |
| self.checkElement('true', 'or', 'true', 'true') |
| self.checkElement('true', 'or', 'true', 'false') |
| self.checkElement('true', 'or', 'false', 'true') |
| self.checkElement('false', 'or', 'false', 'false') |
| |
| self.checkElement('true', 'match', 'apple', 'a*l?') |
| self.checkElement('false', 'match', 'apple', 'p*') |
| |
| self.checkElement('false', 'not match', 'apple', 'a*l?') |
| self.checkElement('true', 'not match', 'apple', 'p*') |
| |
| self.checkElement('6.0', 'sum', '["1", "2", "3"]') |
| self.checkElement('10.0', 'sum', '["5.0", "5.0"]') |
| |
| self.checkElement('true', 'exists', '["false", "true", "false"]') |
| self.checkElement('false', 'exists', '["false", "false", "false"]') |
| |
| self.checkElement('false', 'forAll', '["true", "false", "true"]') |
| self.checkElement('true', 'forAll', '["true", "true", "true"]') |
| |
| response = self.get('Sources') |
| sources = [element['node']['val'] for element in response['list']] |
| self.assertEqual(2, len(sources)) |
| self.assertIn(self.handler.SOURCE_ID, sources) |
| self.assertIn(self.mockedHandler.SOURCE_ID, sources) |
| |
| self.checkEncodedRequest('2', 'sizeOf', 'Sources') |
| self.checkEncodedRequest('1', 'sizeOf', 'help') |
| self.checkEncodedRequest('true', 'dataElementPresent', 'not', 'Par1', 'false') |
| self.checkEncodedRequest('false', 'dataElementPresent', 'NoElement') |
| |
| |
| def test_externalElements(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_builtInElements |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - DataSource |
| Requirement: handling external elements in DataSource |
| Action_To_Be_taken: |
| get element and check result |
| Expected_Result: pass |
| ''' |
| |
| response = self.datasourceHandlers[self.handler.SOURCE_ID]['getDataHandler']({'source': 'Dummy_DS', 'element': 'string', 'params': []}, ADMIN) |
| self.assertEqual(response, self.mockedHandler.elements['string']) |
| |
| response = self.datasourceHandlers[self.handler.SOURCE_ID]['getDataHandler']({'source': 'NoSource', 'element': 'SomeElement', 'params': []}, ADMIN) |
| self.assertIsNone(response) |
| |
| def test_setDataHandling(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_setDataHandling |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - DataSource |
| Requirement: handling setData requests in DataSource |
| Action_To_Be_taken: |
| set element |
| check result |
| Expected_Result: pass |
| ''' |
| |
| self.datasourceHandlers[self.handler.SOURCE_ID]['setDataHandler']({'source': 'Dummy_DS', 'element': 'string', 'params': [], 'content': 'newstr', 'tp': 4, 'indxsInList': []}, ADMIN) |
| response = self.datasourceHandlers[self.handler.SOURCE_ID]['getDataHandler']({'source': 'Dummy_DS', 'element': 'string', 'params': []}, ADMIN) |
| self.assertEqual(response['node']['val'], 'newstr') |
| |
| response = self.datasourceHandlers[self.handler.SOURCE_ID]['setDataHandler']({'source': 'DataSource', 'element': 'Sources', 'params': []}, ADMIN) |
| self.assertIsNone(response) |
| |
| def test_microService(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_microService |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - DataSource |
| Requirement: DataSource behaving as a microservice |
| Action_To_Be_taken: |
| check api extension |
| send request |
| check response |
| Expected_Result: pass |
| ''' |
| |
| # check api extension |
| self.assertEqual('api.appagent', Microservices.DataSource.EXTENSION) |
| # send request |
| response = getEmptyResponse() |
| self.handler.handleMessage('POST', 'api.appagent', {}, '''{ |
| "requests": [ |
| { |
| "getData": { |
| "source": "DataSource", |
| "element": "Sources" |
| } |
| } |
| ] |
| }''', ADMIN, response) |
| # check response |
| sources = [element['node']['val'] for element in json.loads(response['body'])['contentList'][0]['list']] |
| self.assertEqual(2, len(sources)) |
| |
| def test_help(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_help |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - DataSource |
| Requirement: DataSource help |
| Action_To_Be_taken: |
| get help |
| check help syntax |
| check sources |
| Expected_Result: pass |
| ''' |
| |
| # get help |
| help = self.get('help')['node']['val'] |
| # check help syntax |
| help = json.loads(help.decode('hex')) |
| # check sources |
| sources = [element['source'] for element in help['sources']] |
| self.assertEqual(2, len(sources)) |
| self.assertIn('DataSource', sources) |
| self.assertIn('Dummy_DS', sources) |
| |
| def checkEncodedRequest(self, expected, element, originalElement, paramName = None, paramValue = None): |
| request = { |
| 'source': self.handler.SOURCE_ID, |
| 'element': element, |
| 'params': [ |
| { |
| 'paramName': 'Source', |
| 'paramValue': 'DataSource' |
| }, |
| { |
| 'paramName': 'Element', |
| 'paramValue': originalElement |
| } |
| ] |
| } |
| if paramName is not None and paramValue is not None: |
| request['params'].append({ |
| 'paramName': 'ParamName', |
| 'paramValue': paramName |
| }) |
| request['params'].append({ |
| 'paramName': 'ParamValue', |
| 'paramValue': paramValue |
| }) |
| response = self.datasourceHandlers[self.handler.SOURCE_ID]['getDataHandler'](request, ADMIN) |
| self.assertEqual(response['node']['val'], expected) |
| |
| def checkElement(self, expected, element, par1 = None, par2 = None): |
| response = self.get(element, par1, par2) |
| self.assertEqual(response['node']['val'], expected) |
| |
| def get(self, element, par1 = None, par2 = None): |
| request = { |
| 'source': self.handler.SOURCE_ID, |
| 'element': element, |
| 'params': [] |
| } |
| if par1 is not None: |
| request['params'].append({ |
| 'paramName': 'Par1', |
| 'paramValue': par1 |
| }) |
| if par2 is not None: |
| request['params'].append({ |
| 'paramName': 'Par2', |
| 'paramValue': par2 |
| }) |
| return self.datasourceHandlers[self.handler.SOURCE_ID]['getDataHandler'](request, ADMIN) |
| |
| def set(self, element, content, tp): |
| request = { |
| 'source': self.handler.SOURCE_ID, |
| 'element': element, |
| 'params': [], |
| 'tp': tp, |
| 'content': content |
| } |
| return self.datasourceHandlers[self.handler.SOURCE_ID]['setDataHandler'](request, ADMIN) |
