Consolidate the get running config way.

generic_config_updater/change_applier.py

@@ -9,6 +9,7 @@
 from swsscommon.swsscommon import ConfigDBConnector
 from sonic_py_common import multi_asic
 from .gu_common import GenericConfigUpdaterError, genericUpdaterLogging
+from .gu_common import get_config_db_as_json
 
 SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
 UPDATER_CONF_FILE = f"{SCRIPT_DIR}/gcu_services_validator.conf.json"
@@ -161,29 +162,4 @@ def remove_backend_tables_from_config(self, data):
             data.pop(key, None)
 
     def _get_running_config(self):
-        _, fname = tempfile.mkstemp(suffix="_changeApplier")
-
-        if self.scope:
-            cmd = ['sonic-cfggen', '-d', '--print-data', '-n', self.scope]
-        else:
-            cmd = ['sonic-cfggen', '-d', '--print-data']
-
-        with open(fname, "w") as file:
-            result = subprocess.Popen(cmd, stdout=file, stderr=subprocess.PIPE, text=True)
-            _, err = result.communicate()
-
-        return_code = result.returncode
-        if return_code:
-            os.remove(fname)
-            raise GenericConfigUpdaterError(
-                f"Failed to get running config for scope: {self.scope}," +
-                f"Return code: {return_code}, Error: {err}")
-
-        run_data = {}
-        try:
-            with open(fname, "r") as file:
-                run_data = json.load(file)
-        finally:
-            if os.path.isfile(fname):
-                os.remove(fname)
-        return run_data
+        return get_config_db_as_json(self.scope)

generic_config_updater/gu_common.py

@@ -53,31 +53,39 @@ def __eq__(self, other):
             return self.patch == other.patch
         return False
 
+
+def get_config_db_as_json(scope=None):
+    text = get_config_db_as_text(scope=scope)
+    config_db_json = json.loads(text)
+    config_db_json.pop("bgpraw", None)
+    return config_db_json
+
+
+def get_config_db_as_text(scope=None):
+    if scope is not None and scope != multi_asic.DEFAULT_NAMESPACE:
+        cmd = ['sonic-cfggen', '-d', '--print-data', '-n', scope]
+    else:
+        cmd = ['sonic-cfggen', '-d', '--print-data']
+    result = subprocess.Popen(cmd, shell=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    text, err = result.communicate()
+    return_code = result.returncode
+    if return_code:
+        raise GenericConfigUpdaterError(f"Failed to get running config for namespace: {scope},"
+                                        f" Return code: {return_code}, Error: {err}")
+    return text
+
+
 class ConfigWrapper:
     def __init__(self, yang_dir=YANG_DIR, scope=multi_asic.DEFAULT_NAMESPACE):
         self.scope = scope
         self.yang_dir = YANG_DIR
         self.sonic_yang_with_loaded_models = None
 
     def get_config_db_as_json(self):
-        text = self._get_config_db_as_text()
-        config_db_json = json.loads(text)
-        config_db_json.pop("bgpraw", None)
-        return config_db_json
+        return get_config_db_as_json(self.scope)
 
     def _get_config_db_as_text(self):
-        if self.scope is not None and self.scope != multi_asic.DEFAULT_NAMESPACE:
-            cmd = ['sonic-cfggen', '-d', '--print-data', '-n', self.scope]
-        else:
-            cmd = ['sonic-cfggen', '-d', '--print-data']
-
-        result = subprocess.Popen(cmd, shell=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        text, err = result.communicate()
-        return_code = result.returncode
-        if return_code: # non-zero means failure
-            raise GenericConfigUpdaterError(f"Failed to get running config for namespace: {self.scope},"
-                                            f" Return code: {return_code}, Error: {err}")
-        return text
+        return get_config_db_as_text(self.scope)
 
     def get_sonic_yang_as_json(self):
         config_db_json = self.get_config_db_as_json()

tests/generic_config_updater/change_applier_test.py

@@ -72,28 +72,25 @@
 def debug_print(msg):
     print(msg)
 
-
-# Mimics os.system call for sonic-cfggen -d --print-data > filename
+# Mimics os.system call for `sonic-cfggen -d --print-data` output
 def subprocess_Popen_cfggen(cmd, *args, **kwargs):
     global running_config
 
-    # Extract file name from kwargs if 'stdout' is a file object
-    stdout = kwargs.get('stdout')
-    if hasattr(stdout, 'name'):
-        fname = stdout.name
+    stdout = kwargs.get('stdout', None)
+
+    if stdout is None:
+        output = json.dumps(running_config, indent=4)
+    elif isinstance(stdout, int) and stdout == -1:
+        output = json.dumps(running_config, indent=4)
     else:
-        raise ValueError("stdout is not a file")
+        raise ValueError("stdout must be set to subprocess.PIPE or omitted for capturing output")
 
-    # Write the running configuration to the file specified in stdout
-    with open(fname, "w") as s:
-        json.dump(running_config, s, indent=4)
-
     class MockPopen:
         def __init__(self):
-            self.returncode = 0  # Simulate successful command execution
+            self.returncode = 0
 
         def communicate(self):
-            return "", ""  # Simulate empty stdout and stderr
+            return output.encode(), "".encode()
 
     return MockPopen()
 
@@ -225,7 +222,7 @@ def vlan_validate(old_cfg, new_cfg, keys):
 
 class TestChangeApplier(unittest.TestCase):
 
-    @patch("generic_config_updater.change_applier.subprocess.Popen")
+    @patch("generic_config_updater.gu_common.subprocess.Popen")
     @patch("generic_config_updater.change_applier.get_config_db")
     @patch("generic_config_updater.change_applier.set_config")
     def test_change_apply(self, mock_set, mock_db, mock_subprocess_Popen):