Improve Simple Content Access UX (RHEL 8)

anaconda.spec.in

@@ -103,7 +103,7 @@ Requires: python3-dasbus >= %{dasbusver}
 Requires: flatpak-libs
 %if %{defined rhel} && %{undefined centos}
 Requires: python3-syspurpose
-Requires: subscription-manager >= 1.26
+Requires: subscription-manager >= 1.28.26
 %endif
 
 # pwquality only "recommends" the dictionaries it needs to do anything useful,

pyanaconda/modules/subscription/runtime.py

@@ -231,6 +231,8 @@ def run(self):
         """Register the system with Red Hat account username and password.
 
         :raises: RegistrationError if calling the RHSM DBus API returns an error
+        :return: JSON string describing registration state
+        :rtype: str
         """
         log.debug("subscription: registering with username and password")
         with RHSMPrivateBus(self._rhsm_register_server_proxy) as private_bus:
@@ -241,13 +243,16 @@ def run(self):
                 # We do not yet support setting organization for username & password
                 # registration, so organization is blank for now.
                 organization = ""
-                private_register_proxy.Register(organization,
-                                                self._username,
-                                                self._password,
-                                                {},
-                                                {},
-                                                locale)
+                registration_data = private_register_proxy.Register(
+                    organization,
+                    self._username,
+                    self._password,
+                    {},
+                    {},
+                    locale
+                )
                 log.debug("subscription: registered with username and password")
+                return registration_data
             except DBusError as e:
                 log.debug("subscription: failed to register with username and password: %s",
                           str(e))
@@ -282,19 +287,24 @@ def run(self):
         """Register the system with organization name and activation key.
 
         :raises: RegistrationError if calling the RHSM DBus API returns an error
+        :return: JSON string describing registration state
+        :rtype: str
         """
         log.debug("subscription: registering with organization and activation key")
         with RHSMPrivateBus(self._rhsm_register_server_proxy) as private_bus:
             try:
                 locale = os.environ.get("LANG", "")
                 private_register_proxy = private_bus.get_proxy(RHSM.service_name,
                                                                RHSM_REGISTER.object_path)
-                private_register_proxy.RegisterWithActivationKeys(self._organization,
-                                                                  self._activation_keys,
-                                                                  {},
-                                                                  {},
-                                                                  locale)
+                registration_data = private_register_proxy.RegisterWithActivationKeys(
+                    self._organization,
+                    self._activation_keys,
+                    {},
+                    {},
+                    locale
+                )
                 log.debug("subscription: registered with organization and activation key")
+                return registration_data
             except DBusError as e:
                 log.debug("subscription: failed to register with organization & key: %s", str(e))
                 # RHSM exception contain details as JSON due to DBus exception handling limitations

pyanaconda/modules/subscription/subscription.py

@@ -52,6 +52,7 @@
     UnregisterTask, AttachSubscriptionTask, SystemPurposeConfigurationTask, \
     ParseAttachedSubscriptionsTask
 from pyanaconda.modules.subscription.rhsm_observer import RHSMObserver
+from pyanaconda.modules.subscription.utils import detect_sca_from_registration_data
 
 
 from pykickstart.errors import KickstartParseWarning
@@ -104,6 +105,10 @@ def __init__(self):
         self.registered_changed = Signal()
         self._registered = False
 
+        # simple content access
+        self.simple_content_access_enabled_changed = Signal()
+        self._sca_enabled = False
+
         # subscription status
         self.subscription_attached_changed = Signal()
         self._subscription_attached = False
@@ -482,6 +487,28 @@ def set_registered(self, system_registered):
         self.module_properties_changed.emit()
         log.debug("System registered set to: %s", system_registered)
 
+    # simple content access status
+    @property
+    def simple_content_access_enabled(self):
+        """Return True if the system has been registered with SCA enabled.
+
+        :return: True if the system has been registered in SCA mode, False otherwise
+        :rtype: bool
+        """
+        return self._sca_enabled
+
+    def set_simple_content_access_enabled(self, sca_enabled):
+        """Set if Simple Content Access is enabled.
+
+        :param bool sca_enabled: True if SCA is enabled, False otherwise
+        """
+        self._sca_enabled = sca_enabled
+        self.simple_content_access_enabled_changed.emit()
+        # as there is no public setter in the DBus API, we need to emit
+        # the properties changed signal here manually
+        self.module_properties_changed.emit()
+        log.debug("Simple Content Access enabled set to: %s", sca_enabled)
+
     # subscription status
 
     @property
@@ -640,6 +667,13 @@ def register_username_password_with_task(self):
         # if the task succeeds, it means the system has been registered
         task.succeeded_signal.connect(
             lambda: self.set_registered(True))
+        # set SCA state based on data returned by the registration task
+        task.succeeded_signal.connect(
+            lambda: self.set_simple_content_access_enabled(
+                detect_sca_from_registration_data(task.get_result())
+            )
+        )
+
         return task
 
     def register_organization_key_with_task(self):
@@ -659,6 +693,12 @@ def register_organization_key_with_task(self):
         # if the task succeeds, it means the system has been registered
         task.succeeded_signal.connect(
             lambda: self.set_registered(True))
+        # set SCA state based on data returned by the registration task
+        task.succeeded_signal.connect(
+            lambda: self.set_simple_content_access_enabled(
+                detect_sca_from_registration_data(task.get_result())
+            )
+        )
         return task
 
     def unregister_with_task(self):
@@ -677,6 +717,10 @@ def unregister_with_task(self):
         # and clear attached subscriptions
         task.succeeded_signal.connect(
             lambda: self.set_attached_subscriptions([]))
+        # also when we are no longer registered then we are are
+        # thus no longer in Simple Content Access mode
+        task.succeeded_signal.connect(
+            lambda: self.set_simple_content_access_enabled(False))
         return task
 
     def attach_subscription_with_task(self):

pyanaconda/modules/subscription/subscription_interface.py

@@ -43,6 +43,8 @@ def connect_signals(self):
                             self.implementation.connect_to_insights_changed)
         self.watch_property("IsRegistered",
                             self.implementation.registered_changed)
+        self.watch_property("IsSimpleContentAccessEnabled",
+                            self.implementation.simple_content_access_enabled_changed)
         self.watch_property("IsSubscriptionAttached",
                             self.implementation.subscription_attached_changed)
 
@@ -144,6 +146,11 @@ def IsRegistered(self) -> Bool:
         """Report if the system is registered."""
         return self.implementation.registered
 
+    @property
+    def IsSimpleContentAccessEnabled(self) -> Bool:
+        """Report if Simple Content Access is enabled."""
+        return self.implementation.simple_content_access_enabled
+
     @property
     def IsSubscriptionAttached(self) -> Bool:
         """Report if an entitlement has been successfully attached."""

pyanaconda/modules/subscription/utils.py

@@ -0,0 +1,60 @@
+#
+# Utility functions for network module
+#
+# Copyright (C) 2022 Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+
+import json
+
+from pyanaconda.anaconda_loggers import get_module_logger
+log = get_module_logger(__name__)
+
+
+def detect_sca_from_registration_data(registration_data_json):
+    """Detect SCA/entitlement mode from registration data.
+
+    This function checks JSON data describing registration state as returned
+    by the the Register() or RegisterWithActivationKeys() RHSM DBus methods.
+    Based on the value of the "contentAccessMode" key present in a dictionary available
+    under the "owner" top level key.
+
+    :param str registration_data_json: registration data in JSON format
+    :return: True if data inicates SCA enabled, False otherwise
+    """
+    # we can't try to detect SCA mode if we don't have any registration data
+    if not registration_data_json:
+        log.warning("no registraton data provided, skipping SCA mode detection attempt")
+        return False
+    registration_data = json.loads(registration_data_json)
+    owner_data = registration_data.get("owner")
+
+    if owner_data:
+        content_access_mode = owner_data.get("contentAccessMode")
+        if content_access_mode == "org_environment":
+            # SCA explicitely noted as enabled
+            return True
+        elif content_access_mode == "entitlement":
+            # SCA explicitely not enabled
+            return False
+        else:
+            log.warning("contentAccessMode mode not set to known value:")
+            log.warning(content_access_mode)
+            # unknown mode or missing data -> not SCA
+            return False
+    else:
+        # we have no data indicating SCA is enabled
+        return False

pyanaconda/ui/gui/spokes/subscription.py

@@ -1081,7 +1081,10 @@ def _update_subscription_state(self):
         # check how many we have & set the subscription status string accordingly
         subscription_count = len(attached_subscriptions)
         if subscription_count == 0:
-            subscription_string = _("No subscriptions are attached to the system")
+            if self._subscription_module.IsSimpleContentAccessEnabled:
+                subscription_string = _("Subscribed in Simple Content Access mode.")
+            else:
+                subscription_string = _("No subscriptions are attached to the system")
         elif subscription_count == 1:
             subscription_string = _("1 subscription attached to the system")
         else:

tests/nosetests/pyanaconda_tests/module_subscription_tasks_tests.py

@@ -591,18 +591,23 @@ def username_password_success_test(self, private_bus, environ_get):
         # private register proxy
         get_proxy = private_bus.return_value.__enter__.return_value.get_proxy
         private_register_proxy = get_proxy.return_value
+        sca_enabled_json = '{"owner":{"contentAccessMode": "org_environment"}}'
+        private_register_proxy.Register.return_value = sca_enabled_json
         # instantiate the task and run it
         task = RegisterWithUsernamePasswordTask(rhsm_register_server_proxy=register_server_proxy,
                                                 username="foo_user",
                                                 password="bar_password")
-        task.run()
+        registration_data = task.run()
         # check the private register proxy Register method was called correctly
         private_register_proxy.Register.assert_called_once_with("",
                                                                 "foo_user",
                                                                 "bar_password",
                                                                 {},
                                                                 {},
                                                                 "en_US.UTF-8")
+        # check the returned registration data is properly forwarded from
+        # the (mocked) DBus call
+        self.assertEqual(sca_enabled_json, registration_data)
 
     @patch("os.environ.get", return_value="en_US.UTF-8")
     @patch("pyanaconda.modules.subscription.runtime.RHSMPrivateBus")
@@ -639,12 +644,14 @@ def org_key_success_test(self, private_bus, environ_get):
         # private register proxy
         get_proxy = private_bus.return_value.__enter__.return_value.get_proxy
         private_register_proxy = get_proxy.return_value
-        private_register_proxy.Register.return_value = True, ""
+        private_register_proxy.RegisterWithActivationKeys.return_value = True, ""
+        sca_enabled_json = '{"owner":{"contentAccessMode": "org_environment"}}'
+        private_register_proxy.RegisterWithActivationKeys.return_value = sca_enabled_json
         # instantiate the task and run it
         task = RegisterWithOrganizationKeyTask(rhsm_register_server_proxy=register_server_proxy,
                                                organization="123456789",
                                                activation_keys=["foo", "bar", "baz"])
-        task.run()
+        registration_data = task.run()
         # check private register proxy RegisterWithActivationKeys method was called correctly
         private_register_proxy.RegisterWithActivationKeys.assert_called_with(
             "123456789",
@@ -653,6 +660,9 @@ def org_key_success_test(self, private_bus, environ_get):
             {},
             'en_US.UTF-8'
         )
+        # check the returned registration data is properly forwarded from
+        # the (mocked) DBus call
+        self.assertEqual(sca_enabled_json, registration_data)
 
     @patch("os.environ.get", return_value="en_US.UTF-8")
     @patch("pyanaconda.modules.subscription.runtime.RHSMPrivateBus")

tests/nosetests/pyanaconda_tests/module_subscription_tests.py

@@ -927,6 +927,29 @@ def custom_setter(value):
         # at the end the property should be True
         self.assertTrue(self.subscription_interface.IsRegistered)
 
+    def simple_content_access_property_test(self):
+        """Test the IsSimpleContentAccessEnabled property."""
+        # should be false by default
+        self.assertFalse(self.subscription_interface.IsSimpleContentAccessEnabled)
+
+        # this property can't be set by client as it is set as the result of
+        # subscription attempts, so we need to call the internal module interface
+        # via a custom setter
+
+        def custom_setter(value):
+            self.subscription_module.set_simple_content_access_enabled(value)
+
+        # check the property is True and the signal was emitted
+        # - we use fake setter as there is no public setter
+        self._check_dbus_property(
+          "IsSimpleContentAccessEnabled",
+          True,
+          setter=custom_setter
+        )
+
+        # at the end the property should be True
+        self.assertTrue(self.subscription_interface.IsSimpleContentAccessEnabled)
+
     def subscription_attached_property_test(self):
         """Test the IsSubscriptionAttached property."""
         # should be false by default
@@ -1246,10 +1269,16 @@ def register_with_username_password_test(self, publisher):
         self.assertEqual(obj.implementation._rhsm_register_server_proxy, register_server_proxy)
         self.assertEqual(obj.implementation._username, "foo_user")
         self.assertEqual(obj.implementation._password, "foo_password")
+        # mock a result
+        obj.implementation.get_result = Mock()
+        sca_json = '{"owner":{"contentAccessMode": "org_environment"}}'
+        obj.implementation.get_result.return_value = sca_json
         # trigger the succeeded signal
         obj.implementation.succeeded_signal.emit()
         # check this set the registered property to True
         self.assertTrue(self.subscription_interface.IsRegistered)
+        # check the SCA property is True due to the JSON data saying so
+        self.assertTrue(self.subscription_interface.IsSimpleContentAccessEnabled)
 
     @patch_dbus_publish_object
     def register_with_organization_key_test(self, publisher):
@@ -1285,16 +1314,24 @@ def register_with_organization_key_test(self, publisher):
         self.assertEqual(obj.implementation._rhsm_register_server_proxy, register_server_proxy)
         self.assertEqual(obj.implementation._organization, "123456789")
         self.assertEqual(obj.implementation._activation_keys, ["key1", "key2", "key3"])
+        # mock a result
+        obj.implementation.get_result = Mock()
+        sca_json = '{"owner":{"contentAccessMode": "org_environment"}}'
+        obj.implementation.get_result.return_value = sca_json
         # trigger the succeeded signal
         obj.implementation.succeeded_signal.emit()
         # check this set the registered property to True
         self.assertTrue(self.subscription_interface.IsRegistered)
+        # check the SCA property is True due to the JSON data saying so
+        self.assertTrue(self.subscription_interface.IsSimpleContentAccessEnabled)
 
     @patch_dbus_publish_object
     def unregister_test(self, publisher):
         """Test UnregisterTask creation."""
         # simulate system being subscribed
         self.subscription_module.set_subscription_attached(True)
+        # simulate system being in SCA mode
+        self.subscription_module.set_simple_content_access_enabled(True)
         # make sure the task gets dummy rhsm unregister proxy
         observer = Mock()
         self.subscription_module._rhsm_observer = observer
@@ -1306,9 +1343,11 @@ def unregister_test(self, publisher):
         self.assertEqual(obj.implementation._rhsm_unregister_proxy, rhsm_unregister_proxy)
         # trigger the succeeded signal
         obj.implementation.succeeded_signal.emit()
-        # check this set the subscription-attached & registered properties to False
+        # check this set the subscription-attached, registered and
+        # simple content access properties to False
         self.assertFalse(self.subscription_interface.IsRegistered)
         self.assertFalse(self.subscription_interface.IsSubscriptionAttached)
+        self.assertFalse(self.subscription_interface.IsSimpleContentAccessEnabled)
 
     @patch_dbus_publish_object
     def attach_subscription_test(self, publisher):