import sys
import json
from functools import partial
+from time import sleep
import pytest
# Save the Current Working Directory to find configuration files.
# set up PE bridges with the EVPN member interfaces facing the CE hosts
pe1.run("ip link add name br101 type bridge stp_state 0")
+ pe1.run("ip addr add 10.10.1.1/24 dev br101")
pe1.run("ip link set dev br101 up")
pe1.run(
"ip link add vxlan101 type vxlan id 101 dstport 4789 local 10.10.10.10 nolearning"
pe1.run("ip link set dev PE1-eth0 master br101")
pe2.run("ip link add name br101 type bridge stp_state 0")
+ pe2.run("ip addr add 10.10.1.3/24 dev br101")
pe2.run("ip link set dev br101 up")
pe2.run(
"ip link add vxlan101 type vxlan id 101 dstport 4789 local 10.30.30.30 nolearning"
mac_output = local.vtysh_cmd("show evpn mac vni 101 mac {} json".format(mac))
mac_output_json = json.loads(mac_output)
assertmsg = "Local MAC output does not match interface mac {}".format(mac)
- assert mac_output_json[mac]["type"] == "local"
+ assert mac_output_json[mac]["type"] == "local", assertmsg
def mac_test_local_remote(local, remote):
# Memory leak test template
+def ip_learn_test(tgen, host, local, remote, ip_addr):
+ "check the host IP gets learned by the VNI"
+ host_output = host.vtysh_cmd("show interface {}-eth0".format(host.name))
+ int_lines = host_output.splitlines()
+ mac_line = int_lines[7].split(": ")
+ mac = mac_line[1]
+ print(host_output)
+
+ # check we have a local association between the MAC and IP
+ local_output = local.vtysh_cmd("show evpn mac vni 101 mac {} json".format(mac))
+ print(local_output)
+ local_output_json = json.loads(local_output)
+ mac_type = local_output_json[mac]["type"]
+ assertmsg = "Failed to learn local IP address on host {}".format(host.name)
+ assert local_output_json[mac]["neighbors"] != "none", assertmsg
+ learned_ip = local_output_json[mac]["neighbors"]["active"][0]
+
+ assertmsg = "local learned mac wrong type: {} ".format(mac_type)
+ assert mac_type == "local", assertmsg
+
+ assertmsg = "learned address mismatch with configured address host: {} learned: {}".format(
+ ip_addr, learned_ip
+ )
+ assert ip_addr == learned_ip, assertmsg
+
+ # now lets check the remote
+ count = 0
+ converged = False
+ while count < 30:
+ remote_output = remote.vtysh_cmd(
+ "show evpn mac vni 101 mac {} json".format(mac)
+ )
+ print(remote_output)
+ remote_output_json = json.loads(remote_output)
+ type = remote_output_json[mac]["type"]
+ if not remote_output_json[mac]["neighbors"] == "none":
+ # due to a kernel quirk, learned IPs can be inactive
+ if (
+ remote_output_json[mac]["neighbors"]["active"]
+ or remote_output_json[mac]["neighbors"]["inactive"]
+ ):
+ converged = True
+ break
+ count += 1
+ sleep(1)
+
+ print("tries: {}".format(count))
+ assertmsg = "{} remote learned mac no address: {} ".format(host.name, mac)
+ # some debug for this failure
+ if not converged == True:
+ log_output = remote.run("cat zebra.log")
+ print(log_output)
+
+ assert converged == True, assertmsg
+ if remote_output_json[mac]["neighbors"]["active"]:
+ learned_ip = remote_output_json[mac]["neighbors"]["active"][0]
+ else:
+ learned_ip = remote_output_json[mac]["neighbors"]["inactive"][0]
+ assertmsg = "remote learned mac wrong type: {} ".format(type)
+ assert type == "remote", assertmsg
+
+ assertmsg = "remote learned address mismatch with configured address host: {} learned: {}".format(
+ ip_addr, learned_ip
+ )
+ assert ip_addr == learned_ip, assertmsg
+
+
+def test_ip_pe1_learn():
+ "run the IP learn test for PE1"
+
+ tgen = get_topogen()
+ host1 = tgen.gears["host1"]
+ pe1 = tgen.gears["PE1"]
+ pe2 = tgen.gears["PE2"]
+ pe2.vtysh_cmd("debug zebra vxlan")
+ pe2.vtysh_cmd("debug zebra kernel")
+ # lets populate that arp cache
+ host1.run("ping -c1 10.10.1.1")
+ ip_learn_test(tgen, host1, pe1, pe2, "10.10.1.55")
+ # tgen.mininet_cli()
+
+
+def test_ip_pe2_learn():
+ "run the IP learn test for PE2"
+
+ tgen = get_topogen()
+ host2 = tgen.gears["host2"]
+ pe1 = tgen.gears["PE1"]
+ pe2 = tgen.gears["PE2"]
+ pe1.vtysh_cmd("debug zebra vxlan")
+ pe1.vtysh_cmd("debug zebra kernel")
+ # lets populate that arp cache
+ host2.run("ping -c1 10.10.1.3")
+ ip_learn_test(tgen, host2, pe2, pe1, "10.10.1.56")
+ # tgen.mininet_cli()
+
+
def test_memory_leak():
"Run the memory leak test and report results."
tgen = get_topogen()