| #// 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 unittest, threading, time, urllib2, shutil, os |
| from AppAgent import * |
| from test.utils.Utils import * |
| from utils.MockedHandler import * |
| |
| class AppAgentTest(unittest.TestCase): |
| |
| def runServer(self): |
| # start AppAgent |
| self.server.serve_forever() |
| |
| def setUp(self): |
| # create test directory |
| if os.path.exists(HTTPSERVER_DIR): |
| shutil.rmtree(HTTPSERVER_DIR) |
| os.makedirs(HTTPSERVER_DIR) |
| # change to test directory that the http server will serve |
| self.directory = os.getcwd() |
| os.chdir(HTTPSERVER_DIR) |
| # a mocked microservice |
| self.handler = MockedDsRestApiHandler() |
| # create a server on a free port |
| self.port = 8000 |
| while True: |
| try: |
| self.server = ThreadedHTTPServer(('localhost', self.port), MainHandler) |
| break |
| except: |
| self.port += 1 |
| # mock user credentials to always return the 'admin' user |
| self.server.getUserCredentials = lambda headers: ADMIN |
| # add the mocked handler |
| self.server.requestHandlers = {'api.dummy': self.handler} |
| # start the server on a separate thread |
| self.serverThread = threading.Thread(target = self.runServer) |
| self.serverThread.daemon = True |
| self.serverThread.start() |
| # check if the server is running after 0.5 seconds |
| time.sleep(0.5) |
| self.assertTrue(self.serverThread.isAlive(), 'Server failed to start') |
| |
| def tearDown(self): |
| # change back to the original directory |
| os.chdir(self.directory) |
| # stop the server |
| self.server.shutdown() |
| # check if the server stopped after 0.5 seconds |
| time.sleep(0.5) |
| self.assertFalse(self.serverThread.isAlive(), 'Server thread is still alive after shutdown') |
| # delete http server test dir |
| if os.path.exists(HTTPSERVER_DIR): |
| shutil.rmtree(HTTPSERVER_DIR) |
| |
| def test_fileHandling(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_fileHandling |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - AppAgent |
| Requirement: file handling in AppAgent http server |
| Action_To_Be_taken: |
| check if server dir is empty |
| create a new directory and 2 new files inside it |
| check if the files can be listed |
| check if the file contents are correct |
| delete the new directory |
| check if server dir is empty again |
| Expected_Result: pass |
| ''' |
| |
| # check if server dir is empty |
| files = [file['fileName'] for file in json.loads(self.sendRequest('/', 'LSDIR'))['fileList']] |
| self.assertEqual(0, len(files), 'Http server test directory should be empty initially') |
| # create a new directory and 2 new files inside it |
| self.sendRequest('/newdir', 'MKDIR') |
| self.sendRequest('/newdir/file1.txt', 'PUT', 'str1') |
| self.sendRequest('/newdir/file2.txt', 'PUT', 'str2') |
| # check if the files can be listed |
| files = [file['fileName'] for file in json.loads(self.sendRequest('/newdir', 'LSDIR'))['fileList']] |
| self.assertEqual(2, len(files), 'Two files should have been created') |
| self.assertIn('newdir/file1.txt', files) |
| self.assertIn('newdir/file2.txt', files) |
| # check if the file contents are correct |
| self.assertEqual('str1', self.sendRequest('/newdir/file1.txt')) |
| self.assertEqual('str2', self.sendRequest('/newdir/file2.txt')) |
| # delete the new directory |
| self.sendRequest('/newdir', 'RMDIR') |
| # check if server dir is empty again |
| files = [file['fileName'] for file in json.loads(self.sendRequest('/', 'LSDIR'))['fileList']] |
| self.assertEqual(0, len(files)) |
| |
| def test_proxy(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_proxy |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - AppAgent |
| Requirement: proxy request handling in AppAgent http server |
| Action_To_Be_taken: |
| check if server dir is empty |
| create a new directory and 2 new files inside it |
| check if the files can be listed |
| check if the file contents are correct |
| delete the new directory |
| check if server dir is empty again |
| Expected_Result: pass |
| ''' |
| |
| # check if server dir is empty |
| files = [file['fileName'] for file in json.loads(self.sendRequest(self.proxyEncode('/'), 'LSDIR'))['fileList']] |
| self.assertEqual(0, len(files), 'Http server test directory should be empty initially') |
| # create a new directory and 2 new files inside it |
| self.sendRequest(self.proxyEncode('/newdir'), 'MKDIR') |
| self.sendRequest(self.proxyEncode('/newdir/file1.txt'), 'PUT', 'str1') |
| self.sendRequest(self.proxyEncode('/newdir/file2.txt'), 'PUT', 'str2') |
| # check if the files can be listed |
| files = [file['fileName'] for file in json.loads(self.sendRequest(self.proxyEncode('/newdir'), 'LSDIR'))['fileList']] |
| self.assertEqual(2, len(files), 'Two files should have been created') |
| self.assertIn('newdir/file1.txt', files) |
| self.assertIn('newdir/file2.txt', files) |
| # check if the file contents are correct |
| self.assertEqual('str1', self.sendRequest(self.proxyEncode('/newdir/file1.txt'))) |
| self.assertEqual('str2', self.sendRequest(self.proxyEncode('/newdir/file2.txt'))) |
| # delete the new directory |
| self.sendRequest(self.proxyEncode('/newdir'), 'RMDIR') |
| # check if server dir is empty again |
| files = [file['fileName'] for file in json.loads(self.sendRequest(self.proxyEncode('/'), 'LSDIR'))['fileList']] |
| self.assertEqual(0, len(files)) |
| |
| def test_handlerHandling(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_handlerHandling |
| Tested component: ServiceFramework |
| Feature: ServiceFramework - AppAgent |
| Requirement: microservice request handler handling in AppAgent http server |
| Action_To_Be_taken: |
| check requests and their responses |
| check a request through proxy |
| check handling a wrong request |
| Expected_Result: pass |
| ''' |
| |
| # check requests and their responses |
| self.assertEqual(json.loads(self.sendRequest('/api.dummy', 'POST', '{"requests": [{"getData": {"source": "Dummy_DS", "element": "int"}}]}'))['contentList'][0], self.handler.elements['int']) |
| self.assertEqual(json.loads(self.sendRequest('/api.dummy', 'POST', '{"requests": [{"getData": {"source": "Dummy_DS", "element": "string"}}]}'))['contentList'][0], self.handler.elements['string']) |
| self.assertEqual(json.loads(self.sendRequest('/api.dummy', 'POST', '{"requests": [{"getData": {"source": "Dummy_DS", "element": "list"}}]}'))['contentList'][0], self.handler.elements['list']) |
| # check a request through proxy |
| self.assertEqual(json.loads(self.sendRequest(self.proxyEncode('/api.dummy'), 'POST', '{"requests": [{"getData": {"source": "Dummy_DS", "element": "int"}}]}'))['contentList'][0], self.handler.elements['int']) |
| # check handling a wrong request |
| with self.assertRaises(urllib2.HTTPError) as context: |
| self.sendRequest('/api.notExistingApp', 'POST', 'dummy_body') |
| self.assertIn('404', str(context.exception)) |
| |
| def proxyEncode(self, path): |
| return '/proxy/' + ('http://localhost:' + str(self.port)).encode('hex') + path |
| |
| def sendRequest(self, path, method = 'GET', body = None): |
| request = urllib2.Request('http://localhost:' + str(self.port) + path, body) |
| request.get_method = lambda: method |
| return urllib2.urlopen(request).read() |
