[sonic-host-services]: Support GCU and reload

host_modules/config_engine.py

@@ -0,0 +1,60 @@
+"""Config command handler"""
+
+from host_modules import host_service
+import subprocess
+import os
+
+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_db_json):
+
+        cmd = ['/usr/local/bin/config', 'reload', '-y']
+        config_db_json = config_db_json.strip()
+        if config_db_json and len(config_db_json):
+            config_file = '/tmp/config_db.json'
+            try:
+                if (os.path.exists(config_file)):
+                    os.remove(config_file)
+                with open(config_file, 'w') as fp:
+                    fp.write(config_db_json)
+            except Exception as err:
+                return -1, "Fail to create config file: %s"%str(err)
+            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:
+                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:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+def register():
+    """Return class and module name"""
+    return Config, MOD_NAME
+

host_modules/gcu.py

@@ -0,0 +1,87 @@
+"""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_text):
+        patch_file_path = '/tmp/config_db.patch'
+        try:
+            with open(patch_file_path, 'w') as fp:
+                fp.write(patch_text)
+        except Exception as err:
+            return -1, "Fail to create patch file: %s"%str(err)
+
+        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:
+                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_text):
+        patch_file_path = '/tmp/config_yang.patch'
+        try:
+            with open(patch_file_path, 'w') as fp:
+                fp.write(patch_text)
+        except Exception as err:
+            return -1, "Fail to create patch file: %s"%str(err)
+
+        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:
+                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:
+                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:
+                if 'Error' in line:
+                    msg = line
+                    break
+        return result.returncode, msg
+
+def register():
+    """Return class and module 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/config_test.py

@@ -0,0 +1,78 @@
+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):
+        config_file = "/tmp/config_db.json"
+        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_db_json = "{}"
+            config_stub = config_engine.Config(config_engine.MOD_NAME)
+            ret, msg = config_stub.reload(config_db_json)
+            call_args = mock_run.call_args[0][0]
+            assert "reload" 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_db_json = "{}"
+            config_stub = config_engine.Config(config_engine.MOD_NAME)
+            ret, msg = config_stub.reload(config_db_json)
+            call_args = mock_run.call_args[0][0]
+            assert "reload" 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"
+
+    @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/gcu_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_text = "{}"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_db(patch_text)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "CONFIGDB" in call_args
+            assert '/tmp/config_db.patch' 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_text = "{}"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_db(patch_text)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "CONFIGDB" in call_args
+            assert '/tmp/config_db.patch' 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_text = "{}"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_yang(patch_text)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "SONICYANG" in call_args
+            assert '/tmp/config_yang.patch' 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_text = "{}"
+            gcu_stub = gcu.GCU(gcu.MOD_NAME)
+            ret, msg = gcu_stub.apply_patch_yang(patch_text)
+            call_args = mock_run.call_args[0][0]
+            assert "apply-patch" in call_args
+            assert "SONICYANG" in call_args
+            assert '/tmp/config_yang.patch' 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"
+