--- /dev/null
+"""
+CSIT test tools.
+Homepage: https://github.com/yeasy/CSIT_Test
+Updated: 2013-11-07
+"""
+__all__ = ['restlib', 'testmodule']
\ No newline at end of file
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-01
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class ArpHandler(TestModule):
+ """
+ Test for the arp handler.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+
+ def __init__(self, restSubContext='/controller/nb/v2/subnetservice', user=DEFAULT_USER, password=DEFAULT_PWD,
+ container=DEFAULT_CONTAINER, contentType='json', prefix=DEFAULT_PREFIX):
+ super(self.__class__, self).__init__(restSubContext, user, password, container, contentType, prefix)
+
+ def get_subnets(self):
+ """
+ The name is suggested to match the NB API.
+ list all subnets and their properties.
+ """
+ return super(self.__class__, self).get_entries('subnets')
+
+ def add_subnet_gateway(self, name, body):
+ """
+ Add a subnet gateway.
+ """
+ super(self.__class__, self).add_entry('subnet', name, body)
+
+ def remove_subnet_gateway(self, name):
+ """
+ Remove a subnet gateway.
+ """
+ super(self.__class__, self).remove_entry('subnet', name)
+
+ def test_subnet_operations(self, name, body):
+ """
+ Test subnet operations, like adding and removeing a subnet.
+ >>> ArpHandler().test_subnet_operations('test',{'name':'test','subnet':'10.0.0.254/8'})
+ True
+ """
+ return super(self.__class__, self).test_add_remove_operations('subnets', 'subnet', name, body, 'subnetConfig')
+
+
+if __name__ == '__main__':
+ print 'arp handler'
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-01
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class ContainerManager(TestModule):
+ """
+ Test for the container manager.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+
+ def __init__(self, restSubContext='/controller/nb/v2/containermanager', user=DEFAULT_USER, password=DEFAULT_PWD,
+ container=None, contentType='json', prefix=DEFAULT_PREFIX):
+ super(self.__class__, self).__init__(restSubContext, user, password, container, contentType, prefix)
+
+ def get_containers(self):
+ """
+ The name is suggested to match the NB API.
+ Show the containers
+ """
+ return super(self.__class__, self).get_entries('containers')
+
+ def add_container(self, name, body):
+ """
+ Add a container
+ """
+ self.container = 'container'
+ super(self.__class__, self).add_entry('containermanager', name, body)
+
+ def remove_container(self, name):
+ """
+ Remove a container
+ """
+ self.container = 'container'
+ super(self.__class__, self).remove_entry('containermanager', name)
+
+ def test_container_operations(self, name, body):
+ """
+ Test subnet operations, like adding and removeing a subnet.
+ >>> ContainerManager().test_container_operations('cont1',{'container':'cont1','flowSpecs': [], 'staticVlan':'10','nodeConnectors':["OF|1@OF|00:00:00:00:00:00:00:01","OF|23@OF|00:00:00:00:00:00:20:21"]})
+ True
+ """
+ return super(self.__class__, self).test_add_remove_operations('containers', 'container', name, body,
+ 'container-config')
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-01
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class ForwardingManager(TestModule):
+ """
+ Test for the forwarding manager.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+ def __init__(self,restSubContext='/controller/nb/v2/staticroute',user=DEFAULT_USER, password=DEFAULT_PWD,container=DEFAULT_CONTAINER,contentType='json',prefix=DEFAULT_PREFIX):
+ super(self.__class__,self).__init__(restSubContext,user,password,container,contentType,prefix)
+
+ def get_routes(self):
+ """
+ The name is suggested to match the NB API.
+ list all routes
+ """
+ return super(self.__class__, self).get_entries('routes')
+
+ def add_static_route(self, name, body):
+ """
+ Add a static route.
+ """
+ r = super(self.__class__, self).add_entry('route', name, body)
+
+ def remove_static_route(self, name):
+ """
+ Remove a static route
+ """
+ r = super(self.__class__, self).remove_entry('route', name)
+
+ def test_static_route_operations(self, name, body):
+ """
+ Test static route operations, like adding and removeing a route.
+ >>> ForwardingManager().test_static_route_operations('route1',{'name':'route1','prefix':'192.168.1.0/24','nextHop':'10.0.0.2'})
+ True
+ """
+ return super(self.__class__, self).test_add_remove_operations('routes', 'route', name, body, 'staticRoute')
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-05
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class ForwardingRuleManager(TestModule):
+ """
+ Test for the forwarding rule manager.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+
+ def __init__(self, restSubContext='/controller/nb/v2/flowprogrammer', user=DEFAULT_USER, password=DEFAULT_PWD,
+ container=DEFAULT_CONTAINER, contentType='json', prefix=DEFAULT_PREFIX):
+ super(self.__class__, self).__init__(restSubContext, user, password, container, contentType, prefix)
+
+ def get_flows(self):
+ """
+ The name is suggested to match the NB API.
+ Show the flows
+ """
+ return super(self.__class__, self).get_entries('')
+
+ def add_flow_to_node(self, node_type, node_id, name, body):
+ suffix = 'node/' + node_type + '/' + node_id + '/staticFlow'
+ r = super(self.__class__, self).add_entry(suffix, name, body)
+
+ def remove_flow_from_node(self, node_type, node_id, name):
+ suffix = 'node/' + node_type + '/' + node_id + '/staticFlow'
+ r = super(self.__class__, self).remove_entry(suffix, name)
+
+ def test_flow_operations(self, node_type, node_id, name, body):
+ """
+ Test the add,remove,show actions on flows.
+ >>> body = {'installInHw':'true','name':'flow1','node':{'id':'00:00:00:00:00:00:00:02','type':'OF'},'priority':'1','etherType':'0x800','nwDst':'10.0.0.1/32','actions':['OUTPUT=1']}
+ >>> ForwardingRuleManager().test_flow_operations('OF','00:00:00:00:00:00:00:02','flow1',body)
+ True
+ >>> body = {'installInHw':'true','name':'flow2','node':{'id':'00:00:00:00:00:00:00:02','type':'OF'},'priority':'1','etherType':'0x800','nwDst':'10.0.0.2/32','actions':['OUTPUT=2']}
+ >>> ForwardingRuleManager().test_flow_operations('OF','00:00:00:00:00:00:00:02','flow2',body)
+ True
+ """
+ result = []
+ #current flow table should be empty.
+ r = self.get_flows()
+ result.append(body not in r['flowConfig'])
+ #Add a flow
+ self.add_flow_to_node(node_type, node_id, name, body)
+ r = self.get_flows()
+ result.append(body in r['flowConfig'])
+ #Remove the flow and test if succeed
+ if result == [True, True]:
+ self.remove_flow_from_node(node_type, node_id, name)
+ r = self.get_flows()
+ result.append(body not in r['flowConfig'])
+ return result == [True, True, True]
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-06
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class HostTracker(TestModule):
+ """
+ Test for the host tracker..
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+ def __init__(self,restSubContext='/controller/nb/v2/hosttracker',user=DEFAULT_USER, password=DEFAULT_PWD,container=DEFAULT_CONTAINER,contentType='json',prefix=DEFAULT_PREFIX):
+ super(self.__class__,self).__init__(restSubContext,user,password,container,contentType,prefix)
+
+ def get_hosts(self):
+ """
+ The name is suggested to match the NB API.
+ list all active hosts, should be done after using h1 ping h2 in mininet
+ """
+ return super(self.__class__, self).get_entries(['hosts/active', 'hosts/inactive'], 'hostConfig')
+
+ def add_host(self, name, body):
+ """
+ Add a host.
+ """
+ r = super(self.__class__, self).add_entry('address', name, body)
+
+ def remove_host(self, name):
+ """
+ Remove a host.
+ """
+ r = super(self.__class__, self).remove_entry('address', name)
+
+ def test_host_operations(self, name, body):
+ """
+ Test host operations, like adding and removing.
+ >>> HostTracker().test_host_operations('10.0.1.4',{'nodeType': 'OF', 'dataLayerAddress': '5e:bf:79:84:10:a6', 'vlan': '1', 'nodeId': '00:00:00:00:00:00:00:03', 'nodeConnectorId': '9', 'networkAddress': '10.0.1.4', 'staticHost': True, 'nodeConnectorType': 'OF'})
+ True
+ """
+ return super(self.__class__, self).test_add_remove_operations(['hosts/active', 'hosts/inactive'], 'address',
+ name, body,
+ 'hostConfig')
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-01
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class StatisticsManager(TestModule):
+ """
+ Test for the statistics manager.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+
+ def __init__(self, restSubContext='/controller/nb/v2/statistics', user=DEFAULT_USER, password=DEFAULT_PWD,
+ container=DEFAULT_CONTAINER, contentType='json', prefix=DEFAULT_PREFIX):
+ super(self.__class__, self).__init__(restSubContext, user, password, container, contentType, prefix)
+
+ def get_flow_stats(self):
+ """
+ The name is suggested to match the NB API.
+ Show the flow statistics
+ """
+ return super(self.__class__, self).get_entries('flow')
+
+ def get_port_stats(self):
+ """
+ The name is suggested to match the NB API.
+ Show the port statistics
+ """
+ return super(self.__class__, self).get_entries('port')
+
+ def get_table_stats(self):
+ """
+ The name is suggested to match the NB API.
+ Show the table statistics
+ """
+ return super(self.__class__, self).get_entries('table')
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-01
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class SwitchManager(TestModule):
+ """
+ Test for the switch manager, including read switch nodes.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+
+ def __init__(self, restSubContext='/controller/nb/v2/switchmanager', user=DEFAULT_USER, password=DEFAULT_PWD,
+ container=DEFAULT_CONTAINER, contentType='json', prefix=DEFAULT_PREFIX):
+ super(self.__class__, self).__init__(restSubContext, user, password, container, contentType, prefix)
+
+ def get_nodes(self):
+ """
+ The name is suggested to match the NB API.
+ list all nodes and their properties
+ """
+ suffix = 'nodes'
+ r = super(self.__class__, self).read(suffix)
+ if r:
+ return r
+
+ def get_node(self, suffix):
+ """
+ The name is suggested to match the NB API.
+ list nodeconnector and properties of a node.
+ """
+ r = super(self.__class__, self).read(suffix)
+ if r:
+ return r
+
+ def add_property_to_node(self, node_type, node_id, property, value):
+ """
+ Add a property to given node.
+ """
+ suffix = 'node/' + node_type + '/' + node_id + '/property'
+ r = super(self.__class__, self).update(suffix + '/' + property + '/' + str(value))
+
+ def remove_property_from_node(self, node_type, node_id, property):
+ """
+ Remove a property from given node.
+ """
+ suffix = 'node/' + node_type + '/' + node_id + '/property'
+ r = super(self.__class__, self).delete(suffix + '/' + property)
+
+ def add_property_to_nodeconnector(self, node_type, node_id, nc_type, nc_id, property, value):
+ """
+ Add a property to given node.
+ """
+ suffix = 'nodeconnector/' + node_type + '/' + node_id + '/' + nc_type + '/' + nc_id + '/property'
+ r = super(self.__class__, self).update(suffix + '/' + property + '/' + str(value))
+
+ def remove_property_from_nodeconnector(self, node_type, node_id, nc_type, nc_id, property):
+ """
+ Add a property to given node.
+ """
+ suffix = 'nodeconnector/' + node_type + '/' + node_id + '/' + nc_type + '/' + nc_id + '/property'
+ r = super(self.__class__, self).delete(suffix + '/' + property)
+
+ def test_list_nodes(self):
+ """
+ The name is suggested to match the NB API.
+ list all nodes and their properties
+ >>> SwitchManager().test_list_nodes()
+ True
+ """
+ result = []
+ r = self.get_nodes()
+ if r:
+ t = [e['node'] for e in r['nodeProperties']]
+ result.append({u'type': u'OF', u'id': u'00:00:00:00:00:00:00:01'} in t)
+ result.append({u'type': u'OF', u'id': u'00:00:00:00:00:00:00:02'} in t)
+ result.append({u'type': u'OF', u'id': u'00:00:00:00:00:00:00:03'} in t)
+ return result == [True, True, True]
+
+ def test_node_property_operations(self, node_type, node_id, property, value):
+ """
+ Test the add,remove,show actions on node properties.
+
+ >>> SwitchManager().test_node_property_operations('OF','00:00:00:00:00:00:00:01','description','Switch1')
+ True
+ >>> SwitchManager().test_node_property_operations('OF','00:00:00:00:00:00:00:02','description','Switch2')
+ True
+ >>> SwitchManager().test_node_property_operations('OF','00:00:00:00:00:00:00:03','description','Switch3')
+ True
+ """
+ result = []
+ #current node properties should not include description
+ r = self.get_nodes()
+ v = [e['properties'].get(property) for e in r['nodeProperties'] if
+ e['node'] == {u'type': node_type, u'id': node_id}]
+ result.append(v == [{u'value': u'None'}] or v == [None])
+ #After adding, current node properties should include description
+ self.add_property_to_node(node_type, node_id, property, value)
+ r = self.get_nodes()
+ v = [e['properties'].get(property) for e in r['nodeProperties'] if
+ e['node'] == {u'type': node_type, u'id': node_id}]
+ result.append(v == [{u'value': value}])
+ #After removing, current node properties should not include description
+ self.remove_property_from_node(node_type, node_id, property)
+ r = self.get_nodes()
+ v = [e['properties'].get(property) for e in r['nodeProperties'] if
+ e['node'] == {u'type': node_type, u'id': node_id}]
+ result.append(v == [{u'value': u'None'}] or v == [None])
+ return result == [True, True, True]
+
+ def test_nodeconnector_property_operations(self, node_type, node_id, nc_type, nc_id, property, value):
+ """
+ Test the add,remove,show actions on nodeconnector properties.
+
+ >>> SwitchManager().test_nodeconnector_property_operations('OF','00:00:00:00:00:00:00:01','OF','1','bandwidth',1000)
+ True
+ """
+ result = []
+ node_suffix = 'node/' + node_type + '/' + node_id
+ #default bw should be 10000000000L
+ r = self.get_node(node_suffix)
+ default_value = [e['properties'][property] for e in r['nodeConnectorProperties'] if
+ property in e['properties'] and e['nodeconnector'] == {
+ u'node': {u'type': node_type, u'id': node_id}, u'type': nc_type, u'id': nc_id}]
+ #After setting, the value should be the value
+ self.add_property_to_nodeconnector(node_type, node_id, nc_type, nc_id, property, value)
+ r = self.get_node(node_suffix)
+ current_value = [e['properties'][property] for e in r['nodeConnectorProperties'] if
+ property in e['properties'] and e['nodeconnector'] == {
+ u'node': {u'type': node_type, u'id': node_id}, u'type': nc_type, u'id': nc_id}]
+ result.append(current_value == [{'value': value}])
+ #After removing, and restoring the default value, the bandwidth property should be default
+ self.remove_property_from_nodeconnector(node_type, node_id, nc_type, nc_id, property)
+ r = self.get_node(node_suffix)
+ v = [e['properties'][property] for e in r['nodeConnectorProperties'] if
+ property in e['properties'] and e['nodeconnector'] == {u'node': {u'type': node_type, u'id': node_id},
+ u'type': nc_type, u'id': nc_id}]
+ result.append(v == [])
+ self.add_property_to_nodeconnector(node_type, node_id, nc_type, nc_id, property, default_value[0]['value'])
+ r = self.get_node(node_suffix)
+ current_value = [e['properties'][property] for e in r['nodeConnectorProperties'] if
+ property in e['properties'] and e['nodeconnector'] == {
+ u'node': {u'type': node_type, u'id': node_id}, u'type': nc_type, u'id': nc_id}]
+ result.append(current_value == default_value)
+ return result == [True, True, True]
\ No newline at end of file
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-01
+"""
+
+import sys
+
+sys.path.append('..')
+from restlib import *
+from testmodule import TestModule
+
+sys.path.remove('..')
+
+
+class TopologyManager(TestModule):
+ """
+ Test for the topology manager.
+ Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'
+ """
+
+ def __init__(self, restSubContext='/controller/nb/v2/topology', user=DEFAULT_USER, password=DEFAULT_PWD,
+ container=DEFAULT_CONTAINER, contentType='json', prefix=DEFAULT_PREFIX):
+ super(self.__class__, self).__init__(restSubContext, user, password, container, contentType, prefix)
+
+ def get_topology(self):
+ """
+ The name is suggested to match the NB API.
+ Show the topology
+ >>> TopologyManager().get_topology()
+ True
+ """
+ result = []
+ r = super(self.__class__, self).get_entries()
+ if r:
+ v = [e['edge'] for e in r['edgeProperties']]
+ result.append({u'tailNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:01'},
+ u'type': u'OF', u'id': u'2'},
+ u'headNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:03'},
+ u'type': u'OF', u'id': u'3'}} in v)
+ result.append({u'tailNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:03'},
+ u'type': u'OF', u'id': u'3'},
+ u'headNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:01'},
+ u'type': u'OF', u'id': u'2'}} in v)
+ result.append({u'tailNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:02'},
+ u'type': u'OF', u'id': u'3'},
+ u'headNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:01'},
+ u'type': u'OF', u'id': u'1'}} in v)
+ result.append({u'tailNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:01'},
+ u'type': u'OF', u'id': u'1'},
+ u'headNodeConnector': {u'node': {u'type': u'OF', u'id': u'00:00:00:00:00:00:00:02'},
+ u'type': u'OF', u'id': u'3'}} in v)
+ print result == [True, True, True, True]
+
+ def get_userlinks(self):
+ """
+ The name is suggested to match the NB API.
+ Show the userlinks.
+ """
+ suffix = 'userLinks'
+ r = super(self.__class__, self).read(suffix)
+ if r:
+ return r
+
+ def add_userlink(self, name, body):
+ """
+ Add a userlink.
+ """
+ suffix = 'userLink'
+ r = super(self.__class__, self).update(suffix + '/' + name, body)
+ return r
+
+ def remove_userlink(self, name):
+ """
+ Remove a userlink.
+ """
+ suffix = 'userLink'
+ r = super(self.__class__, self).delete(suffix + '/' + name)
+ return r
+
+ def test_userlink_operations(self, name, body):
+ """
+ Test userlink operations, like adding and removing.
+ >>> TopologyManager().test_userlink_operations('link1', {'status':'Success','name':'link1','srcNodeConnector':'OF|1@OF|00:00:00:00:00:00:00:02','dstNodeConnector':'OF|1@OF|00:00:00:00:00:00:00:03'})
+ True
+ """
+ return super(self.__class__, self).test_add_remove_operations('userLinks', 'userLink', name, body, 'userLinks')
--- /dev/null
+"""
+CSIT test tools.
+Authors: Denghui Huang@IBM, Baohua Yang@IBM
+Updated: 2013-11-06
+"""
+import json
+
+import requests
+
+
+# Global variables
+DEFAULT_CONTROLLER_IP = '127.0.0.1'
+#DEFAULT_CONTROLLER_IP = '9.186.105.113' #just for temp test
+DEFAULT_PORT = '8080'
+DEFAULT_PREFIX = 'http://' + DEFAULT_CONTROLLER_IP + ':' + DEFAULT_PORT
+DEFAULT_CONTAINER = 'default'
+DEFAULT_USER = 'admin'
+DEFAULT_PWD = 'admin'
+CASES_DIR = 'cases'
+TIMEOUTS = 2
+
+'''
+Send a POST request.
+'''
+
+
+def do_post_request(url, content_type, payload=None, user=DEFAULT_USER, password=DEFAULT_PWD):
+ data = payload
+ headers = {}
+ if content_type == 'json':
+ headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
+ if payload != None:
+ data = json.dumps(payload)
+ elif content_type == 'xml':
+ headers = {'Content-type': 'application/xml', 'Accept': 'application/xml'}
+ else:
+ print 'unsupported content-type'
+ try:
+ r = requests.post(url, data, headers=headers, auth=(user, password), timeout=TIMEOUTS)
+ r.raise_for_status()
+ except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
+ return 400
+ else:
+ return r.status_code
+
+
+def do_get_request_with_status_code(url, content_type, user=DEFAULT_USER, password=DEFAULT_PWD):
+ '''
+ Send a GET request.
+ @return The status code.
+ '''
+ r = None
+ try:
+ r = requests.get(url, auth=(user, password), timeout=TIMEOUTS)
+ r.raise_for_status()
+ except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
+ print e
+ return r.status_code
+ finally:
+ return r.status_code
+
+
+def do_put_request(url, content_type, payload=None, user=DEFAULT_USER, password=DEFAULT_PWD):
+ '''
+ Send a PUT request.
+ @return The status code.
+ '''
+ data = payload
+ headers = {}
+ if content_type == 'json':
+ headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
+ if payload != None:
+ data = json.dumps(payload)
+ elif content_type == 'xml':
+ headers = {'Content-type': 'application/xml', 'Accept': 'application/xml'}
+ else:
+ print 'unsupported content-type'
+ try:
+ r = requests.put(url, data, headers=headers, auth=(user, password), timeout=TIMEOUTS)
+ r.raise_for_status()
+ except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
+ return 400
+ else:
+ return r.status_code
+
+
+def do_delete_request(url, user=DEFAULT_USER, password=DEFAULT_PWD):
+ '''
+ Send a DELETE request.
+ @return The status code.
+ '''
+ r = None
+ try:
+ r = requests.delete(url, auth=(user, password), timeout=TIMEOUTS)
+ r.raise_for_status()
+ except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
+ print e
+ finally:
+ if r:
+ return r.status_code
+
+
+def convert_result_to_list(result):
+ '''
+ Convert the result content to list.
+ '''
+ list2 = []
+ #print result
+ content = result.values()
+ for list1 in content:
+ list2 = [dict1.values() for dict1 in list1]
+ #print list2
+ list3 = []
+ for list4 in list2:
+ for element in list4:
+ list3.append(element)
+ #print list3
+ return list3
+
+
+def do_get_request_with_response_content(url, content_type, user=DEFAULT_USER, password=DEFAULT_PWD,
+ convert_to_list=False):
+ '''
+ Send a GET request and get the response.
+ @return response content as list.
+ '''
+ try:
+ r = requests.get(url, auth=(user, password), timeout=TIMEOUTS)
+ r.raise_for_status()
+ except (requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
+ print e
+ return None
+ else:
+ if r != None:
+ if content_type == 'json':
+ content = r.json()
+ return convert_result_to_list(content) if convert_to_list else content
+ elif content_type == 'xml':#TODO: add parser to xml
+ return None
+
+
+if __name__ == '__main__':
+ #example
+ #Note: in json body, all field name and value (if it is string type) must be enclosed in double quotes.
+ #This constraint maybe cause by json parser.
+ body = {"status": "Success", "dstNodeConnector": "OF|1@OF|00:00:00:00:00:00:00:01", "name": "link3",
+ "srcNodeConnector": "OF|1@OF|00:00:00:00:00:00:00:03"}
+ url = 'http://127.0.0.1:8080/controller/nb/v2/topology/default/userLink/link3'
+ content_type = 'json'
+ print do_put_request(url, content_type, body)
+
--- /dev/null
+#!/usr/bin/env python
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-11-07
+
+Usage: Before running the test tool, should
+ 1. Start 2-layer tree topology network. e.g., in Mininet, run 'sudo mn --controller=remote,ip=127.0.0.1 --mac --topo tree,2'.
+ 2. Configure gateway in the controller web GUI, name = 'gateway', subnet = '10.0.0.254/24'.
+ 3. In Mininet, run 'h1 ping h2' to make sure the network is connected.
+"""
+import doctest
+import os
+from restlib import *
+
+
+def test_case(module_name):
+ '''
+ Run single test on given module.
+ '''
+ print "#Test case: " + module_name.replace('_', ' ')
+ cmd = 'python -m doctest ' + module_name + '.py'
+ os.system(cmd)
+
+def run(modules=None):
+ '''
+ Run test cases according to the given modules.
+ If no parameter is given, then will scan the case directory,
+ and try to run all cases.
+ '''
+ backup_dir = os.getcwd()
+ if not modules:
+ modules = [e[:-3] for e in os.listdir(CASES_DIR) if e.endswith('.py')]
+ os.chdir(backup_dir + '/' + CASES_DIR)
+ for name in modules:
+ test_case(name)
+ os.chdir(backup_dir)
+
+
+if __name__ == '__main__':
+ doctest.testmod()
+ test_modules = ['switch_manager', 'topology_manager', 'forwarding_rule_manager', 'statistics_manager',
+ 'host_tracker', 'arp_handler', 'forwarding_manager', 'container_manager']
+ #test_modules = ['topology_manager']
+ run(test_modules)
+ #run()
--- /dev/null
+"""
+CSIT test tools.
+Authors: Baohua Yang@IBM, Denghui Huang@IBM
+Updated: 2013-10-30
+"""
+
+from restlib import *
+
+
+class TestModule(object):
+ """
+ Basic module class for test restful APIS.
+ Support the standard Create, Read, Update, Delete (CRUD) actions.
+ """
+
+ def __init__(self, restSubContext, user=DEFAULT_USER, password=DEFAULT_PWD, container=DEFAULT_CONTAINER,
+ contentType='json', prefix=DEFAULT_PREFIX):
+ self.restSubContext = restSubContext
+ self.container = container
+ self.user = user
+ self.password = password
+ self.contentType = contentType
+ self.prefix = prefix
+
+ def get_entries(self, suffix=None, key=None):
+ """
+ Get the existed entries in the service.
+ """
+ if isinstance(suffix, list) and key:
+ result = {}
+ result[key] = []
+ for s in suffix:
+ result[key].extend(self.get_entries(s).get(key))
+ return result
+ elif isinstance(suffix, str):
+ return self.read(suffix)
+ elif not suffix:
+ return self.read()
+ else:
+ return None
+
+ def add_entry(self, suffix, name, body):
+ """
+ Add entry to the service.
+ """
+ self.update(suffix + '/' + name, body)
+
+ def remove_entry(self, suffix, name):
+ """
+ Remove entry from the service.
+ """
+ self.delete(suffix + '/' + name)
+
+ def test_add_remove_operations(self, suffix_entries, suffix_entry, name, body, key):
+ result = []
+ #Add an entry
+ self.add_entry(suffix_entry, name, body)
+ r = self.get_entries(suffix_entries, key)
+ if r:
+ v = r.get(key)
+ result.append(body in v if v else False)
+ #Remove the added entry
+ if result == [True]:
+ self.remove_entry(suffix_entry, name)
+ r = self.get_entries(suffix_entries, key)
+ v = r.get(key)
+ result.append(body not in v if v else True)
+ return result == [True, True]
+
+ def create(self, suffix, body=None):
+ """
+ POST to given suffix url.
+ TODO: complete
+ """
+ url = self.prefix + self.restSubContext
+ if self.container:
+ url += '/' + self.container
+ if suffix:
+ url += '/' + suffix
+ return do_post_request(url, self.contentType, body, self.user, self.password)
+
+ def read(self, suffix=None):
+ """
+ GET from given suffix url.
+ """
+ url = self.prefix + self.restSubContext
+ if self.container:
+ url += '/' + self.container
+ if suffix:
+ url += '/' + suffix
+ return do_get_request_with_response_content(url, self.contentType, self.user, self.password)
+
+ def update(self, suffix, body=None):
+ """
+ PUT to given suffix url.
+ """
+ url = self.prefix + self.restSubContext
+ if self.container:
+ url += '/' + self.container
+ if suffix:
+ url += '/' + suffix
+ return do_put_request(url, self.contentType, body, self.user, self.password)
+
+ def delete(self, suffix):
+ """
+ DELETE to given suffix url.
+ TODO: complete
+ """
+ url = self.prefix + self.restSubContext
+ if self.container:
+ url += '/' + self.container
+ if suffix:
+ url += '/' + suffix
+ return do_delete_request(url, self.user, self.password)
Code Review / integration / test.git / commitdiff