2121350: Implement "force" register option in rhsm dbus python binding

src/rhsmlib/dbus/objects/register.py

@@ -16,9 +16,11 @@
 import threading
 import dbus
 import dbus.service
+import subscription_manager.injection as inj
 
 from rhsmlib.dbus import constants, exceptions, dbus_utils, base_object, server, util
 from rhsmlib.services.register import RegisterService
+from rhsmlib.services.unregister import UnregisterService
 from rhsmlib.services.attach import AttachService
 from rhsmlib.services.entitlement import EntitlementService
 from rhsmlib.client_info import DBusSender
@@ -262,6 +264,25 @@ def _remove_enable_content_option(options: dict) -> bool:
             enable_content = False
         return enable_content
 
+    def _unregister(self, connection_options: dict) -> None:
+        """
+        Unregisters a system that is currently registered and
+        cleans the cp_provider to handle authorization after un-registering.
+        :param connection_options: dictionary with connection options
+        :return: None
+        """
+        self.ensure_registered()
+        log.info("This system is already registered, attempting to un-register...")
+
+        cp_provider = inj.require(inj.CP_PROVIDER)
+
+        cp = self.build_uep(connection_options)
+        UnregisterService(cp).unregister()
+
+        # The CPProvider object must be cleaned and the cp object must
+        # be re-initialized to handle authorization after un-registration.
+        cp_provider.clean()
+
     @dbus.service.method(
         dbus_interface=constants.PRIVATE_REGISTER_INTERFACE,
         in_signature="sssa{sv}a{sv}s",
@@ -279,19 +300,26 @@ def Register(self, org, username, password, options, connection_options, locale)
 
         Note this method is registration ONLY.  Auto-attach is a separate process.
         """
-        if self.is_registered():
-            raise dbus.DBusException("This system is already registered")
-
         org = dbus_utils.dbus_to_python(org, expected_type=str)
         connection_options = dbus_utils.dbus_to_python(connection_options, expected_type=dict)
         connection_options["username"] = dbus_utils.dbus_to_python(username, expected_type=str)
         connection_options["password"] = dbus_utils.dbus_to_python(password, expected_type=str)
         options = dbus_utils.dbus_to_python(options, expected_type=dict)
         locale = dbus_utils.dbus_to_python(locale, expected_type=str)
 
+        force_registration: bool = options.get("force", False)
+        system_is_registered: bool = self.is_registered()
+        if system_is_registered and not force_registration:
+            raise dbus.DBusException("This system is already registered.")
+
         with DBusSender() as dbus_sender:
             dbus_sender.set_cmd_line(sender=self.sender, cmd_line=self.cmd_line)
             Locale.set(locale)
+
+            # If the system is registered and the 'force' option is specified,
+            # Unregister the system and register the system again.
+            if system_is_registered and force_registration:
+                self._unregister(connection_options)
             cp = self.build_uep(connection_options)
 
             register_service = RegisterService(cp)
@@ -340,9 +368,19 @@ def RegisterWithActivationKeys(self, org, activation_keys, options, connection_o
         org = dbus_utils.dbus_to_python(org, expected_type=str)
         locale = dbus_utils.dbus_to_python(locale, expected_type=str)
 
+        force_registration: bool = options.get("force", False)
+        system_is_registered: bool = self.is_registered()
+        if system_is_registered and not force_registration:
+            raise dbus.DBusException("This system is already registered.")
+
         with DBusSender() as dbus_sender:
             dbus_sender.set_cmd_line(sender=self.sender, cmd_line=self.cmd_line)
             Locale.set(locale)
+
+            # If the system is registered and the 'force' option is specified,
+            # Unregister the system and register the system again.
+            if system_is_registered and force_registration:
+                self._unregister(connection_options)
             cp = self.build_uep(connection_options)
 
             register_service = RegisterService(cp)

test/rhsmlib/dbus/test_register.py

@@ -14,6 +14,8 @@
 import json
 import mock
 import socket
+import dbus
+
 
 from rhsm import connection
 
@@ -370,6 +372,12 @@ def setUpClass(cls) -> None:
         cls.patches["register"] = register_patch.start()
         cls.addClassCleanup(register_patch.stop)
 
+        unregister_patch = mock.patch(
+            "rhsmlib.services.unregister.UnregisterService.unregister", name="unregister"
+        )
+        cls.patches["unregister"] = unregister_patch.start()
+        cls.addClassCleanup(unregister_patch.stop)
+
         is_registered_patch = mock.patch(
             "rhsmlib.dbus.base_object.BaseObject.is_registered",
             name="is_registered",
@@ -413,6 +421,26 @@ def test_Register(self):
         result = self.obj.Register.__wrapped__(self.obj, "org", "username", "password", {}, {}, self.LOCALE)
         self.assertEqual(expected, json.loads(result))
 
+    def test_Register__with_force_option(self):
+        expected = json.loads(CONSUMER_CONTENT_JSON)
+        self.patches["register"].return_value = expected
+        self.patches["unregister"].return_value = None
+        self.patches["is_registered"].return_value = True
+
+        result = self.obj.Register.__wrapped__(
+            self.obj, "org", "username", "password", {"force": True}, {}, self.LOCALE
+        )
+        self.assertEqual(expected, json.loads(result))
+
+    def test_Register__already_registered(self):
+        expected = json.loads(CONSUMER_CONTENT_JSON)
+        self.patches["register"].return_value = expected
+        self.patches["unregister"].return_value = None
+        self.patches["is_registered"].return_value = True
+
+        with self.assertRaisesRegex(dbus.DBusException, "This system is already registered."):
+            self.obj.Register.__wrapped__(self.obj, "org", "username", "password", {}, {}, self.LOCALE)
+
     def test_Register__enable_content(self):
         """Test including 'enable_content' in entitlement mode with no content."""
         expected = json.loads(CONSUMER_CONTENT_JSON)
@@ -475,3 +503,34 @@ def test_RegisterWithActivationKeys(self):
             self.LOCALE,
         )
         self.assertEqual(expected, json.loads(result))
+
+    def test_RegisterWithActivationKeys__already_registered(self):
+        expected = json.loads(CONSUMER_CONTENT_JSON)
+        self.patches["is_registered"].return_value = True
+        self.patches["register"].return_value = expected
+
+        with self.assertRaisesRegex(dbus.DBusException, "This system is already registered."):
+            self.obj.RegisterWithActivationKeys.__wrapped__(
+                self.obj,
+                "username",
+                ["key1", "key2"],
+                {},
+                {"host": "localhost", "port": "8443", "handler": "/candlepin"},
+                self.LOCALE,
+            )
+
+    def test_RegisterWithActivationKeys__with_force_option(self):
+        expected = json.loads(CONSUMER_CONTENT_JSON)
+        self.patches["is_registered"].return_value = True
+        self.patches["unregister"].return_value = None
+        self.patches["register"].return_value = expected
+
+        result = self.obj.RegisterWithActivationKeys.__wrapped__(
+            self.obj,
+            "username",
+            ["key1", "key2"],
+            {"force": True},
+            {"host": "localhost", "port": "8443", "handler": "/candlepin"},
+            self.LOCALE,
+        )
+        self.assertEqual(expected, json.loads(result))