[dualtor] Add neighbor entry orchagent testcase

tests/common/dualtor/dual_tor_utils.py

@@ -738,6 +738,7 @@ def _get_iface_ip(mg_facts, ifacename):
     servers = mux_cable_server_ip(standby_tor)
     random_server_iface = random.choice(servers.keys())
 
+    res['selected_port'] = random_server_iface
     res['target_server_ip'] = servers[random_server_iface]['server_ipv4'].split('/')[0]
     res['target_server_port'] = standby_tor_mg_facts['minigraph_ptf_indices'][random_server_iface]
 

tests/dualtor/test_orchagent_standby_tor_downstream.py

@@ -1,12 +1,18 @@
+import ptf
 import pytest
 import random
 import time
 import logging
 import ipaddress
+import scapy.all as scapyall
+
+from ptf import testutils, mask
 from tests.common.dualtor.dual_tor_mock import *
-from tests.common.dualtor.dual_tor_utils import dualtor_info, check_tunnel_balance, flush_neighbor
+from tests.common.dualtor.dual_tor_utils import dualtor_info, check_tunnel_balance, flush_neighbor, get_t1_ptf_ports, flush_neighbor
 from tests.common.fixtures.ptfhost_utils import copy_ptftests_directory, change_mac_addresses, run_garp_service, run_icmp_responder   # lgtm[py/unused-import]
 from tests.common.helpers.assertions import pytest_require as pt_require
+from tests.common.dualtor.tunnel_traffic_utils import tunnel_traffic_monitor
+from tests.common.dualtor.server_traffic_utils import ServerTrafficMonitor
 
 pytestmark = [
     pytest.mark.topology('t0'),
@@ -19,6 +25,7 @@
 
 logger = logging.getLogger(__file__)
 
+
 def shutdown_random_one_t1_link(dut):
     """
     Shutdown a random t1 link
@@ -185,3 +192,64 @@ def test_standby_tor_downstream_loopback_route_readded(ptfhost, rand_selected_du
     # Readd loopback routes and verify traffic is equally distributed
     add_loopback_routes(rand_selected_dut, active_tor_loopback0)
     check_tunnel_balance(**params)
+
+
+def test_standby_tor_remove_neighbor_downstream_standby(
+    conn_graph_facts, ptfadapter, ptfhost,
+    rand_selected_dut, rand_unselected_dut, tbinfo,
+    tunnel_traffic_monitor, vmhost
+):
+    """
+    @summary: Verify that after removing neighbor entry for a server over standby
+    ToR, the packets sent to the server will be dropped(neither passed to the server
+    or redirected to the active ToR).
+    """
+    def build_packet_to_server(tor, ptfadapter, target_server_ip, tunnel_traffic_monitor):
+        """Build packet destinated to server."""
+        pkt_dscp = random.choice(range(0, 33))
+        pkt_ttl = random.choice(range(3, 65))
+        pkt = testutils.simple_ip_packet(
+            eth_dst=tor.facts["router_mac"],
+            eth_src=ptfadapter.dataplane.get_mac(0, 0),
+            ip_src="1.1.1.1",
+            ip_dst=target_server_ip,
+            ip_dscp=pkt_dscp,
+            ip_ttl=pkt_ttl
+        )
+        logging.info(
+            "the packet destinated to server %s:\n%s", target_server_ip,
+            tunnel_traffic_monitor._dump_show_str(pkt)
+        )
+        return pkt
+
+    def build_expected_packet_to_server(packet):
+        """Build expected mask packet downstream to server."""
+        exp_pkt = mask.Mask(packet)
+        exp_pkt.set_do_not_care_scapy(scapyall.Ether, "dst")
+        exp_pkt.set_do_not_care_scapy(scapyall.Ether, "src")
+        exp_pkt.set_do_not_care_scapy(scapyall.IP, "tos")
+        exp_pkt.set_do_not_care_scapy(scapyall.IP, "ttl")
+        exp_pkt.set_do_not_care_scapy(scapyall.IP, "chksum")
+        return exp_pkt
+
+    tor = rand_selected_dut
+    test_params = dualtor_info(ptfhost, rand_selected_dut, rand_unselected_dut, tbinfo)
+    server_ipv4 = test_params["target_server_ip"]
+
+    pkt = build_packet_to_server(tor, ptfadapter, server_ipv4, tunnel_traffic_monitor)
+    exp_pkt = build_expected_packet_to_server(pkt)
+    ptf_t1_intf = random.choice(get_t1_ptf_ports(tor, tbinfo))
+    logging.info("send traffic to server %s from ptf t1 interface %s", server_ipv4, ptf_t1_intf)
+    tunnel_monitor = tunnel_traffic_monitor(tor, existing=True)
+    with tunnel_monitor:
+        testutils.send(ptfadapter, int(ptf_t1_intf.strip("eth")), pkt, count=10)
+
+    logging.info("send traffic to server %s after removing neighbor entry", server_ipv4)
+    tunnel_monitor.existing = False
+    with flush_neighbor(tor, server_ipv4):
+        with tunnel_monitor:
+            with ServerTrafficMonitor(tor, vmhost,
+                                      test_params["selected_port"],
+                                      conn_graph_facts, exp_pkt,
+                                      existing=False):
+                testutils.send(ptfadapter, int(ptf_t1_intf.strip("eth")), pkt, count=10)