[warm-reboot] Add lag flap check after warm boot

ansible/roles/test/files/ptftests/advanced-reboot.py

@@ -777,15 +777,19 @@ def wait_for_ssh_threads():
             if self.reboot_type == 'fast-reboot' and no_cp_replies < 0.95 * self.nr_vl_pkts:
                 self.fails['dut'].add("Dataplane didn't route to all servers, when control-plane was down: %d vs %d" % (no_cp_replies, self.nr_vl_pkts))
 
-            if self.reboot_type == 'warm-reboot' and self.preboot_oper is not None:
-                if self.pre_handle is not None:
+            if self.reboot_type == 'warm-reboot':
+                if self.preboot_oper is not None and self.pre_handle is not None:
                     self.log("Postboot checks:")
                     log_info, fails = self.pre_handle.verify(pre_check=False)
                     self.populate_fail_info(fails)
                     for log in log_info:
                         self.log(log)
                     self.log(" ")
 
+                else:
+                    # verify there are no interface flaps after warm boot
+                    self.neigh_lag_status_check()
+
         except Exception as e:
             self.fails['dut'].add(e)
         finally:
@@ -864,6 +868,21 @@ def wait_for_ssh_threads():
 
             self.assertTrue(is_good, errors)
 
+    def neigh_lag_status_check(self):
+        """
+        Ensure there are no interface flaps after warm-boot
+        """
+        for neigh in self.ssh_targets:
+            self.neigh_handle = Arista(neigh, None, self.test_params)
+            self.neigh_handle.connect()
+            fails, is_flap, flap_cnt = self.neigh_handle.verify_neigh_lag_no_flap()
+            self.neigh_handle.disconnect()
+            self.fails[neigh] |= fails
+            if not is_flap:
+                self.log("No LAG flaps seen on %s after warm boot" % neigh)
+            else:
+                self.fails[neigh].add("LAG flapped %s times on %s after warm boot" % (flap_cnt, neigh))
+
     def extract_no_cpu_replies(self, arr):
       """
       This function tries to extract number of replies from dataplane, when control plane is non working

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

@@ -137,6 +137,9 @@ def run(self):
             sample["po_changetime"] = json.loads(portchannel_output, strict=False)['interfaces']['Port-Channel1']['lastStatusChangeTimestamp']
 
             if not run_once:
+                # clear Portchannel counters
+                self.do_cmd("clear counters Port-Channel 1")
+
                 self.ipv4_gr_enabled, self.ipv6_gr_enabled, self.gr_timeout = self.parse_bgp_neighbor_once(bgp_neig_output)
                 if self.gr_timeout is not None:
                     log_first_line = "session_begins_%f" % cur_time
@@ -423,6 +426,25 @@ def verify_neigh_lag_state(self, lag, state="connected", pre_check=True):
         self.fails.add('%s: Invalid interface name' % msg_prefix[pre_check])
         return self.fails, lag_state
 
+    def verify_neigh_lag_no_flap(self):
+        is_flap = True
+        flap_cnt = -1
+        output = self.do_cmd('show interfaces Po1 | json')
+        if 'Invalid' not in output:
+            data = '\n'.join(output.split('\r\n')[1:-1])
+            obj = json.loads(data)
+
+            if 'interfaces' in obj and 'Port-Channel1' in obj['interfaces']:
+                intf_cnt_info = obj['interfaces']['Port-Channel1']['interfaceCounters']
+                flap_cnt = intf_cnt_info['linkStatusChanges']
+                is_flap = (flap_cnt != 0)
+            else:
+                self.fails.add('Object missing in output for Port-Channel1')
+            return self.fails, is_flap, flap_cnt
+
+        self.fails.add('Invalid interface name - Po1')
+        return self.fails, is_flap, flap_cnt
+
     def check_gr_peer_status(self, output):
         # [0] True 'ipv4_gr_enabled', [1] doesn't matter 'ipv6_enabled', [2] should be >= 120
         if not self.ipv4_gr_enabled: