| #// Copyright (c) 2000-2017 Ericsson Telecom AB // |
| #// All rights reserved. This program and the accompanying materials are made available under the terms // |
| #// of the Eclipse Public License v1.0 which accompanies this distribution, and is available at // |
| #// http://www.eclipse.org/legal/epl-v10.html // |
| #///////////////////////////////////////////////////////////////////////////////////////////////////////// |
| import unittest, threading, time, urllib2, cookielib, os |
| from AppAgent import * |
| from test.utils.Utils import * |
| |
| TEST_PLAYLIST = '''[ |
| { |
| "apiUrl": "http://localhost:8000/api.appagent", |
| "playlist": [ |
| { |
| "requests": [ |
| { |
| "getData": { |
| "source": "DataSource", |
| "element": "Sources" |
| } |
| } |
| ], |
| "condition": { |
| "evaluatingPeriod": 0.1, |
| "numberOfExecutions": 200, |
| "expression": { |
| "dataValue": "false" |
| } |
| } |
| } |
| ] |
| } |
| ]''' |
| |
| class IntegrationTest(unittest.TestCase): |
| |
| def runServer(self): |
| # start AppAgent |
| self.server.serve_forever() |
| |
| def setUp(self): |
| # url opener that acts like a browser |
| self.opener = urllib2.build_opener(urllib2.HTTPHandler(), urllib2.HTTPSHandler(), urllib2.HTTPCookieProcessor(cookielib.CookieJar()), urllib2.ProxyHandler({})) |
| # create a server on a free port |
| directory = os.path.abspath('src') |
| self.port = 8000 |
| while True: |
| try: |
| self.server = ThreadedHTTPServer(('localhost', self.port), MainHandler) |
| break |
| except: |
| self.port += 1 |
| # start the server on a separate thread |
| self.serverThread = threading.Thread(target = runAppAgent, args = (self.server, directory, 'WebGUI/src/WebGUI/htdocs', False)) |
| 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): |
| # 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') |
| |
| def test_microservicesIntegration(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_microservicesIntegration |
| Tested component: ServiceFramework |
| Feature: ServiceFramework |
| Requirement: microservices' integration into ServiceFramework |
| Action_To_Be_taken: |
| login |
| check DataSource sources |
| check playlist microservice |
| check whether 'admin' user is in the 'admin' group |
| logout |
| try and fail to use proxy without being logged in |
| Expected_Result: pass |
| ''' |
| |
| # login |
| self.sendRequest('/login/api.authenticate', 'POST', '{"username": "admin", "password": "admin"}') |
| # check DataSource sources |
| sources = [element['node']['val'] for element in json.loads(self.sendRequest('/api.appagent', 'POST', '{"requests": [{"getData": {"source": "DataSource", "element": "Sources"}}]}'))['contentList'][0]['list']] |
| self.assertEqual(3, len(sources), 'List lenght should be 3, got: ' + str(sources)) |
| self.assertIn('DataSource', sources) |
| self.assertIn('Playlist', sources) |
| self.assertIn('Authentication', sources) |
| # check playlist microservice |
| response = json.loads(self.sendRequest('/api.playlist', 'POST', '{"requests": [{"getData": {"source": "Playlist", "element": "Playlists"}}]}'))['contentList'][0] |
| self.assertIn('list', response) |
| # check whether 'admin' user is in the 'admin' group |
| groups = [element['node']['val'] for element in json.loads(self.sendRequest(self.proxyEncode('/api.authenticate'), 'POST', '{"requests": [{"getData": {"source": "Authentication", "element": "ListGroups"}}]}'))['contentList'][0]['list']] |
| self.assertIn('admin', groups) |
| # logout |
| self.sendRequest('/logout/api.authenticate', 'POST', '') |
| # try and fail to use proxy without being logged in |
| with self.assertRaises(urllib2.HTTPError) as context: |
| self.sendRequest(self.proxyEncode('/api.authenticate'), 'POST', '{"requests": [{"getData": {"source": "Authentication", "element": "ListGroups"}}]}') |
| self.assertIn('401', str(context.exception)) |
| |
| def test_playlistIntegration(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_playlistIntegration |
| Tested component: ServiceFramework |
| Feature: ServiceFramework |
| Requirement: playlist integration into ServiceFramework |
| Action_To_Be_taken: |
| login |
| create a new playlist |
| start the playlist |
| check if playlist status led is green |
| stop the playlist |
| check if playlist status led is black |
| delete playlist |
| check if the playlist no longer exists |
| Expected_Result: pass |
| ''' |
| |
| # login |
| self.sendRequest('/login/api.authenticate', 'POST', '{"username": "admin", "password": "admin"}') |
| # create a new playlist |
| self.sendRequest('/api.playlist', 'POST', '{"requests": [{"setData": {"source": "Playlist", "element": "Descriptor", "tp": 4, "content": "' + TEST_PLAYLIST.replace('8000', str(self.port)).replace('\n', '').replace('"', r'\"') + '", "params": [{"paramName": "Playlist", "paramValue": "admin/test"}]}}]}') |
| # start the playlist |
| self.sendRequest('/api.playlist', 'POST', '{"requests": [{"setData": {"source": "Playlist", "element": "Start", "params": [{"paramName": "Playlist", "paramValue": "admin/test"}]}}]}') |
| time.sleep(0.5) |
| # check if playlist status led is green |
| self.assertIn('green', json.loads(self.sendRequest('/api.playlist', 'POST', '{"requests": [{"getData": {"source": "Playlist", "element": "Status", "params": [{"paramName": "Playlist", "paramValue": "admin/test"}]}}]}'))['contentList'][0]['node']['val']) |
| # stop the playlist |
| self.sendRequest('/api.playlist', 'POST', '{"requests": [{"setData": {"source": "Playlist", "element": "Stop", "tp": 1, "content": "1", "params": [{"paramName": "Playlist", "paramValue": "admin/test"}]}}]}') |
| time.sleep(0.5) |
| # check if playlist status led is black |
| self.assertIn('black', json.loads(self.sendRequest('/api.playlist', 'POST', '{"requests": [{"getData": {"source": "Playlist", "element": "Status", "params": [{"paramName": "Playlist", "paramValue": "admin/test"}]}}]}'))['contentList'][0]['node']['val']) |
| # delete playlist |
| self.sendRequest('/api.playlist', 'POST', '{"requests": [{"setData": {"source": "Playlist", "element": "Delete", "tp": 1, "content": "1", "params": [{"paramName": "Playlist", "paramValue": "admin/test"}]}}]}') |
| time.sleep(0.5) |
| # check if the playlist no longer exists |
| playlists = [element['node']['val'] for element in json.loads(self.sendRequest('/api.playlist', 'POST', '{"requests": [{"getData": {"source": "Playlist", "element": "Playlists"}}]}'))['contentList'][0]['list']] |
| self.assertNotIn('admin/test', playlists) |
| |
| def test_fileHandling(self): |
| ''' |
| Author: EDNIGBO Daniel Gobor |
| Testcase: test_fileHandling |
| Tested component: ServiceFramework |
| Feature: ServiceFramework |
| Requirement: serving files for browsers |
| Action_To_Be_taken: |
| get index.html |
| check contents |
| Expected_Result: pass |
| ''' |
| |
| # get index.html |
| response = self.sendRequest('') |
| # check contents |
| self.assertIn('TSGuiFrameworkMain', response) |
| |
| 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 self.opener.open(request).read() |
