[gNOI] Add support for gNOI Factory Reset

host_modules/gnoi_reset.py

@@ -0,0 +1,147 @@
+"""gNOI reset module which performs factory reset."""
+
+import json
+import logging
+import threading
+import time
+from host_modules import host_service
+from host_modules.reboot import Reboot
+
+MOD_NAME = "gnoi_reset"
+
+logger = logging.getLogger(__name__)
+
+
+class GnoiReset(host_service.HostModule):
+    """DBus endpoint that executes the factory reset and returns the reset status and response."""
+
+    def __init__(self, mod_name):
+        self.lock = threading.Lock()
+        self.is_reset_ongoing = False
+        self.reset_request = {}
+        self.reset_response = {}
+        super(GnoiReset, self).__init__(mod_name)
+
+    def populate_reset_response(
+        self,
+        reset_success=True,
+        factory_os_unsupported=False,
+        zero_fill_unsupported=False,
+        detail="",
+    ) -> tuple[int, str]:
+        """Populate the factory reset response.
+        """
+        with self.lock:
+            self.reset_response = {}
+            response = {}
+            if reset_success:
+                self.reset_response["reset_success"] = {}
+                response["reset_success"] = {}
+            else:
+                self.reset_response["reset_error"] = {}
+                response["reset_error"] = {}
+                if factory_os_unsupported:
+                    self.reset_response["reset_error"]["factory_os_unsupported"] = True
+                elif zero_fill_unsupported:
+                    self.reset_response["reset_error"]["zero_fill_unsupported"] = True
+                else:
+                    self.reset_response["reset_error"]["other"] = True
+                response["reset_error"]["detail"] = detail
+            response_data = json.dumps(response)
+        return 0, response_data
+
+    def _check_reboot_in_progress(self) -> int:
+        """Checks if reboot is already in progress."""
+        if self.is_reset_ongoing:
+            return 1
+        else:
+            return 0
+
+    def _parse_arguments(self, options) -> tuple[int, str]:
+        """Parses and validates the given arguments into a reset request."""
+        try:
+            raw = json.loads(options)
+        except ValueError as e:
+            logger.error("[%s]:Failed to parse factory reset request: %s", MOD_NAME, str(e))
+            return self.populate_reset_response(
+                reset_success=False,
+                detail="Failed to parse json formatted factory reset request into python dict.",
+            )
+
+        # Normalize: support both camelCase and snake_case
+        self.reset_request = {
+            "factoryOs": raw.get("factoryOs", raw.get("factory_os", False)),
+            "zeroFill": raw.get("zeroFill", raw.get("zero_fill", False)),
+            "retainCerts": raw.get("retainCerts", raw.get("retain_certs", False)),
+        }
+
+        # Reject the request if zero_fill is set.
+        if self.reset_request["factoryOs"] and self.reset_request["zeroFill"]:
+            return self.populate_reset_response(
+                reset_success=False,
+                zero_fill_unsupported=True,
+                detail="zero_fill operation is currently unsupported.",
+            )
+        # Issue a warning if retain_certs is set.
+        if self.reset_request["factoryOs"] and self.reset_request["retainCerts"]:
+            logger.warning("%s: retain_certs is currently ignored.", MOD_NAME)
+            return self.populate_reset_response(
+                reset_success=False,
+                detail="Method FactoryReset.Start is currently unsupported."
+            )
+        # Reject the request if factoryOs is set. As the method is currently unsupported 
+        if self.reset_request["factoryOs"]:
+            return self.populate_reset_response(
+                reset_success=False,
+                detail="Method FactoryReset.Start is currently unsupported."
+            )
+
+        # Default fallback if no valid options triggered any action
+        return self.populate_reset_response(
+            reset_success=False,
+            detail="Method FactoryReset.Start is currently unsupported."
+        )
+
+    def _execute_reboot(self) -> int:
+        try:
+            r = Reboot("reboot")
+            t = threading.Thread(target=r.execute_reboot, args=("COLD",))
+            t.start()
+        except RuntimeError:
+            self.is_reset_ongoing = False
+            return 1
+
+        return 0
+
+    @host_service.method(
+        host_service.bus_name(MOD_NAME), in_signature="as", out_signature="is"
+    )
+
+    def issue_reset(self, options) -> tuple[int, str]:
+        """Issues the factory reset."""
+        print("Issuing reset from back end")
+
+        rc, resp = self._parse_arguments(options)
+        if not rc:
+            return rc, resp
+
+        rc = self._check_reboot_in_progress()
+        if rc:
+            return self.populate_reset_response(reset_success=False, detail="Previous reset is ongoing.")
+
+        self.is_reset_ongoing = True
+
+        rc, resp = self._execute_reboot()
+        if rc:
+            return self.populate_reset_response(reset_success=False,detail="Failed to start thread to execute reboot.")
+
+        # Default fallback if no valid options triggered any action
+        return self.populate_reset_response(
+            reset_success=False,
+            detail="Method FactoryReset.Start is currently unsupported."
+        )
+
+
+def register():
+    """Return the class name"""
+    return GnoiReset, MOD_NAME

scripts/sonic-host-server

@@ -22,7 +22,8 @@ from host_modules import (
     image_service,
     docker_service,
     reboot,
-    debug_service
+    debug_service,
+    gnoi_reset
 )
 
 
@@ -38,8 +39,9 @@ def register_dbus():
         'image_service': image_service.ImageService('image_service'),
         'docker_service': docker_service.DockerService('docker_service'),
         'file_stat': file_service.FileService('file'),
-        'debug_service': debug_service.DebugExecutor('DebugExecutor')
-    }
+        'debug_service': debug_service.DebugExecutor('DebugExecutor'),
+        'gnoi_reset': gnoi_reset.GnoiReset('gnoi_reset')
+        }
     for mod_name, handler_class in mod_dict.items():
         handlers[mod_name] = handler_class
 

tests/gnoi_reset_test.py

@@ -0,0 +1,125 @@
+"""Tests for gnoi_reset."""
+
+import imp
+import os
+import sys
+import logging
+
+if sys.version_info >= (3, 3):
+    from unittest import mock
+else:
+    import mock
+
+test_path = os.path.dirname(os.path.abspath(__file__))
+sonic_host_service_path = os.path.dirname(test_path)
+host_modules_path = os.path.join(sonic_host_service_path, "host_modules")
+sys.path.insert(0, sonic_host_service_path)
+
+# Input requests based on gNOI CLI examples (snake_case and camelCase mix)
+VALID_RESET_REQUEST = '{"factory_os": true}'
+ZERO_FILL_REQUEST = '{"factory_os": true, "zero_fill": true}'
+RETAIN_CERTS_REQUEST = '{"factory_os": true, "retainCerts": true}'
+INVALID_RESET_REQUEST = '"factory_os": true, "zero_fill": true'
+
+imp.load_source("host_service", host_modules_path + "/host_service.py")
+imp.load_source("gnoi_reset", host_modules_path + "/gnoi_reset.py")
+from gnoi_reset import *
+
+
+class TestGnoiReset:
+    @classmethod
+    def setup_class(cls):
+        with mock.patch("gnoi_reset.super"):
+            cls.gnoi_reset_module = GnoiReset(MOD_NAME)
+
+    def test_zero_fill_unsupported(self):
+        result = self.gnoi_reset_module.issue_reset(ZERO_FILL_REQUEST)
+        assert result[0] == 0
+        assert result[1] == (
+            '{"reset_error": {"detail": "zero_fill operation is currently unsupported."}}'
+        )
+
+    def test_retain_certs_warning(self, caplog):
+        with caplog.at_level(logging.WARNING):
+            result = self.gnoi_reset_module.issue_reset(RETAIN_CERTS_REQUEST)
+            assert (
+                caplog.records[0].message
+                == "gnoi_reset: retain_certs is currently ignored."
+            )
+            assert result[0] == 0
+            assert result[1] == (
+                '{"reset_error": {"detail": "Method FactoryReset.Start is currently unsupported."}}'
+            )
+
+    def test_invalid_json_format(self):
+        result = self.gnoi_reset_module.issue_reset(INVALID_RESET_REQUEST)
+        assert result[0] == 0
+        assert result[1] == (
+            '{"reset_error": {"detail": "Failed to parse json formatted factory reset request into python dict."}}'
+        )
+
+    def test_valid_request_unimplemented(self):
+        result = self.gnoi_reset_module.issue_reset(VALID_RESET_REQUEST)
+        assert result[0] == 0
+        assert result[1] == (
+            '{"reset_error": {"detail": "Method FactoryReset.Start is currently unsupported."}}'
+        )
+
+    def test_populate_reset_response_success(self):
+        _, response = self.gnoi_reset_module.populate_reset_response(reset_success=True)
+        assert response == '{"reset_success": {}}'
+
+    def test_populate_reset_response_other_error(self):
+        _, response = self.gnoi_reset_module.populate_reset_response(
+            reset_success=False,
+            detail="Generic failure."
+        )
+        assert response == (
+            '{"reset_error": {"detail": "Generic failure."}}'
+        )
+
+    def test_check_reboot_in_progress_true(self):
+        self.gnoi_reset_module.is_reset_ongoing = True
+        rc = self.gnoi_reset_module._check_reboot_in_progress()
+        assert rc == 1
+
+    def test_check_reboot_in_progress_false(self):
+        self.gnoi_reset_module.is_reset_ongoing = False
+        rc = self.gnoi_reset_module._check_reboot_in_progress()
+        assert rc == 0
+
+    @mock.patch("gnoi_reset.Reboot")
+    @mock.patch("threading.Thread")
+    def test_execute_reboot_success(self, mock_thread_cls, mock_reboot_cls):
+        mock_reboot_instance = mock.Mock()
+        mock_reboot_cls.return_value = mock_reboot_instance
+
+        mock_thread_instance = mock.Mock()
+        mock_thread_cls.return_value = mock_thread_instance
+
+        rc = self.gnoi_reset_module._execute_reboot()
+
+        mock_reboot_cls.assert_called_once_with("reboot")
+        mock_thread_cls.assert_called_once_with(
+            target=mock_reboot_instance.execute_reboot,
+            args=("COLD",)
+        )
+        mock_thread_instance.start.assert_called_once()
+        assert rc == 0
+
+    @mock.patch("gnoi_reset.Reboot")
+    @mock.patch("threading.Thread", side_effect=RuntimeError)
+    def test_execute_reboot_runtime_error(self, mock_thread_cls, mock_reboot_cls):
+        self.gnoi_reset_module.is_reset_ongoing = True
+        rc = self.gnoi_reset_module._execute_reboot()
+        assert rc == 1
+        assert self.gnoi_reset_module.is_reset_ongoing is False
+
+    def test_register(self):
+        result = register()
+        assert result[0] == GnoiReset
+        assert result[1] == MOD_NAME
+
+    @classmethod
+    def teardown_class(cls):
+        print("TEARDOWN")