tests: refactor pcap file deletion to improve robustness

Type: test

Change-Id: I504c079126bd8b33c5e217a1b9086788a8c778e5
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
This commit is contained in:
Dave Wallace
2023-11-27 20:23:03 -05:00
committed by Florin Coras
parent 4b08632748
commit 87a8826d17
2 changed files with 23 additions and 16 deletions

View File

@ -815,6 +815,26 @@ class VppAsfTestCase(CPUInterface, unittest.TestCase):
"""Allow subclass specific teardown logging additions."""
self.logger.info("--- No test specific show commands provided. ---")
def unlink_testcase_file(self, path):
MAX_ATTEMPTS = 9
retries = MAX_ATTEMPTS
while retries > 0:
retries = retries - 1
self.logger.debug(f"Unlinking {path}")
try:
path.unlink()
# Loop until unlink() fails with FileNotFoundError to ensure file is removed
except FileNotFoundError:
break
except OSError:
self.logger.debug(f"OSError: unlinking {path}")
self.sleep(0.25, f"{retries} retries left")
if retries == 0 and os.path.isfile(path):
self.logger.error(
f"Unable to delete testcase file in {MAX_ATTEMPTS} attempts: {path}"
)
raise OSError
def tearDown(self):
"""Show various debug prints after each test"""
self.logger.debug(
@ -853,17 +873,7 @@ class VppAsfTestCase(CPUInterface, unittest.TestCase):
if hasattr(self, "pg_interfaces") and len(self.pg_interfaces) > 0:
testcase_dir = os.path.dirname(self.pg_interfaces[0].out_path)
for p in Path(testcase_dir).glob("pg*.pcap"):
retries = 8
while retries > 0:
retries = retries - 1
self.logger.debug(f"Unlinking {p}")
try:
p.unlink()
break
except OSError:
self.logger.debug(f"OSError: unlinking {p}")
self.sleep(0.25, f"{retries} retries left")
self.unlink_testcase_file(p)
self.logger.debug(
f"--- END tearDown() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
)

View File

@ -172,11 +172,8 @@ class VppPGInterface(VppInterface):
)
def remove_old_pcap_file(self, path):
try:
self.test.logger.debug(f"Removing {path}")
os.remove(path)
except OSError:
self.test.logger.debug(f"OSError: Could not remove {path}")
self.wait_for_pg_stop()
self.test.unlink_testcase_file(self.test, Path(path))
return
def decode_pcap_files(self, pcap_dir, filename_prefix):