-welcome to open vstf(Virtual Switch Test Framework
-)
+Bottleneck testsuite VSTF
+=========================
+welcome to open vstf(Virtual Switch Test Framework)
},
"body": {
"from": [
- "hangzhou_vstf_ci@huawei.com"
],
"to": [
- "Yiting.Li@gerrit.opnfv.org",
],
- "cc": [ ],
- "bcc": [],
- "subject": "Virtual Switching Performance Test Report"
+ "cc": [],
+ "bcc": []
}
}
#!/bin/bash
+##############################################################################
+# Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
SERVER=$2
PORT=$3
[metadata]
name = vstf
-version = 2014.2.3
-summary = Open Virtual Switch Test Framework
+version = 2015.12
+summary = Virtual Switch Test Framework
description-file =
README.rst
-author = OpenNFV
-author-email = openatf@huawei.com
-home-page = https://12.234.32.89/mediawiki/
+author = BottleneckGroup
+home-page = https://wiki.opnfv.org/bottlenecks
classifier =
- Environment :: OpenATF
+ Environment :: VSTF
Intended Audience :: Information Technology
Intended Audience :: System Administrators
License :: OSI Approved :: Apache Software License
vstf-agent = vstf.agent.agent:main
vstfadm = vstf.controller.vstfadm:main
vstf-settings = vstf.controller.settings.settings_input:main
- vstf-driver = vstf.controller.functiontest.driver.drivertest:main
- vstf-switch = vstf.controller.functiontest.switch.switchtest:main
agent.plugins =
soft = vstf.agent.softagent:softAgent
#!/bin/bash
+##############################################################################
+# Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
function usage()
{
+++ /dev/null
-Modules in the directory are for "Environment Building".
-"Environment Building" is for creating "virtual network" for Performance Evaluation.
-The builder.py contains the quick test code for creates a "virtual network" from a config file.
-
-Usage:
- python builder.py --config /etc/vstf/env/Tn.json
-
-Please modify example config file "/etc/vstf/model/Tn.json" to suit your own Environment.
-
+++ /dev/null
-the modules in this directory include some basic common purpose functions that can be used for other plugins.
-you can see the modules in this directory as libraries that you can import for ease of creating your own plugin.
-
-the main functions of the modules are listed as follows:
-
-ImageManger:
- role:
- virtual machine images manager. used by VmManager.
- wraps 'qemu-img' command as underlying mechanism.
- features:
- provides a function to create child image from a parent image.
- provides a function to delete child image.
-
-SourceManager:
- role:
- source code manager.
- using 'pramiko' as underlying mechanism.
- features:
- provides a function to download source code
- provides a function which wrap 'make clean;make;make install' to compile and install.
-
-VmManager:
- role:
- virtual machine manager which can create vm and configure vm on bootstrap.
- use libvirt as underlying mechanism. wrap 'virsh' command for creating vm.
- features:
- provides a function to create vm from a vm configuration context.
- provides functions to detect if vm boots up successfully and onfigure bootstrap options for vm like ip and rabbitmq conf file.
-
-vm9pfs:
- used by VmManager.
- host side 9pfs operation (mainly file operations) to communicate with vm by libvirt 9pfs.
- see as communication protocol on client side. The vm runs a agent for receiving the commands.
-
-VmXmlHelp:
- used by VmManager.
- just some divided libvirt xml string templates for building a complete libvirt xml file on demand by VmManager.
-
-
+++ /dev/null
-FSMonitor is a daemon process which runs in the vm.
-FSMonitor receive "commands" from 'VmManager' by libvirt 9pfs.
-
-Basically the process works like this:
- 1.The 'VmManager' writes 'command string' to 'command' file on the libvirt 9pfs used by vm.
- 2.The FSMonitor constantly detects file changes on the libvirt 9pfs.
- 3.The FSMonitor finds the newly created file, it then reads the file and execute the command.
-
-All the dependencies of FSMonitor should be satisfied by modules in this directory.
-
-When deploying FSMonitor, just copy this directory into the vm.
-
-
+++ /dev/null
-All the plugins should subclass EnvBuilderPlugin from "model.py".
-
-The EnvBuilderPlugin is a template class with a template algorithm:
-
- def __init__(self, ):
- pass
- @abstractmethod
- def clean(self):
- #clean Environment before goes further.
- @abstractmethod
- def install(self):
- #install network virtualization software from source code.
- @abstractmethod
- def load_drivers(self):
- #loads drivers for network card.
- @abstractmethod
- def create_brs(self):
- #creates virtual switches.
- @abstractmethod
- def config_br_ports(self):
- #config the vlan property for vswitch ports.
- def create_vms(self):
- #create vms
- def wait_vms(self):
- #wait vm to boot up and config vm for ips and other configurations.
- def check_vm_connectivity(self):
- #check if the vms correctly setup the control panel ips.
- def build(self, cfg_intent):
- self.host_cfg = cfg_intent #please retrieve options from self.host_cfg for your use in other methods.
- self.clean()
- self.download_and_compile()
- self.load_drivers()
- self.create_brs()
- self.create_vms()
- self.wait_vms()
- self.config_tap_vlans()
- self.check_vm_connectivity()
-
-You should implements the abstract methods left empty, however you can make some methods do nothing to skip steps..
-
-The plugin receives a "cfg_intent", The "cfg_intent" is a python dict parsed from a env-build configuration file.
-
-It contains the detail configurations for the plugin to build a "virtual network" for testing.
-
-There are some example json config files for building different type of "virtual network" under "etc/vstf/env" that you can refer to.
-
-Before you creates a new plugin, you should make sure you understand these json config file properly.
-
-
+++ /dev/null
-equalizer of the performance of network
-#!/usr/bin/python
+##############################################################################
+# Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+
+
import time
from spirent import stcPython
+++ /dev/null
-# Copyright Huawei Technologies Co., Ltd. 1998-2015.
-# All Rights Reserved.
-#
-# 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.
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-
-eth_for_test = [
- "eth4",
- "p57p2"
-]
-
-mac_of_eth = [
- "90:e2:ba:20:1f:d8",
- '90:e2:ba:20:1f:d9'
-]
-
-bdf_of_eth = [
- "04:00.0",
- "04:00.1"
-]
-
-source_repo = {
- "vnx-bin":{
- "install": False,
- "url": "root@192.168.188.10:/root/src/vnx-bin",
- "repo_type": "git"
- }
-}
\ No newline at end of file
+++ /dev/null
-To enable Env unittest, please make sure:
-
- 1. br0 exists on TargetHost and connects with 'management network'.
- 2. a vm image exists on Target Host.
- 3. change options in configuration files in the "configuration" directory.
- 4. copy public key to git server, for example:
- ssh-copy-id root@192.168.188.10
- note:
- 192.168.188.10 is the server you config in "../configuration.py"
- this is needed for source manager which use 'git' to pull down sources.
-
-run tests:
- python run_test.py
\ No newline at end of file
+++ /dev/null
-# Copyright Huawei Technologies Co., Ltd. 1998-2015.
-# All Rights Reserved.
-#
-# 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.
+++ /dev/null
-{
- "scenario": "ti",
- "env-build": [
- {
- "ip": "192.168.188.16",
- "drivers": [
- "ixgbe",
- "vhost_net"
- ],
- "vms": [
- {
- "vm_cpu": 3,
- "vm_name": "test1",
- "vm_memory": 4194304,
- "image_path": "/mnt/sdb/test1.qcow2",
- "image_type": "qcow2",
- "init_config": {
- "amqp_server": "192.168.188.10",
- "ctrl_ip_setting": "192.168.188.200/23",
- "ctrl_gw": "192.168.188.1"
- },
- "taps":[
- {
- "tap_name": "tap_in",
- "br_type": "ovs",
- "br_name": "ovs1",
- "tap_mac": "56:6f:44:a5:3f:a2",
- "vlan_mode": "access",
- "vlan_id": "100"
- }
- ],
- "ctrl_mac": "16:6f:44:a5:3f:a2",
- "ctrl_br": "br0"
- }
- ],
- "bridges": [
- {
- "type": "ovs",
- "name": "ovs1",
- "uplinks": [
- {
- "bdf": "04:00.0",
- "vlan_mode": "trunk",
- "vlan_id": "100,200,300,400"
- }
- ],
- "vtep":{},
- "fastlink":[
- {
- "inport":"04: 00.0",
- "outport":"tap_in"
- }
- ]
- }
- ]
- }
- ]
-}
\ No newline at end of file
+++ /dev/null
-{
- "env-build": [
- {
- "ip": "192.168.188.16",
- "drivers": [
- "ixgbe"
- ],
- "bridges": [
- {
- "type": "ovs",
- "name": "ovs1",
- "uplinks": [
- {
- "bdf": "04:00.0",
- "vlan_mode": "trunk",
- "vlan_id": "1,100,200"
- },
- {
- "bdf": "04:00.1",
- "vlan_mode": "trunk",
- "vlan_id": "1,100,200"
- }
- ],
- "vtep": {},
- "fastlink": [
- {
- "inport": "04: 00.0",
- "outport": "04: 00.1"
- }
- ]
- }
- ],
- "vms": []
- }
- ]
-}
+++ /dev/null
-{
- "scenario": "tn",
- "env-build": [
- {
- "ip": "192.168.188.16",
- "drivers": [
- "ixgbe"
- ],
- "vms": [],
- "bridges": [
- {
- "type": "ovs",
- "name": "ovs1",
- "uplinks": [
- {
- "bdf": "04:00.0",
- "vlan_mode": "trunk",
- "vlan_id": "1,100,200"
- },
- {
- "bdf": "04:00.1",
- "vlan_mode": "trunk",
- "vlan_id": "1,100,200"
- }
- ],
- "vtep": {},
- "fastlink": [
- {
- "inport": "04:00.0",
- "outport": "04:00.1"
- }
- ]
- }
- ]
- }
- ]
-}
\ No newline at end of file
+++ /dev/null
-{
- "scenario": "tnv",
- "env-build": [
- {
- "ip": "192.168.188.16",
- "drivers": [
- "ixgbe",
- "tap_vhost"
- ],
- "vms": [
- {
- "vm_cpu": 3,
- "vm_name": "test1",
- "vm_memory": 4194304,
- "image_path": "/mnt/sdb/test1.qcow2",
- "image_type": "qcow2",
- "init_config": {
- "amqp_server": "192.168.188.10",
- "ctrl_ip_setting": "192.168.188.200/23",
- "tap_pktloop_config": "dpdk",
- "ctrl_gw": "192.168.188.1"
- },
- "taps": [
- {
- "tap_name": "tap_in",
- "br_type": "ovs",
- "br_name": "ovs1",
- "tap_mac": "56:6f:44:a5:3f:a2",
- "vlan_mode": "access",
- "vlan_id": "100"
- },
- {
- "tap_name": "tap_out",
- "br_type": "ovs",
- "br_name": "ovs22",
- "tap_mac": "56:6f:44:a5:3f:a3",
- "vlan_mode": "access",
- "vlan_id": "200"
- }
- ],
- "ctrl_mac": "16:6f:44:a5:3f:a2",
- "ctrl_br": "br0"
- }
- ],
- "bridges": [
- {
- "type": "ovs",
- "name": "ovs1",
- "uplinks": [
- {
- "bdf": "04: 00.0",
- "vlan_mode": "trunk",
- "vlan_id": "100"
- }
- ],
- "vtep":{},
- "fastlink":[
- {
- "inport":"04: 00.0",
- "outport":"tap_in"
- }
- ]
- },
- {
- "type": "ovs",
- "name": "ovs2",
- "uplinks": [
- {
- "bdf": "04: 00.1",
- "vlan_mode": "trunk",
- "vlan_id": "200"
- }
- ],
- "vtep":{},
- "fastlink":[
- {
- "inport":"04: 00.1",
- "outport":"tap_out"
- }
- ]
- }
- ]
- }
- ]
-}
\ No newline at end of file
+++ /dev/null
-{
- "scenario": "tu",
- "env-build": [
- {
- "ip": "192.168.188.16",
- "drivers": [
- "tap_vhost"
- ],
- "vms": [
- {
- "vm_cpu": 3,
- "vm_name": "test1",
- "vm_memory": 4194304,
- "image_path": "/mnt/sdb/test1.qcow2",
- "image_type": "qcow2",
- "init_config": {
- "amqp_server": "192.168.188.10",
- "ctrl_ip_setting": "192.168.188.200/23",
- "ctrl_gw": "192.168.188.1"
- },
- "taps": [
- {
- "tap_name": "tap1",
- "br_type": "ovs",
- "br_name": "ovs1",
- "tap_mac": "56:6f:44:a5:3f:a2",
- "vlan_mode": "access",
- "vlan_id": "100"
- }
- ]
- },
- {
- "vm_cpu": 3,
- "vm_name": "test2",
- "vm_memory": 4194304,
- "image_path": "/mnt/sdb/test2.qcow2",
- "image_type": "qcow2",
- "init_config": {
- "amqp_server": "192.168.188.10",
- "ctrl_ip_setting": "192.168.188.201/23",
- "ctrl_gw": "192.168.188.1"
- },
- "taps": [
- {
- "tap_name": "tap2",
- "br_type": "ovs",
- "br_name": "ovs1",
- "tap_mac": "56:6f:44:a5:3f:a3",
- "vlan_mode": "access",
- "vlan_id": "100"
- }
- ]
- }
- ],
- "bridges": [
- {
- "type": "ovs",
- "name": "ovs1",
- "uplinks": [],
- "vtep": {},
- "fastlink": [
- {
- "inport": "tap1",
- "outport": "tap2"
- }
- ]
- }
- ]
- }
- ]
-}
\ No newline at end of file
+++ /dev/null
-"""
-Created on 2015-10-9
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest import configuration
-
-
-class Test(unittest.TestCase):
- def setUp(self):
- self.eth_for_test = configuration.eth_for_test
- self.mac_of_eth = configuration.mac_of_eth
- self.source_repo = configuration.source_repo
- self.bdf_of_eth = configuration.bdf_of_eth
-
- def tearDown(self):
- pass
-
-
-if __name__ == "__main__":
- # import sys;sys.argv = ['', 'Test.testName']
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-import importlib
-
-test_order_list = [
- "vstf.agent.unittest.env.test_origin_driver",
- "vstf.agent.unittest.env.test_bridge_plugin",
- "vstf.agent.unittest.env.test_drivermanager",
- "vstf.agent.unittest.env.test_devicemanager",
- "vstf.agent.unittest.env.test_vs_plugin_manager",
- "vstf.agent.unittest.env.test_builder",
- "vstf.agent.unittest.env.test_sourcemanager",
-]
-
-
-def main():
- import logging
- logging.getLogger(__name__)
- logging.basicConfig(level=logging.INFO)
- suite = unittest.TestSuite()
- for mod_name in test_order_list:
- mod = importlib.import_module(mod_name)
- suit = unittest.TestLoader().loadTestsFromModule(mod)
- suite.addTest(suit)
- unittest.TextTestRunner().run(suite)
-
-
-if __name__ == '__main__':
- main()
+++ /dev/null
-"""
-Created on 2015-10-12
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.common.utils import check_call
-from vstf.agent.unittest.env import model
-from vstf.agent.env.vswitch_plugins import bridge_plugin
-from vstf.agent.env.driver_plugins import manager
-
-
-class Test(model.Test):
- def setUp(self):
- super(Test, self).setUp()
- self.plugin = bridge_plugin.BridgePlugin()
- self.dr_mgr = manager.DriverPluginManager()
- self.br_cfg = {
- "name": "br1",
- "uplinks": [
- {
- "bdf": self.bdf_of_eth[0],
- },
- {
- "bdf": self.bdf_of_eth[1],
- }
- ]
- }
- self.dr_mgr.clean()
- self.dr_mgr.load(['ixgbe'])
-
- def tearDown(self):
- super(Test, self).tearDown()
-
- def _check_br_exists(self, name):
- try:
- check_call('ifconfig %s' % name, shell=True)
- except Exception, e:
- return False
- return True
-
- def test_create_br(self):
- self.plugin.clean()
- self.plugin.create_br(self.br_cfg)
- self.assertTrue(self._check_br_exists(self.br_cfg['name']))
- self.plugin.clean()
- self.assertFalse(self._check_br_exists(self.br_cfg['name']))
-
-
-if __name__ == "__main__":
- import logging
-
- logging.basicConfig(level=logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-import os
-
-from vstf.controller.env_build.cfg_intent_parse import IntentParser
-from vstf.agent.env import builder
-
-
-class Test(unittest.TestCase):
- def setUp(self):
- self.mgr = builder.PluginManager()
- self.dir = os.path.dirname(__file__)
-
- def tearDown(self):
- self.mgr.clean()
-
- def __build(self, filepath, drivers=None):
- parser = IntentParser(filepath)
- cfg_intent = parser.parse_cfg_file()
- host_cfg = cfg_intent['env-build'][0]
- print filepath
- print host_cfg
- host_cfg["src-install"] = {}
- if drivers:
- host_cfg['drivers'] = drivers
- return self.mgr.build(host_cfg)
-
- def test_build_tn_using_origin_driver(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Tn.json'), drivers=['ixgbe'])
- self.assertTrue(ret, "test_build_tn_using_origin_driver failed, ret = %s" % ret)
-
- def test_build_tn1v_using_origin_driver(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Tn1v.json'), drivers=['ixgbe', 'vhost_net'])
- self.assertTrue(ret, "test_build_tn1v_using_origin_driver failed, ret = %s" % ret)
-
- def test_build_ti_using_origin_driver(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Ti.json'), drivers=['ixgbe', 'vhost_net'])
- self.assertTrue(ret, "test_build_ti_using_origin_driver failed, ret = %s" % ret)
-
- def test_build_tu_using_origin_driver(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Tu.json'), drivers=['vhost_net'])
- self.assertTrue(ret, "test_build_ti_using_origin_driver failed, ret = %s" % ret)
-
- @unittest.skip('can be tested by tn1v')
- def test_build_tn(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Tn.json'))
- self.assertTrue(ret, "test_build_tn failed, ret = %s" % ret)
-
- @unittest.skip('can be tested by tn1v')
- def test_build_tn1v(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Tn1v.json'))
- self.assertTrue(ret, "test_build_tn failed,ret = %s" % ret)
-
- @unittest.skip('can be tested by tn1v')
- def test_build_ti(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Ti.json'))
- self.assertTrue(ret, "test_build_tn failed, ret = %s" % ret)
-
- @unittest.skip('can be tested by tn1v')
- def test_build_tu(self):
- ret = self.__build(os.path.join(self.dir, 'configuration/Tu.json'))
- self.assertTrue(ret, "test_build_tn failed, ret = %s" % ret)
-
-
-if __name__ == "__main__":
- import logging
-
- LOG = logging.getLogger(__name__)
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-25
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest.env import model
-from vstf.agent.env.basic.device_manager import DeviceManager
-
-
-class Test(model.Test):
- def setUp(self):
- super(Test, self).setUp()
- self.dm = DeviceManager()
- self.device_list = self.dm.list_nic_devices()
- self.device_detail = self.device_list[0]
-
- def tearDown(self):
- super(Test, self).tearDown()
-
- def test_get_device_detail(self):
- detail1 = self.dm.get_device_detail(self.device_detail['bdf'])
- detail2 = self.dm.get_device_detail(self.device_detail['mac'])
- detail3 = self.dm.get_device_detail(self.device_detail['device'])
- self.assertTrue(detail1 == detail2 == detail3 == self.device_detail)
-
- def test_list_nic_devices(self):
- import json
- print json.dumps(self.device_list, indent=4)
-
-
-if __name__ == "__main__":
- import logging
-
- logging.basicConfig(level=logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-10-9
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.common.utils import check_output
-from vstf.agent.unittest.env import model
-from vstf.agent.env.driver_plugins import manager
-
-
-class Test(model.Test):
- def setUp(self):
- super(Test, self).setUp()
- self.driver_mgr = manager.DriverPluginManager()
-
- def tearDown(self):
- super(Test, self).tearDown()
-
- def _driver_exists(self, drivers=[]):
- all_drivers = check_output("lsmod | awk '{print $1}'", shell=True).split()
- for mod in drivers:
- if mod not in all_drivers:
- return False
- return True
-
- def test_load(self):
- self.driver_mgr.clean()
- for _, drivers in self.driver_mgr.get_all_supported_drivers().items():
- self.assertFalse(self._driver_exists(drivers))
-
-
- self.driver_mgr.load(['ixgbe', 'vhost_net'])
- self.assertTrue(self._driver_exists(['ixgbe', 'vhost_net']))
-
- self.driver_mgr.clean()
- self.assertFalse(self._driver_exists(['ixgbe', 'vhost_net']))
-
- def test_load_unsuported_pair(self):
- with self.assertRaises(Exception):
- self.driver_mgr.load(['ixgbe', 'tap_vhost'])
-
-
-if __name__ == "__main__":
- import logging
-
- logging.basicConfig(level=logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-10-9
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.common.utils import check_output
-from vstf.agent.unittest.env import model
-from vstf.agent.env.driver_plugins import origin_driver
-
-
-class Test(model.Test):
-
- def setUp(self):
- super(Test, self).setUp()
- self.driver_mgr = origin_driver.OriginDriverPlugin()
-
- def tearDown(self):
- super(Test, self).tearDown()
-
- def _driver_exists(self, drivers=[]):
- all_drivers = check_output("lsmod | awk '{print $1}'",shell = True).split()
- for mod in drivers:
- if mod not in all_drivers:
- return False
- return True
-
- def test_load(self):
- self.driver_mgr.clean()
- self.assertFalse(self._driver_exists(self.driver_mgr.get_supported_drivers()))
-
- self.driver_mgr.load(['ixgbe','vhost_net'])
- self.assertTrue(self._driver_exists(['ixgbe','vhost_net']))
-
- self.driver_mgr.clean()
- self.assertFalse(self._driver_exists(self.driver_mgr.get_supported_drivers()))
-
-
-if __name__ == "__main__":
- import logging
- logging.basicConfig(level = logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
\ No newline at end of file
+++ /dev/null
-"""
-Created on 2015-10-9
-
-@author: y00228926
-"""
-import unittest
-import shutil
-import time
-import os
-
-from vstf.agent.env.basic.source_manager import SourceCodeManager
-from vstf.agent.unittest.env import model
-
-
-class TestSourceManager(model.Test):
- def setUp(self):
- super(TestSourceManager, self).setUp()
- self.sm = SourceCodeManager()
- self.dest_path = '/tmp/test_source_manager'
- os.mkdir(self.dest_path)
-
- def tearDown(self):
- shutil.rmtree(self.dest_path, ignore_errors = True)
-
- def _time(self,func):
- def _deco(*args):
- start_time = time.time()
- func(*args)
- end_time = time.time()
- return end_time - start_time
- return _deco
-
- def test_download_source_code(self):
- for key, item in self.source_repo.items():
- print self.source_repo
- url = item['url']
- target = os.path.join(self.dest_path, key)
- install = item['install']
- if install:
- self.sm._git_pull(url, target)
- self.assertTrue(os.path.isdir(target))
- my_download = self._time(self.sm._git_pull)
- t = my_download(url, target)
- self.assertTrue(t < 1.0)
- else:
- self.assertFalse(os.path.isdir(target))
-
-
-if __name__ == "__main__":
- import logging
- logging.basicConfig(level = logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
\ No newline at end of file
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest.env import model
-from vstf.agent.env.basic.vm9pfs import VMConfigBy9pfs
-from vstf.agent.env.basic.vm_manager import VMControlOperation
-
-
-class TestVM9pfs(model.Test):
- def setUp(self):
- super(TestVM9pfs, self).setUp()
- self.vm_config = {
- 'vm_name': 'vm1',
- 'vm_cpu': 5,
- 'image_path': "/mnt/sdb/ubuntu_salt_master.img",
- 'child_dir': '/mnt/sdb/',
- 'image_type': 'qcow2',
- 'ctrl_br': 'br0',
- 'ctrl_mac': '56:6f:44:a5:3f:a4',
- "taps": [
- {
- "tap_name": "tap_in",
- "br_type": "bridge",
- "br_name": "br0",
- "tap_mac": "56:6f:44:a5:3f:a2",
- },
- {
- "tap_name": "tap_out",
- "br_type": "bridge",
- "br_name": "br0",
- "tap_mac": "56:6f:44:a5:3f:a3",
- }
- ],
- 'init_config': {}
- }
- self.init_config = {
- 'amqp_server': '192.168.188.10',
- 'ctrl_ip_setting': '192.168.188.200/23',
- 'tap_pktloop_config': 'dpdk',
- 'ctrl_gw': '192.168.188.1'
- }
- self.mgr = VMControlOperation()
- self.mgr.clean_all_vms()
- self.mgr.create_vm(self.vm_config)
- self.mgr.wait_vm(self.vm_config["vm_name"])
- self.vm9pctrl = self.mgr.vm_9p_controllers[self.vm_config["vm_name"]]
-
- def tearDown(self):
- self.mgr.clean_all_vms()
- super(TestVM9pfs, self).tearDown()
-
- def test_init_config(self):
- ret = self.vm9pctrl.config_ip(self.vm_config['ctrl_mac'], self.init_config['ctrl_ip_setting'])
- self.assertTrue(ret)
- ret = self.vm9pctrl.config_gw(self.init_config['ctrl_gw'])
- self.assertTrue(ret)
- ret = self.vm9pctrl.set_pktloop_dpdk([self.vm_config['taps'][0]['tap_mac'], self.vm_config['taps'][1]['tap_mac']])
- self.assertTrue(ret)
- ret = self.vm9pctrl.config_amqp(self.init_config['ctrl_ip_setting'].split('/')[0], self.init_config['amqp_server'])
- self.assertTrue(ret)
-
-
-if __name__ == "__main__":
- import logging
- logging.basicConfig(level=logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest.env import model
-from vstf.agent.env.basic.vm9pfs import VMConfigBy9pfs
-from vstf.agent.env.basic.vm_manager import VMControlOperation
-
-
-class TestVM9pfs(model.Test):
- def setUp(self):
- super(TestVM9pfs, self).setUp()
- self.vm_config = {
- 'vm_name': 'vm1',
- 'vm_cpu': 5,
- 'image_path': "/mnt/sdb/ubuntu_salt_master.img",
- 'child_dir': '/mnt/sdb/',
- 'image_type': 'qcow2',
- 'ctrl_br': 'br0',
- 'ctrl_mac': '56:6f:44:a5:3f:a4',
- "taps": [
- {
- "tap_name": "tap_in",
- "br_type": "bridge",
- "br_name": "br0",
- "tap_mac": "56:6f:44:a5:3f:a2",
- },
- {
- "tap_name": "tap_out",
- "br_type": "bridge",
- "br_name": "br0",
- "tap_mac": "56:6f:44:a5:3f:a3",
- }
- ],
- 'init_config': {
- "amqp_passwd": "guest",
- "amqp_user": "guest",
- "amqp_server": "192.168.188.10",
- "amqp_port": 5672,
- 'ctrl_ip_setting': '192.168.188.200/23',
- 'tap_pktloop_config': 'dpdk',
- 'ctrl_gw': '192.168.188.1'
- }
- }
- self.mgr = VMControlOperation()
- self.mgr.clean_all_vms()
-
- def tearDown(self):
- self.mgr.clean_all_vms()
- super(TestVM9pfs, self).tearDown()
-
- def test_create_vm_bridge(self):
- self.mgr.create_vm(self.vm_config)
- self.mgr.wait_vm(self.vm_config["vm_name"])
- self.mgr.init_config_vm(self.vm_config["vm_name"])
-
- def _replace_opts(self, cfg, br_type):
- for tap_cfg in cfg["taps"]:
- tap_cfg["br_type"] = br_type
-
-
-if __name__ == "__main__":
- import logging
- logging.basicConfig(level=logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest.env import model
-from vstf.agent.env.vswitch_plugins import manager
-from vstf.common.utils import check_call
-
-
-class TestVsPlugins(model.Test):
- def setUp(self):
- super(TestVsPlugins, self).setUp()
- self.cfg = {
- "type": "ovs",
- "name": "ovs1",
- "uplinks": [
- {
- "bdf": self.bdf_of_eth[0],
- "vlan_mode": "trunk",
- "vlan_id": "100,200,300,400"
- },
- {
- "bdf": self.bdf_of_eth[1],
- "vlan_mode": "trunk",
- "vlan_id": "100,200,300,400"
- }
- ],
- "vtep": {}
- }
- self.mgr = manager.VswitchPluginManager()
-
- def tearDown(self):
- super(TestVsPlugins, self).tearDown()
-
- def _check_br_exists(self, name):
- try:
- check_call('ifconfig %s' % name, shell=True)
- except Exception, e:
- return False
- return True
-
- def test_create_bridge(self):
- self.cfg['name'] = 'br1'
- self.br = self.mgr.get_vs_plugin('bridge')
- self.br.clean()
- self.br.init()
- self.br.create_br(self.cfg)
- self.assertTrue(self._check_br_exists('br1'))
- self.br.clean()
- self.assertFalse(self._check_br_exists('br1'))
-
- def test_clean(self):
- self.mgr.clean()
-
- def test_get_supported_plugins(self):
- ret = self.mgr.get_supported_plugins()
- self.assertEqual(set(ret), {'bridge', 'ovs'})
-
-
-if __name__ == "__main__":
- import logging
-
- logging.basicConfig(level=logging.INFO)
- LOG = logging.getLogger(__name__)
- unittest.main()
+++ /dev/null
-To enable Perf unittest, please make sure:
-
- 1. you have setup a "Tn virtual network" on Target Host, so you can ping from one nic to the other nic.
- 2. change options in "unittest/configuration.py".
-
-run tests:
- python run_test.py
\ No newline at end of file
+++ /dev/null
-# Copyright Huawei Technologies Co., Ltd. 1998-2015.
-# All Rights Reserved.
-#
-# 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.
+++ /dev/null
-"""
-Created on 2015-9-25
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest import configuration
-from vstf.agent.perf import netns
-
-
-class LocalModel(unittest.TestCase):
- def _ping(self):
- device_list, ns_list, ip_setting_list, ip_list = self.device_list, self.ns_list, self.ip_setting_list, self.ip_list
- for ns, dev, ip_setting in zip(ns_list, device_list, ip_setting_list):
- netdev = {
- "namespace": ns,
- "iface": dev,
- 'ip_setting': ip_setting
- }
- self.mgr.config_dev(netdev)
- ip_list_copy = ip_list[:]
- ip_list_copy.reverse()
- for ns, ip in zip(ns_list, ip_list_copy):
- self.assertTrue(sself.mgr.ping(ns, ip), True)
- self.mgr.clean_all_namespace()
-
- def setUp(self):
- # make sure you have set up Tn loop on the "Target Host"
- self.mgr = netns.NetnsManager()
- self.mgr.clean_all_namespace()
- self.device_list = configuration.eth_for_test
- self.mac_list = configuration.mac_of_eth
- self.ns_list = ['send', 'receive']
- self.ip_setting_list = ['192.168.1.1/24', '192.168.1.2/24']
- self.ip_list = ['192.168.1.1', '192.168.1.2']
-
- def tearDown(self):
- self.mgr.clean_all_namespace()
-
-
-class Model(LocalModel):
- def setUp(self):
- # make sure you have set up Tn loop on the "Target Host"
- super(Model, self).setUp()
- self._ping()
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-import importlib
-
-test_order_list = [
- "vstf.agent.unittest.perf.test_utils",
- "vstf.agent.unittest.perf.test_netns",
- "vstf.agent.unittest.perf.test_netperf",
- "vstf.agent.unittest.perf.test_qperf",
- "vstf.agent.unittest.perf.test_pktgen",
- "vstf.agent.unittest.perf.test_vstfperf",
-]
-
-
-def main():
- import logging
- logging.getLogger(__name__)
- logging.basicConfig(level=logging.DEBUG)
- suite = unittest.TestSuite()
- for mod_name in test_order_list:
- mod = importlib.import_module(mod_name)
- suit = unittest.TestLoader().loadTestsFromModule(mod)
- suite.addTest(suit)
- unittest.TextTestRunner().run(suite)
-
-
-if __name__ == '__main__':
- main()
+++ /dev/null
-#!/usr/bin/python
-# -*- coding: utf8 -*-
-# author: wly
-# date: 2015/11/13
-# see license for license details
-
-import unittest
-import logging
-from vstf.agent.perf import ethtool
-from vstf.agent.unittest import configuration
-from vstf.common.log import setup_logging
-
-
-LOG = logging.getLogger(__name__)
-
-
-class Testethtool(unittest.TestCase):
- def setUp(self):
- LOG.info("start Testethtool unit test.")
- self._devices = configuration.eth_for_test
- super(Testethtool, self).setUp()
-
- def teardown(self):
- LOG.info("stop Testethtool unit test.")
-
-# @unittest.skip('for now')
- def test_autoneg_on(self):
- for dev in self._devices:
- self.assertTrue(ethtool.autoneg_on(dev), True)
-
- def test_autoneg_off(self):
- for dev in self._devices:
- self.assertTrue(ethtool.autoneg_off(dev), True)
-
- def test_autoneg_query(self):
- for dev in self._devices:
- result = ethtool.autoneg_query(dev)
- LOG.info(result)
-
-if __name__ == "__main__":
- setup_logging(level=logging.INFO, log_file="/var/log/vstf/vstf-unit-test.log", clevel=logging.INFO)
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-23
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest.perf import model
-from vstf.agent.perf import netns
-
-
-class TestNetnsManager(model.Model):
- def setUp(self):
- super(TestNetnsManager, self).setUp()
- self.ns = netns.Netns()
-
- def tearDown(self):
- super(TestNetnsManager, self).tearDown()
-
- def testNetns(self):
- device_list,ns_list,ip_setting_list,ip_list = self.device_list,self.ns_list,self.ip_setting_list,self.ip_list
- net = self.ns
- for ns in ns_list:
- self.assertTrue(net.create_namespace(ns),'create_namespace failed')
- for ns,dev,ip_setting in zip(ns_list,device_list,ip_setting_list):
- self.assertTrue(net.add_device(ns, dev),'add_device failed')
- self.assertTrue(net.activate_device(ns,dev),'activate_device failed')
- self.assertTrue(net.config_ip(ns,dev,ip_setting),'config_ip failed')
- for ns in ns_list:
- self.assertTrue(net.remove_namespace(ns),'remove_namespace failed')
-
- def testNetNsManager(self):
- mgr = self.mgr
- device_list,ns_list,ip_setting_list,ip_list = self.device_list,self.ns_list,self.ip_setting_list,self.ip_list
- for ns,dev,ip_setting in zip(ns_list,device_list,ip_setting_list):
- netdev = {
- "namespace":ns,
- "iface":dev,
- 'ip_setting':ip_setting
- }
- ret = mgr.config_dev(netdev)
- self.assertTrue(ret,"config_dev failed, netdev=%s" % netdev)
-
- for ns,dev,ip_setting in zip(ns_list,device_list,ip_setting_list):
- netdev = {
- "namespace":ns,
- "iface":dev,
- 'ip_setting':ip_setting
- }
- self.assertTrue(mgr.recover_dev(netdev),"recover_dev failed, netdev=%s" % netdev)
-
- for ns,dev,ip_setting in zip(ns_list,device_list,ip_setting_list):
- netdev = {
- "namespace":ns,
- "iface":dev,
- 'ip_setting':ip_setting
- }
- self.assertTrue(mgr.config_dev(netdev),"config_dev failed, netdev=%s" % netdev)
- self.assertTrue(mgr.clean_all_namespace(),'remove_namespace failed')
-
-
-if __name__ == "__main__":
- import logging
- LOG = logging.getLogger(__name__)
- logging.basicConfig(level = logging.DEBUG)
- #import sys;sys.argv = ['', 'Test.testName']
- unittest.main()
\ No newline at end of file
+++ /dev/null
-'''
-Created on 2015-9-24
-
-@author: y00228926
-'''
-import unittest
-import time
-import subprocess
-
-from vstf.agent.unittest.perf import model
-from vstf.agent.perf import netperf
-from vstf.agent.perf.utils import get_pid_by_name
-
-
-class TestNetperf(model.Model):
- '''
- please make sure 'Tn' network on 'Target Host' is created.
- '''
- def setUp(self):
- super(TestNetperf, self).setUp()
- subprocess.call("killall netperf", shell = True)
- subprocess.call("killall netserver",shell = True)
- for ns, dev, ip_setting in zip(self.ns_list, self.device_list, self.ip_setting_list):
- netdev = {
- "namespace":ns,
- "iface":dev,
- 'ip_setting':ip_setting
- }
- self.mgr.config_dev(netdev)
- self.send_cfg = {
- "namespace": "send",
- "protocol": "udp_bw",
- "dst":[
- {"ip": "192.168.1.2"}
- ],
- "size": 64,
- "threads": 1,
- "time": 10,
- }
-
- def tearDown(self):
- super(TestNetperf, self).tearDown()
-
- def test_netperf_start_success(self):
- perf = netperf.Netperf()
- ret = perf.receive_start(namespace='receive')
- exp = (0, 'start netserver success')
- self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
-
- ret = perf.send_start(**self.send_cfg)
- exp = (0,"start netperf send success")
- self.assertEqual(ret, exp, "failed to start netperf")
-
- time.sleep(3)
-
- ret = perf.send_stop()
- exp = [(0, "process is stopped by killed")]
- self.assertEqual(ret, exp, "send_stop failed, ret = %s" % str(ret))
-
- ret = perf.receive_stop()
- exp = (0, "stop netserver success")
- self.assertEqual(ret, exp, "receive_stop failedf, ret = %s" % str(ret))
-
- def test_netperf_start_success_mutil_threads(self):
- perf = netperf.Netperf()
- ret = perf.receive_start(namespace='receive')
- exp = (0, 'start netserver success')
- self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
-
- self.send_cfg.update({"threads":3})
- exp = (0,"start netperf send success")
- ret = perf.send_start(**self.send_cfg)
- self.assertEqual(ret, exp, "failed to start netperf")
-
- time.sleep(3)
-
- rets = perf.send_stop()
- exp = [(0, 'process is stopped by killed'), (0, 'process is stopped by killed'), (0, 'process is stopped by killed')]
- self.assertEqual(rets, exp, "send_stop failed, rets = %s" % str(rets))
-
- rets = perf.receive_stop()
- self.assertEqual(rets, (0, "stop netserver success"), "receive_stop failedf, rets = %s" % str(rets))
-
- def test_clean(self):
- perf = netperf.Netperf()
- ret = perf.receive_start(namespace='receive')
- exp = (0, 'start netserver success')
- self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
-
- self.send_cfg.update({"threads":3})
- exp = (0,"start netperf send success")
- ret = perf.send_start(**self.send_cfg)
- self.assertEqual(ret, exp, "failed to start netperf")
- perf.clean()
- ret = get_pid_by_name('netperf')
- self.assertEqual(ret, [], "failed to clean netperf")
- ret = get_pid_by_name('netserver')
- self.assertEqual(ret, [], "failed to clean netserver")
-
-if __name__ == "__main__":
- import logging
- logging.getLogger(__name__)
- logging.basicConfig(level = logging.DEBUG)
- #import sys;sys.argv = ['', 'Test.testName']
- unittest.main()
\ No newline at end of file
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-import time
-
-from vstf.agent.unittest.perf import model
-from vstf.agent.perf import pktgen
-
-
-class TestPktgen(model.Model):
- def setUp(self):
- super(TestPktgen, self).setUp()
-
- def tearDown(self):
- super(TestPktgen, self).tearDown()
-
- def test_single_thread(self):
- perf = pktgen.Pktgen()
- print perf.receive_start()
- send = {
- "src": [
- {"iface": self.device_list[0], "mac": self.mac_list[0]}
- ],
- "dst": [
- {"mac": self.mac_list[1]}
- ],
- "size": 64,
- "threads": 1,
- 'ratep': 0
- }
- ret = perf.send_start(**send)
- self.assertEqual((0, 'start pktgen send success'), ret, "send_start failed, ret=%s" % str(ret))
- time.sleep(5)
- ret = perf.send_stop()
- self.assertEqual([(0, '')], ret, "send_start failed, ret=%s" % ret)
- ret = perf.receive_stop()
- self.assertEqual((0, 'pktgen neednt receive stop'), ret, "send_stop failed, ret=%s" % str(ret))
-
- def test_single_thread_bidirectional(self):
- perf = pktgen.Pktgen()
- print perf.receive_start()
- send = {
- "src": [
- {"iface": self.device_list[0], "mac": self.mac_list[0]},
- {"iface": self.device_list[1], "mac": self.mac_list[1]}
- ],
- "dst": [
- {"mac": self.mac_list[1]},
- {"mac": self.mac_list[0]}
- ],
- "size": 64,
- "threads": 1,
- 'ratep': 0
- }
- ret = perf.send_start(**send)
- self.assertEqual((0, 'start pktgen send success'), ret, "send_start failed, ret=%s" % str(ret))
- time.sleep(5)
- ret = perf.send_stop()
- self.assertEqual([(0, '')], ret, "send_start failed, ret=%s" % ret)
- ret = perf.receive_stop()
- self.assertEqual((0, 'pktgen neednt receive stop'), ret, "send_stop failed, ret=%s" % str(ret))
-
- def test_clean(self):
- perf = pktgen.Pktgen()
- print perf.receive_start()
- send = {
- "src": [
- {"iface": self.device_list[0], "mac": self.mac_list[0]}
- ],
- "dst": [
- {"mac": self.mac_list[1]}
- ],
- "size": 64,
- "threads": 1,
- 'ratep': 0
- }
- ret = perf.send_start(**send)
- self.assertEqual((0, 'start pktgen send success'), ret, "send_start failed, ret=%s" % str(ret))
- perf.clean()
-
-
-if __name__ == "__main__":
- import logging
-
- logging.getLogger(__name__)
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
+++ /dev/null
-'''
-Created on 2015-9-24
-
-@author: y00228926
-'''
-import unittest
-import subprocess
-import time
-
-from vstf.agent.unittest.perf import model
-from vstf.agent.perf import qperf
-from vstf.agent.perf.utils import get_pid_by_name
-
-
-class testQperf(model.Model):
- def setUp(self):
- super(testQperf, self).setUp()
- subprocess.call("killall qperf", shell = True)
- for ns, dev, ip_setting in zip(self.ns_list, self.device_list, self.ip_setting_list):
- netdev = {
- "namespace":ns,
- "iface":dev,
- 'ip_setting':ip_setting
- }
- self.mgr.config_dev(netdev)
- self.send_cfg = {
- "namespace": self.ns_list[0],
- "time":1,
- "protocol": "udp_lat",
- "dst":[
- {"ip": self.ip_list[1]}
- ],
- "size": 64,
- }
-
- def tearDown(self):
- super(testQperf, self).tearDown()
-
- def test_qperf_quick(self):
- perf = qperf.Qperf()
- ret = perf.receive_start(namespace=self.ns_list[1])
- exp = (0, "start qperf receive success")
- self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
-
- ret = perf.send_start(**self.send_cfg)
- exp = (0,"start qperf send success")
- self.assertEqual(ret, exp, "send_start failed")
-
- time.sleep(3)
-
- ret = perf.send_stop()
- for r in ret:
- self.assertEqual(r[0], 0, "send_stop failed, ret = %s" % str(ret))
- for key in ('MaximumLatency', 'AverageLatency', 'MinimumLatency'):
- self.assertIn(key, r[1], "send_stop failed, ret = %s" % str(ret))
-
- ret = perf.receive_stop()
- exp = (0, "stop qperf receive success")
- self.assertEqual(ret, exp, "receive_stop failed, ret = %s" % str(ret))
-
- def test_qperf_quick_3s(self):
- perf = qperf.Qperf()
- self.send_cfg.update({"time":3})
- ret = perf.receive_start(namespace='receive')
- exp = (0, 'start qperf receive success')
- self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
-
- self.send_cfg.update({"threads":3})
- exp = (0,"start qperf send success")
- ret = perf.send_start(**self.send_cfg)
- self.assertEqual(ret, exp, "send_start failed %s" % str(ret))
-
- ret = perf.send_stop()
- for r in ret:
- self.assertEqual(r[0], 0, "send_stop failed, ret = %s" % str(ret))
- for key in ('MaximumLatency', 'AverageLatency', 'MinimumLatency'):
- self.assertIn(key, r[1], "send_stop failed, ret = %s" % str(ret))
-
- ret = perf.receive_stop()
- self.assertEqual(ret, (0, 'stop qperf receive success'), "receive_stop failedf, ret = %s" % str(ret))
-
- def test_clean(self):
- perf = qperf.Qperf()
- self.send_cfg.update({"time":10})
- ret = perf.receive_start(namespace=self.ns_list[1])
- exp = (0, "start qperf receive success")
- self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
-
- ret = perf.send_start(**self.send_cfg)
- exp = (0,"start qperf send success")
- self.assertEqual(ret, exp, "send_start failed")
-
- perf.clean()
- ret = get_pid_by_name('qperf')
- self.assertEqual(ret, [], "clean qperf failed")
-
-
-if __name__ == "__main__":
- import logging
- logging.getLogger(__name__)
- logging.basicConfig(level = logging.DEBUG)
- unittest.main()
\ No newline at end of file
+++ /dev/null
-"""
-Created on 2015-9-24
-
-@author: y00228926
-"""
-import unittest
-from vstf.common import utils
-
-
-class TestUtils(unittest.TestCase):
- def setUp(self):
- pass
-
- def tearDown(self):
- pass
-
- def test_ns_cmd(self):
- cmd = "ls"
- ns = "xx"
- exp_cmd = "ip netns exec xx ls"
- ret = utils.ns_cmd(ns, cmd)
- self.assertEqual(ret, exp_cmd, "ns_cmd failed to add ns header prefix:%s != %s" % (ret, exp_cmd))
-
-
-if __name__ == "__main__":
- import logging
-
- logging.getLogger(__name__)
- logging.basicConfig(level=logging.DEBUG)
- # import sys;sys.argv = ['', 'Test.testName']
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-23
-
-@author: y00228926
-"""
-import unittest
-import time
-
-from vstf.agent.perf.utils import get_pid_by_name
-from vstf.agent.unittest.perf import model
-from vstf.agent.perf import vnstat
-
-
-class TestVnstat(model.LocalModel):
- def setUp(self):
- super(TestVnstat, self).setUp()
- self.vnstat = vnstat.VnStat()
- self.namespace = self.ns_list[0]
- self.device = self.device_list[0]
- self.ip_setting = self.ip_setting_list[0]
- netdev = {
- "namespace": self.namespace,
- "iface": self.device,
- 'ip_setting': self.ip_setting
- }
- self.mgr.config_dev(netdev)
-
- def tearDown(self):
- super(TestVnstat, self).tearDown()
-
- def test_run_vnstat(self):
- logging.basicConfig(level=logging.DEBUG)
- pid = self.vnstat.run_vnstat(self.device, self.namespace)
- time.sleep(12)
- raw = self.vnstat.kill_vnstat(pid, self.namespace)
- print raw['raw_data']
- data = self.vnstat.process(raw['raw_data'])
- self.assertTrue(type(data) is dict)
- for key in ('rxmB/s', 'txmB', 'rxpck', 'txpck', \
- 'rxpck_min/s', 'txmB_max/s', 'txpck_max/s', \
- 'txmB/s', 'rxmB', 'rxmB_max/s', 'rxpck/s', 'rxmB_min/s', \
- 'time', 'rxpck_max/s', 'txpck_min/s', 'txpck/s', 'txmB_min/s'):
- self.assertTrue(key in data)
-
- def test_clean(self):
- self.vnstat.run_vnstat(self.device, self.namespace)
- self.vnstat.clean()
- self.assertTrue(get_pid_by_name('vnstat') == [])
-
-
-if __name__ == "__main__":
- import logging
-
- LOG = logging.getLogger(__name__)
- logging.basicConfig(level=logging.DEBUG)
- # import sys;sys.argv = ['', 'Test.testName']
- unittest.main()
+++ /dev/null
-"""
-Created on 2015-9-28
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.perf.vstfperf import Vstfperf
-from vstf.agent.unittest.perf import model
-
-
-class Test(model.Model):
- def setUp(self):
- super(Test, self).setUp()
-
- for ns, dev, ip_setting in zip(self.ns_list, self.device_list, self.ip_setting_list):
- net_dev = {
- "namespace":ns,
- "iface":dev,
- 'ip_setting':ip_setting
- }
- self.mgr.config_dev(net_dev)
-
- self.start = {
- "operation": "start",
- "action": "send",
- "tool": "netperf",
- "params":{
- "namespace": self.ns_list[0],
- "protocol": "tcp_lat",
- "dst":[
- {"ip": self.ip_list[1]}
- ],
- "size": 64,
- "threads": 1,
- "time": 1,
- },
- }
- self.stop = {
- "operation": "stop",
- "action": "send",
- "tool": "netperf",
- "params":{
- "namespace": self.ns_list[1],
- },
- }
-
- def tearDown(self):
- super(Test, self).tearDown()
-
- def testNetperf(self):
- perf = Vstfperf()
- self.start['tool'] = 'netperf'
- self.stop['tool'] = 'netperf'
- self.start['action'] = 'receive'
- self.start['operation'] = 'start'
- self.start['params']['namespace'] = self.ns_list[1]
- self.start['params']['protocol'] = 'udp_bw'
- perf.run(**self.start)
- self.start['action'] = 'send'
- self.start['operation'] = 'start'
- self.start['params']['namespace'] = self.ns_list[0]
- perf.run(**self.start)
- self.stop['action'] = 'send'
- self.stop['operation'] = 'stop'
- self.stop['params']['namespace'] = self.ns_list[0]
- perf.run(**self.stop)
- self.stop['action'] = 'receive'
- self.stop['operation'] = 'stop'
- self.stop['params']['namespace'] = self.ns_list[1]
- perf.run(**self.stop)
-
- def testQperf(self):
- perf = Vstfperf()
- self.start['tool'] = 'qperf'
- self.stop['tool'] = 'qperf'
- self.start['action'] = 'receive'
- self.start['operation'] = 'start'
- self.start['params']['namespace'] = self.ns_list[1]
- perf.run(**self.start)
- self.start['action'] = 'send'
- self.start['operation'] = 'start'
- self.start['params']['namespace'] = self.ns_list[0]
- perf.run(**self.start)
- self.stop['action'] = 'send'
- self.stop['operation'] = 'stop'
- self.stop['params']['namespace'] = self.ns_list[0]
- perf.run(**self.stop)
- self.stop['action'] = 'receive'
- self.stop['operation'] = 'stop'
- self.stop['params']['namespace'] = self.ns_list[1]
- perf.run(**self.stop)
-
-
-if __name__ == "__main__":
- import logging
- logging.basicConfig(level = logging.DEBUG)
- unittest.main()
\ No newline at end of file
+++ /dev/null
-"""
-Created on 2015-9-28
-
-@author: y00228926
-"""
-import unittest
-
-from vstf.agent.unittest.perf import model
-from vstf.agent import softagent
-
-
-class TestCallback(model.Model):
- def setUp(self):
- super(TestCallback, self).setUp()
- self.agent = softagent.softAgent()
- for ns, dev, ip_setting in zip(self.ns_list, self.device_list, self.ip_setting_list):
- netdev = {
- "namespace": ns,
- "iface": dev,
- 'ip_setting': ip_setting
- }
- self.mgr.config_dev(netdev)
-
- self.start = {
- "operation": "start",
- "action": "send",
- "tool": "netperf",
- "params": {
- "namespace": self.ns_list[0],
- "protocol": "tcp_lat",
- "dst": [
- {"ip": self.ip_list[1]}
- ],
- "size": 64,
- "threads": 1,
- "time": 1,
- },
- }
- self.stop = {
- "operation": "stop",
- "action": "send",
- "tool": "netperf",
- "params": {
- "namespace": self.ns_list[1],
- },
- }
-
- def tearDown(self):
- super(TestCallback, self).tearDown()
-
- def test_clean(self):
- agent = self.agent
- agent.perf_clean()
- self.start['tool'] = 'netperf'
- self.stop['tool'] = 'netperf'
- self.start['action'] = 'receive'
- self.start['operation'] = 'start'
- self.start['params']['namespace'] = self.ns_list[1]
- self.start['params']['protocol'] = 'udp_bw'
- agent.perf_run(**self.start)
- self.start['action'] = 'send'
- self.start['operation'] = 'start'
- self.start['params']['namespace'] = self.ns_list[0]
- agent.perf_run(**self.start)
- agent.perf_clean()
-
-
-if __name__ == "__main__":
- import logging
-
- logging.getLogger(__name__)
- logging.basicConfig(level=logging.DEBUG)
- # import sys;sys.argv = ['', 'Test.testName']
- unittest.main()
-
-if __name__ == "__main__":
- import logging
-
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-__author__ = 'wly'
-__version__ = '0.1'
-
import os
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-__version__ = ''' '''
import os
import logging
-#!/usr/bin/python
-# -*- coding: utf8 -*-
-# author: wly
-# date: 2015-12-25
-# see license for license details
-__version__ = ''' '''
+##############################################################################
+# Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+
from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from vstf.controller.database import constants as const
+++ /dev/null
- env_build.py contains a quick test code for create virtual network in a remote host.
-
- usage:
-
- python env_build.py --rpc_server 192.168.188.10 --config /etc/vstf/env/Tn.json
-
- --rpc_server RPC_SERVER
- the rabbitmq server for deliver messages.
- --config CONFIG
- env-build config file to parse
-
-the above command will build a 'Tn-type' network according to config file: /etc/vstf/env/Tn.json.
-
-
-
+++ /dev/null
-Tree
-
-├── __init__.py
-├── mail
-│ ├── __init__.py
-│ ├── mail.py
-│ └── sendmail.py
-├── report
-│ ├── data_factory.py
-│ ├── html
-│ │ ├── html_base.py
-│ │ ├── htmlcreater.py
-│ │ ├── html_text.py
-│ │ └── __init__.py
-│ ├── __init__.py
-│ ├── pdf
-│ │ ├── element.py
-│ │ ├── __init__.py
-│ │ ├── pdfcreater.py
-│ │ ├── pdftemplate.py
-│ │ ├── story.py
-│ │ └── styles.py
-│ └── provider
-│ ├── html_provider.py
-│ └── __init__.py
-└── reporter.py
-
-
-Entry:
- reporter.py
-
- usage: reporter.py [-h] [-rpath RPATH] [-mail_off] [--taskid TASKID]
-
- optional arguments:
- -h, --help show this help message and exit
- -rpath RPATH the path name of test results
- -mail_off is need send mail the for the report
- --taskid TASKID report depand of a history task id.
-
-Settings:
- mail_settings
-
- {
- "server":
- {
- "host": "localhost",
- "username": null,
- "password": null
- },
- "body":
- {
- "from": ["vstf_from@vstf.com"],
- "to": ["vstf_to@vstf.com"],
- "cc": ["vstf_cc@vstf.com"],
- "bcc": ["vstf_bcc@vstf.com"],
- "subject": "Elastic Virtual Switching Performance Test Report"
- }
- }
-
- html_settings
-
- {
- "style":{
- "table":{
- "font-family":"\"Trebuchet MS\", Arial, Helvetica, sans-serif",
- "border":"1px solid green",
- "border-collapse":"collapse",
- "padding":"8px",
- "text-align":"center"
- },
- "td":{
- "border":"1px solid green",
- "padding":"8px",
- "word-wrap":"break-all"
- },
- "th":{
- "background-color":"#EAF2D3",
- "border":"1px solid green",
- "padding":"8px"
- }
- }
- }
-
- data_settings
-
- {
- "ovs":{
- "content":{
- "version":3.0
- },
- "title":"Ovs info"
- },
- "result":{
- "content":{},
- "title":"Performance Result"
- },
- "subject":"ATF Performance Test Tnv Model"
- }
-Module:
- mail
- html
- pdf
-
-
-Others:
- pip processes the package "reportlab"
-
- pip install reportlab
-
if __name__ == "__main__":
m = Mail()
m.attach_addr(["vstf_server@vstf.com"], m.FROM)
- m.attach_addr(["wangli11@huawei.com"], m.TO)
+ m.attach_addr(["test@test.com"], m.TO)
context = """
<!DOCTYPE html>
<html>
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-__version__ = ''' '''
import os
import vstf.common.pyhtml as pyhtm
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-__version__ = ''' '''
import logging
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-__version__ = ''' '''
+
__doc__ = """
it contains the base element for pdf
eImage is used to draw picture on the pdf document
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-__version__ = ''' '''
-
from vstf.controller.reporters.report.pdf.pdftemplate import PdfVswitch
from vstf.controller.reporters.report.pdf.story import TitleStory, SpaceStory, ImageStory, TableStory, \
##############################################################################
-
-__version__ = ''' '''
import logging
LOG = logging.getLogger(__name__)
import vstf.common.constants as cst
-__version__ = ''' '''
LOG = logging.getLogger(__name__)
+++ /dev/null
-This module providers a set of profile management solution
-File:
- settings.py
-Interface:
- Settings
-
- 1. it is a base class and supports two modes "Default" and "Single"
- if the mode is "Default", the program will load the 'json' file from 'default'
- and 'user' , merge the input, save only the 'json' file from 'user'
- if the mode is "Single", the program will only load and save the 'json' file
- 2. it saves a file two, one is only in memory and the other is in file
- 3. it provides two types of functions, one is like "set_" and "add_" and the
- other is like "mset" and "madd". the functions are automatically registered.
-
- 4. You can overload the function _register_func to achieve the functions what you desire
- by function "_setting_file","_adding_file","_setting_memory" and "_addting_memory"
-
- 5. it provides "settings" to show the result in memory
- 6. it provides "reset" to reload the file
-
-Example:
-
- 1. create your-settings file and paste the contents
-
- {
- "items1": "value1",
- "items2": "value2"
- }
-
-
- 2. create your_settings file and paste the codes
-
- import vstf.controller.settings.settings as sets
- class YourSettings(sets.Settings):
- def __init__(self, path="./", filename="your-settings", mode=sets.SETS_SINGLE):
- super(MailSettings, self).__init__(path, filename, mode)
-
- def unit_test():
- setting = YourSettings()
- print setting.settings()
- value1 = "test_set_items1"
- setting.set_items1(value1)
- print setting.settings()
- value2 = "test_set_items2"
- setting.mset_items2(value2)
- print setting.settings()
- settings.reset()
- print setting.settings()
-
- if __name__ == '__main__':
- unit_test()
-
-Tree:
-
- data_settings.py
- flows_settings.py
- perf_settings.py
-
- mail_settings.py
- tool_settings.py
- html_settings.py
\ No newline at end of file
}
mail_settings.set_server(mail_server)
- from_list = ['vstf_from@vstf.com']
- mail_settings.set_from(from_list)
- to_list = ['wangli11@huawei.com']
- mail_settings.set_to(to_list)
- cc_list = ['wangli11@huawei.com']
- mail_settings.set_cc(cc_list)
- bcc_list = ['wangli11@huawei.com']
- mail_settings.set_bcc(bcc_list)
- bcc_list = ['wangli11@huawei.com']
- mail_settings.set_bcc(bcc_list)
-
subject = "Virtual Switching Performance Test Report"
mail_settings.set_subject(subject)
+++ /dev/null
-Tree
-
-|--- flow_producer.py
-|--- model.py
-|--- performance.py
-|--- perf_provider.py
-|--- raw_data.py
-
-Entry:
- performance.py
- usage: performance.py [-h] [-case CASE]
- [-tool {pktgen,netperf,qperf,iperf,netmap}]
- [-protocol {tcp,udp}] [-profile {rdp,fastlink,l2switch}]
- [-type {throughput,latency,frameloss}] [-sizes SIZES]
- [--monitor MONITOR]
-
- optional arguments:
- -h, --help show this help message and exit
- -case CASE test case like Ti-1, Tn-1, Tnv-1, Tu-1...
- -tool {pktgen,netperf,qperf,iperf,netmap}
- -protocol {tcp,udp}
- -profile {rdp,fastlink,l2switch}
- -type {throughput,latency,frameloss}
- -sizes SIZES test size list "64 128"
- --monitor MONITOR which ip to be monitored
-
-Interface:
- usage:
- conn = Server(host=args.monitor)
- flows_settings = FlowsSettings()
- tool_settings = ToolSettings()
- tester_settings = TesterSettings()
- flow_producer = FlowsProducer(conn, flows_settings)
- provider = PerfProvider(flows_settings.settings, tool_settings.settings, tester_settings.settings)
- perf = Performance(conn, provider)
- flow_producer.create(scenario, case)
- LOG.info(flows_settings.settings())
- result = perf.run(tool, protocol, type, sizes)
-
+++ /dev/null
-
-
-the procedure to integrate a module unit testing into the unit testing framework:
-
-1.create your own unit test module, the name should start by 'test', for example, test_env.py
-
-2.create the test cases inside the module, inherit unittest.TestCase, for example:
- class TestNetnsManager(unittest.TestCase)：
- def setUp(self): // preparing the testig
- pass
- def tearDown(self):// cleanup after testing
- pass
- def testCase1(self):// cases
- pass
-
-3.single modules testing, appending below code at the end of the module, execute 'python test_env.py'.
-
-if __name__ == "__main__":
- import logging
- logging.getLogger(__name__)
- logging.basicConfig(level = logging.DEBUG)
- unittest.main()
-
-4.multiple modules integration, create run_test.py，run_test.py the example code as below:
-
-import unittest
-import importlib
-
-test_order_list = [
- "vstf.services.agent.unittest.perf.test_utils",
- "vstf.services.agent.unittest.perf.test_netns",
- "vstf.services.agent.unittest.perf.test_netperf",
- "vstf.services.agent.unittest.perf.test_qperf",
- "vstf.services.agent.unittest.perf.test_pktgen",
-]
-
-if __name__ == '__main__':
- import logging
- logging.getLogger(__name__)
- logging.basicConfig(level = logging.DEBUG)
- for mod_name in test_order_list:
- mod = importlib.import_module(mod_name)
- suit = unittest.TestLoader().loadTestsFromModule(mod)
- unittest.TextTestRunner().run(suit)
-
Code Review / bottlenecks.git / commitdiff
