Python webserver client example to run tests through rest interface.
Change-Id: Ib8e2c014e290a8fc9997a8ad8a33286d3259854e
diff --git a/src/test/setup/webserver-client-sample.py b/src/test/setup/webserver-client-sample.py
new file mode 100644
index 0000000..65932bf
--- /dev/null
+++ b/src/test/setup/webserver-client-sample.py
@@ -0,0 +1,95 @@
+import requests
+import json
+import httplib
+
+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')
+
+if __name__ == '__main__':
+ tester = Tester()
+ tests = ('tls', 'igmp',)
+ for test in tests:
+ print('Getting config for test %s' %test)
+ config = tester.get_config(test)
+ print('%s' %config)
+
+ tls_cfg = { 'VOLTHA_OLT_MAC' : '00:0c:e2:31:10:00' }
+ igmp_cfg = { 'PORT_RX_DEFAULT' : 1, 'PORT_TX_DEFAULT' : 2, 'IGMP_TEST_TIMEOUT' : 10 }
+ manifest = 'manifest-ponsim.json'
+ tests = ( ('tls:eap_auth_exchange.test_eap_tls', tls_cfg, manifest),
+ ('igmp:igmp_exchange.test_igmp_join_verify_traffic', igmp_cfg, manifest),
+ )
+
+ print('Setting up the test with %s' %manifest)
+ _, status = tester.start(manifest = manifest)
+ assert status == httplib.OK, 'Test setup failed with status code %d' %status
+
+ for t, cfg, m in tests:
+ _, status = tester.test(t, manifest = m, test_config = cfg)
+ if status != httplib.OK:
+ print('Test case %s failed with status code %d' %(t, status))
+ else:
+ print('Test case %s executed successfully' %t)
+
+ print('Cleaning up the test with %s' %manifest)
+ tester.cleanup(manifest = manifest)