[FDB]: Add FDB test to validate VLAN FDB entries

ansible/roles/test/files/ptftests/fdb.py

@@ -0,0 +1,22 @@
+from ipaddress import ip_address, ip_network
+
+class Fdb():
+    def __init__(self, file_path):
+
+        self._arp_dict = {}
+        self._vlan_dict = {}
+
+        with open(file_path, 'r') as f:
+            for line in f.readlines():
+                entry = line.split(' ', 1)
+                prefix = ip_network(unicode(entry[0]))
+                self._vlan_dict[prefix] = [int(i) for i in entry[1].split()]
+
+    def insert(self, mac, member):
+        self._arp_dict[member] = mac
+
+    def get_vlan_table(self):
+        return self._vlan_dict
+
+    def get_arp_table(self):
+        return self._arp_dict

ansible/roles/test/files/ptftests/fdb_test.py

@@ -0,0 +1,92 @@
+import fdb
+import json
+import logging
+import subprocess
+
+from collections import defaultdict
+from ipaddress import ip_address, ip_network
+
+import ptf
+import ptf.packet as scapy
+import ptf.dataplane as dataplane
+
+from ptf import config
+from ptf.base_tests import BaseTest
+from ptf.testutils import *
+
+class FdbTest(BaseTest):
+    def __init__(self):
+        BaseTest.__init__(self)
+        self.test_params = test_params_get()
+    #--------------------------------------------------------------------------
+
+    def log(self, message):
+        logging.info(message)
+    #--------------------------------------------------------------------------
+
+    def shell(self, cmds):
+        sp = subprocess.Popen(cmds, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        stdout, stderr = sp.communicate()
+        rc = sp.returncode
+
+        return stdout, stderr, rc
+    #--------------------------------------------------------------------------
+
+    def setUp(self):
+        self.dataplane = ptf.dataplane_instance
+        self.fdb = fdb.Fdb(self.test_params['fdb_info'])
+        self.vlan_ip = ip_address(unicode(self.test_params['vlan_ip']))
+
+        self.setUpFdb()
+        self.setUpArpResponder()
+
+        self.log("Start arp_responder")
+        self.shell(["supervisorctl", "start", "arp_responder"])
+    #--------------------------------------------------------------------------
+
+    def tearDown(self):
+        self.log("Stop arp_responder")
+        self.shell(["supervisorctl", "stop", "arp_responder"])
+    #--------------------------------------------------------------------------
+
+    def setUpFdb(self):
+        vlan_table = self.fdb.get_vlan_table()
+        for vlan in vlan_table:
+            for member in vlan_table[vlan]:
+                mac = self.dataplane.get_mac(0, member)
+                self.fdb.insert(mac, member)
+    #--------------------------------------------------------------------------
+
+    def setUpArpResponder(self):
+        vlan_table = self.fdb.get_vlan_table()
+        arp_table = self.fdb.get_arp_table()
+        d = defaultdict(list)
+        for vlan in vlan_table:
+            network = ip_network(vlan)
+            length = int(network[-1]) - int(network[0])
+            index = 1
+            for member in vlan_table[vlan]:
+                iface = "eth%d" % member
+                index = index + 1 if network[index + 1] != self.vlan_ip else index + 2
+                d[iface].append(str(network[index]))
+        with open('/tmp/from_t1.json', 'w') as file:
+            json.dump(d, file)
+    #--------------------------------------------------------------------------
+
+    def check_route(self, src_mac, dst_mac, src_port, dst_port):
+        pkt = simple_eth_packet(eth_dst=dst_mac,
+                                eth_src=src_mac,
+                                eth_type=0x1234)
+        self.log("Send packet " + str(src_mac) + "->" + str(dst_mac) + " from " + str(src_port) + " to " + str(dst_port) + "...")
+        send(self, src_port, pkt)
+        verify_packet(self, pkt, dst_port)
+    #--------------------------------------------------------------------------
+
+    def runTest(self):
+        vlan_table = self.fdb.get_vlan_table()
+        arp_table = self.fdb.get_arp_table()
+        for vlan in vlan_table:
+            for src in vlan_table[vlan]:
+                for dst in [i for i in vlan_table[vlan] if i != src]:
+                    self.check_route(arp_table[src], arp_table[dst], src, dst)
+    #--------------------------------------------------------------------------

ansible/roles/test/tasks/fdb.yml

@@ -0,0 +1,62 @@
+- fail: msg="testbed_type is not defined"
+  when: testbed_type is not defined
+
+- fail: msg="testbed_type {{test_type}} is invalid"
+  when: testbed_type not in ['t0']
+
+- include_vars: "vars/topo_{{testbed_type}}.yml"
+
+- name: Expand properties into props
+  set_fact: props="{{configuration_properties['common']}}"
+  when: testbed_type in ['t0']
+
+- name: Gather minigraph facts about the device
+  minigraph_facts: host={{inventory_hostname}}
+  connection: local
+
+- name: Remove existing IPs from PTF host
+  script: roles/test/files/helpers/remove_ip.sh
+  delegate_to: "{{ptf_host}}"
+
+- name: Set unique MACs to PTF interfaces
+  script: roles/test/files/helpers/change_mac.sh
+  delegate_to: "{{ptf_host}}"
+
+- name: Copy tests to PTF
+  copy: src=roles/test/files/ptftests dest=/root
+  delegate_to: "{{ptf_host}}"
+
+- name: Copy ARP responder to PTF
+  copy: src=roles/test/files/helpers/arp_responder.py dest=/opt
+  delegate_to: "{{ptf_host}}"
+
+- name: Copy ARP responder supervisor configuration to PTF
+  copy: src=roles/test/files/supervisor/arp_responder.conf dest=/etc/supervisor/conf.d
+  delegate_to: "{{ptf_host}}"
+
+- name: Reread supervisor configuration
+  shell: /usr/bin/supervisorctl reread
+  delegate_to: "{{ptf_host}}"
+
+- name: Update supervisor configuration
+  shell: /usr/bin/supervisorctl update
+  delegate_to: "{{ ptf_host }}"
+
+- name: Copy FDB information file to PTF
+  template: src=roles/test/templates/fdb.j2 dest=/root/fdb_info.txt
+  delegate_to: "{{ ptf_host }}"
+
+- name: "Start PTF runner"
+  include: ptf_runner.yml
+  vars:
+    ptf_test_name: FDB test
+    ptf_test_dir: ptftests
+    ptf_test_path: fdb_test.FdbTest
+    ptf_platform: remote
+    ptf_test_params:
+      - testbed_type='{{testbed_type}}'
+      - router_mac='{{ansible_Ethernet0['macaddress']}}'
+      - fdb_info='/root/fdb_info.txt'
+      - vlan_ip='{{minigraph_vlan_interfaces[0]['addr']}}'
+    ptf_extra_options: "--relax --debug info --log-file /tmp/fdb_test.FdbTest.{{ansible_date_time.iso8601}}.log "
+

ansible/roles/test/tasks/sonic.yml

@@ -116,6 +116,12 @@
   include: fib.yml
   tags: fib
 
+### When calling this FDB test, please add command line of what testbed_type and which PTF docker to test agains
+### -e "testbed_type=t0 ptf_host=10.64.246.22"
+- name: FDB test
+  include: fdb.yml
+  tags: fdb
+
 ### When calling this decap test, please add command line of what testbed_type, dscp_mode, and which PTF docker to test against
 #### -e "testbed_type=t1-lag dscp_mode=pipe ptf_host=10.0.0.200"
 - name: Decap test

ansible/roles/test/templates/fdb.j2

@@ -0,0 +1,3 @@
+{% for vlan in minigraph_vlan_interfaces %}
+{{ vlan['subnet'] }} {% for ifname in minigraph_vlans[vlan['attachto']]['members'] %} {{ minigraph_port_indices[ifname] }} {% endfor %}
+{% endfor %}