Name changed to voltha-test for (23c9607) :
(This is a wrapper done using cord-tester framework)
Documentation will be updated accordingly.
./voltha-test.py -h
usage: voltha-test.py [-h] [-test-type TEST_TYPE] [-manifest MANIFEST]
[-voltha-host VOLTHA_HOST] [-olt-type OLT_TYPE]
[-olt-arg OLT_ARG] [-disable-teardown] [-disable-cleanup]
[-test-mode]
VOLTHA tester
optional arguments:
-h, --help show this help message and exit
-test-type TEST_TYPE, --test-type TEST_TYPE
Test type to run
-manifest MANIFEST, --manifest MANIFEST
Manifest file to use
-voltha-host VOLTHA_HOST, --voltha-host VOLTHA_HOST
VOLTHA host ip
-olt-type OLT_TYPE, --olt-type OLT_TYPE
OLT type
-olt-arg OLT_ARG, --olt-arg OLT_ARG
OLT type argument
-disable-teardown, --disable-teardown
Disable VOLTHA teardown
-disable-cleanup, --disable-cleanup
Dont cleanup cord-tester
-test-mode, --test-mode
Directly run the cord-tester run-test phase without
setup and cleanup
For example:
./voltha-test.py --test-type=tls:eap_auth_exchange.test_eap_tls --manifest=manifest-olt-voltha-local.json
--voltha-host=<voltha node ip> --olt-type=<vendor> --olt-arg=<mac/ip>
--disable-teardown --disable-cleanup --test-mode
Change-Id: Ic28d8486a13274387214dce39d5aeac748205d40
diff --git a/src/test/setup/voltha-test.py b/src/test/setup/voltha-test.py
new file mode 100755
index 0000000..a08c662
--- /dev/null
+++ b/src/test/setup/voltha-test.py
@@ -0,0 +1,192 @@
+#!/usr/bin/env python
+
+# Copyright 2017-present Open Networking Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from argparse import ArgumentParser
+import os
+import sys
+utils_dir = os.path.join( os.path.dirname(os.path.realpath(__file__)), '../utils')
+cli_dir = os.path.join( os.path.dirname(os.path.realpath(__file__)), '../cli')
+sys.path.append(utils_dir)
+sys.path.append(cli_dir)
+sys.path.insert(1, '/usr/local/lib/python2.7/dist-packages')
+from CordTestUtils import getstatusoutput
+import time
+import requests
+import httplib
+import json
+import signal
+
+class CordTesterWebClient(object):
+
+ def __init__(self, host = 'localhost', port = 5000):
+ self.host = host
+ self.port = port
+ self.rest = 'http://{}:{}'.format(self.host, self.port)
+
+ def get_config(self, test_case):
+ rest_uri = '{}/get'.format(self.rest)
+ config = { 'test_case' : test_case }
+ resp = requests.get(rest_uri, data = json.dumps(config))
+ if resp.ok and resp.status_code == 200:
+ config = resp.json()
+ return config
+ return None
+
+ def set_config(self, test_case, test_config):
+ rest_uri = '{}/update'.format(self.rest)
+ config = { 'test_case' : test_case, 'config' : test_config }
+ resp = requests.post(rest_uri, data = json.dumps(config))
+ return resp.ok, resp.status_code
+
+ def restore_config(self, test_case):
+ rest_uri = '{}/restore'.format(self.rest)
+ config = { 'test_case' : test_case }
+ resp = requests.post(rest_uri, data = json.dumps(config))
+ return resp.ok, resp.status_code
+
+ def start(self, manifest = 'manifest.json'):
+ rest_uri = '{}/start'.format(self.rest)
+ config = { 'manifest' : manifest }
+ resp = requests.post(rest_uri, data = json.dumps(config))
+ return resp.ok, resp.status_code
+
+ def cleanup(self, manifest = 'manifest.json'):
+ rest_uri = '{}/cleanup'.format(self.rest)
+ config = { 'manifest' : manifest }
+ resp = requests.post(rest_uri, data = json.dumps(config))
+ return resp.ok, resp.status_code
+
+ def test(self, test, manifest = 'manifest.json', test_config = None):
+ rest_uri = '{}/test'.format(self.rest)
+ config = { 'manifest' : manifest, 'test' : test }
+ if test_config:
+ config['config'] = test_config
+ resp = requests.post(rest_uri, data = json.dumps(config))
+ return resp.ok, resp.status_code
+
+class Tester(CordTesterWebClient):
+
+ def __init__(self, host = 'localhost', port = 5000):
+ super(Tester, self).__init__(host = host, port = port)
+
+ def execute(self, test_case, manifest = 'manifest.json', test_config = None):
+ print('Executing test %s' %test_case)
+ _, status = self.start(manifest = manifest)
+ assert status == httplib.OK, 'Test setup failed with status code %d' %status
+ _, status = self.test(test_case, manifest = manifest, test_config = test_config)
+ assert status == httplib.OK, 'Test run for test %s failed with status %d' %(test_case, status)
+ _, status = self.cleanup(manifest = manifest)
+ assert status == httplib.OK, 'Test cleanup failed with status %d' %status
+ print('Test executed successfully')
+
+
+class CordTesterWeb(object):
+
+ def __init__(self, args, start_in = 3):
+ self.args = args
+ self.tester = Tester()
+ self.start_in = start_in
+
+ def run(self):
+ manifest = self.args.manifest
+ olt_type = self.args.olt_type
+ test_type = self.args.test_type
+ disable_teardown = self.args.disable_teardown
+ test_mode = self.args.test_mode
+ disable_cleanup = self.args.disable_cleanup
+ if test_mode is True:
+ disable_cleanup = True
+ test_config = { 'VOLTHA_HOST' : self.args.voltha_host,
+ 'VOLTHA_OLT_TYPE' : self.args.olt_type,
+ 'VOLTHA_TEARDOWN' : not disable_teardown,
+ }
+ if olt_type.startswith('tibit'):
+ test_config['VOLTHA_OLT_MAC'] = self.args.olt_arg
+ elif olt_type.startswith('maple'):
+ test_config['VOLTHA_OLT_IP'] = self.args.olt_arg
+ elif olt_type.startswith('ponsim'):
+ test_config['VOLTHA_PONSIM_HOST'] = self.args.olt_arg
+ else:
+ print('Unsupported OLT type %s' %olt_type)
+ return 127
+
+ if self.start_in:
+ time.sleep(self.start_in)
+
+ if test_mode is False:
+ _, status = self.tester.start(manifest = manifest)
+ assert status == httplib.OK, 'Test setup failed with status %d' %status
+
+ for test in test_type.split(','):
+ print('Running test case %s' %(test))
+ _, status = self.tester.test(test, manifest = manifest, test_config = test_config)
+ if status != httplib.OK:
+ print('Test case %s failed with status code %d' %(test, status))
+
+ if disable_cleanup is False:
+ print('Cleaning up the test')
+ self.tester.cleanup(manifest = manifest)
+ return 0 if status == httplib.OK else 127
+
+class CordTesterWebServer(object):
+
+ server_path = os.path.dirname(os.path.realpath(__file__))
+ server = 'webserver-run.py'
+ pattern = 'pgrep -f "python ./{}"'.format(server)
+
+ def running(self):
+ st, _ = getstatusoutput(self.pattern)
+ return True if st == 0 else False
+
+ def kill(self):
+ st, output = getstatusoutput(self.pattern)
+ if st == 0 and output:
+ pids = output.strip().splitlines()
+ for pid in pids:
+ try:
+ os.kill(int(pid), signal.SIGKILL)
+ except:
+ pass
+
+ def start(self):
+ if self.running() is False:
+ print('Starting CordTester Web Server')
+ cmd = 'cd {} && python ./{} &'.format(self.server_path, self.server)
+ os.system(cmd)
+
+def run_test(args):
+ testWebServer = CordTesterWebServer()
+ testWebServer.start()
+ testWeb = CordTesterWeb(args, start_in = 3)
+ status = testWeb.run()
+ testWebServer.kill()
+ return status
+
+if __name__ == '__main__':
+ parser = ArgumentParser(description = 'VOLTHA tester')
+ parser.add_argument('-test-type', '--test-type', default = 'tls:eap_auth_exchange.test_eap_tls', help = 'Test type to run')
+ parser.add_argument('-manifest', '--manifest', default='manifest-voltha.json', help = 'Manifest file to use')
+ parser.add_argument('-voltha-host', '--voltha-host', default='172.17.0.1', help = 'VOLTHA host ip')
+ parser.add_argument('-olt-type', '--olt-type', default = 'ponsim_olt', help = 'OLT type')
+ parser.add_argument('-olt-arg', '--olt-arg', default = '172.17.0.1', help = 'OLT type argument')
+ parser.add_argument('-disable-teardown', '--disable-teardown', action='store_true', help = 'Disable VOLTHA teardown')
+ parser.add_argument('-disable-cleanup', '--disable-cleanup', action='store_true', help = 'Dont cleanup cord-tester')
+ parser.add_argument('-test-mode', '--test-mode', action='store_true',
+ help = 'Directly run the cord-tester run-test phase without setup and cleanup')
+
+ parser.set_defaults(func = run_test)
+ args = parser.parse_args()
+ res = args.func(args)
+ sys.exit(res)