Add dbus support for below commands:

config apply-patch
config checkpoint
config delete checkpoint
config reload

Author: ganglyu

host_modules/config_engine.py

@@ -0,0 +1,55 @@
+"""Config command handler"""
+
+from host_modules import host_service
+import subprocess
+import os
+import shutil
+
+MOD_NAME = 'config'
+DEFAULT_CONFIG = '/etc/sonic/config_db.json'
+
+class Config(host_service.HostModule):
+    """
+    DBus endpoint that executes the config command
+    """
+    @host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='is')
+    def reload(self, config_file):
+
+        cmd = ['/usr/local/bin/config', 'reload', '-y']
+        if config_file and config_file != DEFAULT_CONFIG:
+            if not os.path.exists(config_file):
+                return -1, "Can't find %s"%config_file
+            # Persistent Config
+            shutil.move(config_file, DEFAULT_CONFIG)
+
+        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        msg = ''
+        if result.returncode:
+            lines = result.stderr.decode().split('\n')
+            for line in lines[::-1]:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+    @host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='is')
+    def save(self, config_file):
+
+        cmd = ['/usr/local/bin/config', 'save', '-y']
+        if config_file and config_file != DEFAULT_CONFIG:
+            cmd.append(config_file)
+
+        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        msg = ''
+        if result.returncode:
+            lines = result.stderr.decode().split('\n')
+            for line in lines[::-1]:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+def register():
+    """Return the class name"""
+    return Config, MOD_NAME
+

host_modules/gcu.py

@@ -0,0 +1,75 @@
+"""Generic config updater command handler"""
+
+from host_modules import host_service
+import subprocess
+
+MOD_NAME = 'gcu'
+
+class GCU(host_service.HostModule):
+    """
+    DBus endpoint that executes the generic config updater command
+    """
+    @host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='is')
+    def apply_patch_db(self, patch_file_path):
+
+        cmd = ['/usr/local/bin/config', 'apply-patch', '-f', 'CONFIGDB', patch_file_path]
+
+        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        msg = ''
+        if result.returncode:
+            lines = result.stderr.decode().split('\n')
+            for line in lines[::-1]:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+    @host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='is')
+    def apply_patch_yang(self, patch_file_path):
+
+        cmd = ['/usr/local/bin/config', 'apply-patch', '-f', 'SONICYANG', patch_file_path]
+
+        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        msg = ''
+        if result.returncode:
+            lines = result.stderr.decode().split('\n')
+            for line in lines[::-1]:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+    @host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='is')
+    def create_checkpoint(self, checkpoint_file):
+
+        cmd = ['/usr/local/bin/config', 'checkpoint', checkpoint_file]
+
+        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        msg = ''
+        if result.returncode:
+            lines = result.stderr.decode().split('\n')
+            for line in lines[::-1]:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+    @host_service.method(host_service.bus_name(MOD_NAME), in_signature='s', out_signature='is')
+    def delete_checkpoint(self, checkpoint_file):
+
+        cmd = ['/usr/local/bin/config', 'delete-checkpoint', checkpoint_file]
+
+        result = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        msg = ''
+        if result.returncode:
+            lines = result.stderr.decode().split('\n')
+            for line in lines[::-1]:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+def register():
+    """Return the class name"""
+    return GCU, MOD_NAME
+

host_modules/showtech.py

@@ -1,6 +1,6 @@
 """Show techsupport command handler"""
 
-import host_service
+from host_modules import host_service
 import subprocess
 import re
 

pytest.ini

@@ -1,2 +1,2 @@
 [pytest]
-addopts = --cov=scripts --cov-report html --cov-report term --cov-report xml --ignore=tests/*/test*_vectors.py --junitxml=test-results.xml -vv
+addopts = --cov=scripts --cov=host_modules --cov-report html --cov-report term --cov-report xml --ignore=tests/*/test*_vectors.py --junitxml=test-results.xml -vv

tests/host_modules/__init__.py

@@ -0,0 +1,79 @@
+import sys
+import os
+import pytest
+from unittest import mock
+from host_modules import config_engine
+
+class TestConfigEngine(object):
+    @mock.patch("dbus.SystemBus")
+    @mock.patch("dbus.service.BusName")
+    @mock.patch("dbus.service.Object.__init__")
+    def test_reload(self, MockInit, MockBusName, MockSystemBus):
+        with mock.patch("os.path.exists") as mock_exists, mock.patch("shutil.move") as mock_move:
+            mock_exists.return_value=True
+            with mock.patch("subprocess.run") as mock_run:
+                res_mock = mock.Mock()
+                test_ret = 0
+                test_msg = b"Error: this is the test message\nHello world\n"
+                attrs = {"returncode": test_ret, "stderr": test_msg}
+                res_mock.configure_mock(**attrs)
+                mock_run.return_value = res_mock
+                config_file = "test.json"
+                config_stub = config_engine.Config(config_engine.MOD_NAME)
+                ret, msg = config_stub.reload(config_file)
+                call_args = mock_run.call_args[0][0]
+                assert "reload" in call_args
+                assert config_file not in call_args
+                assert ret == test_ret, "Return value is wrong"
+                assert msg == "", "Return message is wrong"
+            with mock.patch("subprocess.run") as mock_run:
+                res_mock = mock.Mock()
+                test_ret = 1
+                test_msg = b"Error: this is the test message\nHello world\n"
+                attrs = {"returncode": test_ret, "stderr": test_msg}
+                res_mock.configure_mock(**attrs)
+                mock_run.return_value = res_mock
+                config_file = "test.json"
+                config_stub = config_engine.Config(config_engine.MOD_NAME)
+                ret, msg = config_stub.reload(config_file)
+                call_args = mock_run.call_args[0][0]
+                assert "reload" in call_args
+                assert config_file not in call_args
+                assert ret == test_ret, "Return value is wrong"
+                assert msg == "Error: this is the test message", "Return message is wrong"
+
+    @mock.patch("dbus.SystemBus")
+    @mock.patch("dbus.service.BusName")
+    @mock.patch("dbus.service.Object.__init__")
+    def test_save(self, MockInit, MockBusName, MockSystemBus):
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 0
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            config_file = "test.patch"
+            config_stub = config_engine.Config(config_engine.MOD_NAME)
+            ret, msg = config_stub.save(config_file)
+            call_args = mock_run.call_args[0][0]
+            assert "save" in call_args
+            assert config_file in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "", "Return message is wrong"
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 1
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            config_file = "test.patch"
+            config_stub = config_engine.Config(config_engine.MOD_NAME)
+            ret, msg = config_stub.save(config_file)
+            call_args = mock_run.call_args[0][0]
+            assert "save" in call_args
+            assert config_file in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "Error: this is the test message", "Return message is wrong"
+

tests/host_modules/config_test.py

@@ -0,0 +1,153 @@
+import sys
+import os
+import pytest
+from unittest import mock
+from host_modules import gcu
+
+class TestGCU(object):
+    @mock.patch("dbus.SystemBus")
+    @mock.patch("dbus.service.BusName")
+    @mock.patch("dbus.service.Object.__init__")
+    def test_apply_patch_db(self, MockInit, MockBusName, MockSystemBus):
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 0
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            patch_file = "test.patch"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_db(patch_file)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "CONFIGDB" in call_args
+            assert patch_file in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "", "Return message is wrong"
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 1
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            patch_file = "test.patch"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_db(patch_file)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "CONFIGDB" in call_args
+            assert patch_file in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "Error: this is the test message", "Return message is wrong"
+
+    @mock.patch("dbus.SystemBus")
+    @mock.patch("dbus.service.BusName")
+    @mock.patch("dbus.service.Object.__init__")
+    def test_apply_patch_yang(self, MockInit, MockBusName, MockSystemBus):
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 0
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            patch_file = "test.patch"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_yang(patch_file)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "SONICYANG" in call_args
+            assert patch_file in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "", "Return message is wrong"
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 1
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            patch_file = "test.patch"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_yang(patch_file)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "SONICYANG" in call_args
+            assert patch_file in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "Error: this is the test message", "Return message is wrong"
+
+    @mock.patch("dbus.SystemBus")
+    @mock.patch("dbus.service.BusName")
+    @mock.patch("dbus.service.Object.__init__")
+    def test_create_checkpoint(self, MockInit, MockBusName, MockSystemBus):
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 0
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            cp_name = "test_name"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.create_checkpoint(cp_name)
+            call_args = mock_run.call_args[0][0]
+            assert "checkpoint" in call_args
+            assert "delete-checkpoint" not in call_args
+            assert cp_name in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "", "Return message is wrong"
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 1
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            cp_name = "test_name"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.create_checkpoint(cp_name)
+            call_args = mock_run.call_args[0][0]
+            assert "checkpoint" in call_args
+            assert "delete-checkpoint" not in call_args
+            assert cp_name in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "Error: this is the test message", "Return message is wrong"
+
+    @mock.patch("dbus.SystemBus")
+    @mock.patch("dbus.service.BusName")
+    @mock.patch("dbus.service.Object.__init__")
+    def test_delete_checkpoint(self, MockInit, MockBusName, MockSystemBus):
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 0
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            cp_name = "test_name"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.delete_checkpoint(cp_name)
+            call_args = mock_run.call_args[0][0]
+            assert "delete-checkpoint" in call_args
+            assert cp_name in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "", "Return message is wrong"
+        with mock.patch("subprocess.run") as mock_run:
+            res_mock = mock.Mock()
+            test_ret = 1
+            test_msg = b"Error: this is the test message\nHello world\n"
+            attrs = {"returncode": test_ret, "stderr": test_msg}
+            res_mock.configure_mock(**attrs)
+            mock_run.return_value = res_mock
+            cp_name = "test_name"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.delete_checkpoint(cp_name)
+            call_args = mock_run.call_args[0][0]
+            assert "delete-checkpoint" in call_args
+            assert cp_name in call_args
+            assert ret == test_ret, "Return value is wrong"
+            assert msg == "Error: this is the test message", "Return message is wrong"
+