mir/ovs
2
0
Fork 0
mirror of https://github.com/openvswitch/ovs synced 2025-08-25 19:37:14 +00:00
Code Issues Releases Wiki Activity
ovs/python/ovstest/tests.py
Russell Bryant 8f808842a0 python: Drop usage of long type. 
Python 2 has both long and int types.  Python 3 only has int, which
behaves like long.

In the case of needing a set of integer types, we can use
six.integer_types which includes int and long for Python 2 and just int
for Python 3.

We can convert all cases of long(value) to int(value), because as of
Python 2.4, when the result of an operation would be too big for an int,
the type is automatically converted to a long.

There were several places in this patch doing type comparisons.  The
preferred way to do this is using the isinstance() or issubclass()
built-in functions, so I converted the similar checks nearby while I was
at it.

Signed-off-by: Russell Bryant <russell@ovn.org>
Acked-by: Ben Pfaff <blp@ovn.org>
2016-02-02 16:40:36 -05:00

251 lines
9.7 KiB
Python
Raw Blame History

# 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 __future__ import print_function
import math
import time
import ovstest.util as util
DEFAULT_TEST_BRIDGE = "ovstestbr0"
DEFAULT_TEST_PORT = "ovstestport0"
DEFAULT_TEST_TUN = "ovstestport1"
NO_HANDLE = -1
def do_udp_tests(receiver, sender, tbwidth, duration, port_sizes):
"""Schedule UDP tests between receiver and sender"""
server1 = util.rpc_client(receiver[0], receiver[1])
server2 = util.rpc_client(sender[0], sender[1])
udpformat = '{0:>15} {1:>15} {2:>15} {3:>15} {4:>15}'
print("UDP test from %s:%u to %s:%u with target bandwidth %s" %
(sender[0], sender[1], receiver[0], receiver[1],
util.bandwidth_to_string(tbwidth)))
print(udpformat.format("Datagram Size", "Snt Datagrams", "Rcv Datagrams",
"Datagram Loss", "Bandwidth"))
for size in port_sizes:
listen_handle = NO_HANDLE
send_handle = NO_HANDLE
try:
packetcnt = (tbwidth * duration) / size
listen_handle = server1.create_udp_listener(receiver[3])
if listen_handle == NO_HANDLE:
print("Server could not open UDP listening socket on port"
" %u. Try to restart the server.\n" % receiver[3])
return
send_handle = server2.create_udp_sender(
(util.ip_from_cidr(receiver[2]),
receiver[3]), packetcnt, size,
duration)
# Using sleep here because there is no other synchronization
# source that would notify us when all sent packets were received
time.sleep(duration + 1)
rcv_packets = server1.get_udp_listener_results(listen_handle)
snt_packets = server2.get_udp_sender_results(send_handle)
loss = math.ceil(((snt_packets - rcv_packets) * 10000.0) /
snt_packets) / 100
bwidth = (rcv_packets * size) / duration
print(udpformat.format(size, snt_packets, rcv_packets,
'%.2f%%' % loss, util.bandwidth_to_string(bwidth)))
finally:
if listen_handle != NO_HANDLE:
server1.close_udp_listener(listen_handle)
if send_handle != NO_HANDLE:
server2.close_udp_sender(send_handle)
print("\n")
def do_tcp_tests(receiver, sender, duration):
"""Schedule TCP tests between receiver and sender"""
server1 = util.rpc_client(receiver[0], receiver[1])
server2 = util.rpc_client(sender[0], sender[1])
tcpformat = '{0:>15} {1:>15} {2:>15}'
print("TCP test from %s:%u to %s:%u (full speed)" % (sender[0], sender[1],
receiver[0], receiver[1]))
print(tcpformat.format("Snt Bytes", "Rcv Bytes", "Bandwidth"))
listen_handle = NO_HANDLE
send_handle = NO_HANDLE
try:
listen_handle = server1.create_tcp_listener(receiver[3])
if listen_handle == NO_HANDLE:
print("Server was unable to open TCP listening socket on port"
" %u. Try to restart the server.\n" % receiver[3])
return
send_handle = server2.create_tcp_sender(util.ip_from_cidr(receiver[2]),
receiver[3], duration)
time.sleep(duration + 1)
rcv_bytes = int(server1.get_tcp_listener_results(listen_handle))
snt_bytes = int(server2.get_tcp_sender_results(send_handle))
bwidth = rcv_bytes / duration
print(tcpformat.format(snt_bytes, rcv_bytes,
util.bandwidth_to_string(bwidth)))
finally:
if listen_handle != NO_HANDLE:
server1.close_tcp_listener(listen_handle)
if send_handle != NO_HANDLE:
server2.close_tcp_sender(send_handle)
print("\n")
def do_l3_tests(node1, node2, bandwidth, duration, ps, type):
"""
Do L3 tunneling tests. Each node is given as 4 tuple - physical
interface IP, control port, test IP and test port.
"""
server1 = util.rpc_client(node1[0], node1[1])
server2 = util.rpc_client(node2[0], node2[1])
servers_with_bridges = []
try:
server1.create_bridge(DEFAULT_TEST_BRIDGE)
servers_with_bridges.append(server1)
server2.create_bridge(DEFAULT_TEST_BRIDGE)
servers_with_bridges.append(server2)
server1.interface_up(DEFAULT_TEST_BRIDGE)
server2.interface_up(DEFAULT_TEST_BRIDGE)
server1.interface_assign_ip(DEFAULT_TEST_BRIDGE, node1[2], None)
server2.interface_assign_ip(DEFAULT_TEST_BRIDGE, node2[2], None)
server1.add_port_to_bridge(DEFAULT_TEST_BRIDGE, DEFAULT_TEST_TUN)
server2.add_port_to_bridge(DEFAULT_TEST_BRIDGE, DEFAULT_TEST_TUN)
server1.ovs_vsctl_set("Interface", DEFAULT_TEST_TUN, "type",
None, type)
server2.ovs_vsctl_set("Interface", DEFAULT_TEST_TUN, "type",
None, type)
server1.ovs_vsctl_set("Interface", DEFAULT_TEST_TUN, "options",
"remote_ip", node2[0])
server2.ovs_vsctl_set("Interface", DEFAULT_TEST_TUN, "options",
"remote_ip", node1[0])
do_udp_tests(node1, node2, bandwidth, duration, ps)
do_udp_tests(node2, node1, bandwidth, duration, ps)
do_tcp_tests(node1, node2, duration)
do_tcp_tests(node2, node1, duration)
finally:
for server in servers_with_bridges:
server.del_bridge(DEFAULT_TEST_BRIDGE)
def do_vlan_tests(node1, node2, bandwidth, duration, ps, tag):
"""
Do VLAN tests between node1 and node2. Each node is given
as 4 tuple - physical interface IP, control port, test IP and
test port.
"""
server1 = util.rpc_client(node1[0], node1[1])
server2 = util.rpc_client(node2[0], node2[1])
br_name1 = None
br_name2 = None
servers_with_test_ports = []
try:
interface_node1 = server1.get_interface(node1[0])
interface_node2 = server2.get_interface(node2[0])
if server1.is_ovs_bridge(interface_node1):
br_name1 = interface_node1
else:
br_name1 = DEFAULT_TEST_BRIDGE
server1.create_test_bridge(br_name1, interface_node1)
if server2.is_ovs_bridge(interface_node2):
br_name2 = interface_node2
else:
br_name2 = DEFAULT_TEST_BRIDGE
server2.create_test_bridge(br_name2, interface_node2)
server1.add_port_to_bridge(br_name1, DEFAULT_TEST_PORT)
servers_with_test_ports.append(server1)
server2.add_port_to_bridge(br_name2, DEFAULT_TEST_PORT)
servers_with_test_ports.append(server2)
server1.ovs_vsctl_set("Port", DEFAULT_TEST_PORT, "tag", None, tag)
server2.ovs_vsctl_set("Port", DEFAULT_TEST_PORT, "tag", None, tag)
server1.ovs_vsctl_set("Interface", DEFAULT_TEST_PORT, "type", None,
"internal")
server2.ovs_vsctl_set("Interface", DEFAULT_TEST_PORT, "type", None,
"internal")
server1.interface_assign_ip(DEFAULT_TEST_PORT, node1[2], None)
server2.interface_assign_ip(DEFAULT_TEST_PORT, node2[2], None)
server1.interface_up(DEFAULT_TEST_PORT)
server2.interface_up(DEFAULT_TEST_PORT)
do_udp_tests(node1, node2, bandwidth, duration, ps)
do_udp_tests(node2, node1, bandwidth, duration, ps)
do_tcp_tests(node1, node2, duration)
do_tcp_tests(node2, node1, duration)
finally:
for server in servers_with_test_ports:
server.del_port_from_bridge(DEFAULT_TEST_PORT)
if br_name1 == DEFAULT_TEST_BRIDGE:
server1.del_test_bridge(br_name1, interface_node1)
if br_name2 == DEFAULT_TEST_BRIDGE:
server2.del_test_bridge(br_name2, interface_node2)
def do_direct_tests(node1, node2, bandwidth, duration, ps):
"""
Do tests between outer IPs without involving Open vSwitch. Each
node is given as 4 tuple - physical interface IP, control port,
test IP and test port. Direct tests will use physical interface
IP as the test IP address.
"""
n1 = (node1[0], node1[1], node1[0], node1[3])
n2 = (node2[0], node2[1], node2[0], node2[3])
do_udp_tests(n1, n2, bandwidth, duration, ps)
do_udp_tests(n2, n1, bandwidth, duration, ps)
do_tcp_tests(n1, n2, duration)
do_tcp_tests(n2, n1, duration)
def configure_l3(conf, tunnel_mode):
"""
This function creates a temporary test bridge and adds an L3 tunnel.
"""
s = util.start_local_server(conf[1][1])
server = util.rpc_client("127.0.0.1", conf[1][1])
server.create_bridge(DEFAULT_TEST_BRIDGE)
server.add_port_to_bridge(DEFAULT_TEST_BRIDGE, DEFAULT_TEST_PORT)
server.interface_up(DEFAULT_TEST_BRIDGE)
server.interface_assign_ip(DEFAULT_TEST_BRIDGE, conf[1][0],
None)
server.ovs_vsctl_set("Interface", DEFAULT_TEST_PORT, "type",
None, tunnel_mode)
server.ovs_vsctl_set("Interface", DEFAULT_TEST_PORT, "options",
"remote_ip", conf[0])
return s
Reference in New Issue View Git Blame Copy Permalink