Add unit test for _wait_until_clear

sonic-net · Nov 22, 2022 · 6f3a879 · 6f3a879
1 parent 0985385
commit 6f3a879
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 3 deletions.
diff --git a/config/main.py b/config/main.py
@@ -728,20 +728,21 @@ def storm_control_delete_entry(port_name, storm_type):
     return True
 
 
-def wait_until_clear(interval=0.5, timeout=30):
+def _wait_until_clear(table, interval=0.5, timeout=30):
     start = time.time()
     empty = False
     app_db = SonicV2Connector(host='127.0.0.1')
     app_db.connect(app_db.APPL_DB)
 
     while not empty and time.time() - start < timeout:
-        current_profiles = app_db.keys(app_db.APPL_DB, "BUFFER_POOL_TABLE:*")
+        current_profiles = app_db.keys(app_db.APPL_DB, table)
         if not current_profiles:
             empty = True
         else:
             time.sleep(interval)
     if not empty:
         click.echo("Operation not completed successfully, please save and reload configuration.")
+    return empty
 
 
 def _clear_qos(delay = False):
@@ -781,7 +782,7 @@ def _clear_qos(delay = False):
         for qos_table in QOS_TABLE_NAMES:
             config_db.delete_table(qos_table)
     if delay:
-        wait_until_clear(interval=0.5, timeout=30)
+        _wait_until_clear("BUFFER_POOL_TABLE:*",interval=0.5, timeout=30)
 
 def _get_sonic_generated_services(num_asic):
     if not os.path.isfile(SONIC_GENERATED_SERVICE_PATH):

diff --git a/tests/config_test.py b/tests/config_test.py
@@ -668,6 +668,28 @@ def setup_class(cls):
         import config.main
         importlib.reload(config.main)
 
+    def _keys(args, kwargs):
+        if not TestConfigQos._keys_counter:
+            return []
+        TestConfigQos._keys_counter-=1
+        return ["BUFFER_POOL_TABLE:egress_lossy_pool"]
+
+    def test_qos_wait_until_clear_empty(self):
+        from config.main import _wait_until_clear
+
+        with mock.patch('swsscommon.swsscommon.SonicV2Connector.keys',  side_effect=TestConfigQos._keys):
+            TestConfigQos._keys_counter = 1
+            empty = _wait_until_clear("BUFFER_POOL_TABLE:*", 0.5,2)
+        assert empty
+
+    def test_qos_wait_until_clear_not_empty(self):
+        from config.main import _wait_until_clear
+
+        with mock.patch('swsscommon.swsscommon.SonicV2Connector.keys', side_effect=TestConfigQos._keys):
+            TestConfigQos._keys_counter = 10
+            empty = _wait_until_clear("BUFFER_POOL_TABLE:*", 0.5,2)
+        assert not empty
+
     def test_qos_reload_single(
             self, get_cmd_module, setup_qos_mock_apis,
             setup_single_broadcom_asic