feat: Add warning message about release version to dnf plugin

* Card ID: CCT-171
* New feature: when a release is set by subscription-manager,
  then dnf plugin "subscription-manager" prints warning about
  this release
* I'm not convinced that it is good to do that, because it
  requires to do another REST API call for any dnf command and
  it will slow down dnf even more

src/plugins/dnf/subscription_manager.py

@@ -24,6 +24,7 @@
 from subscription_manager.utils import chroot, is_simple_content_access
 from subscription_manager.injectioninit import init_dep_injection
 from subscription_manager.i18n import ungettext, ugettext as _
+from subscription_manager.release import get_consumer_release
 from rhsm import logutil
 from rhsm import config
 
@@ -70,6 +71,12 @@
 """
 )
 
+release_lock_waring = _(
+    """
+    This system has release set to {release_version} and it receives only updates for this release.
+    """
+)
+
 log = logging.getLogger("rhsm-app." + __name__)
 
 
@@ -206,6 +213,10 @@ def _warn_or_give_usage_message():
                     msg = not_registered_warning
             elif len(ent_dir.list_valid()) == 0 and not is_simple_content_access(identity=identity):
                 msg = no_subs_warning
+            else:
+                release_version = get_consumer_release(identity=identity)
+                if release_version is not None:
+                    msg = release_lock_waring.format(release_version=release_version)
         finally:
             if msg:
                 logger.info(msg)

src/subscription_manager/cli_command/release.py

@@ -20,7 +20,7 @@
 from subscription_manager.cli import system_exit
 from subscription_manager.cli_command.cli import CliCommand, conf
 from subscription_manager.i18n import ugettext as _
-from subscription_manager.release import ReleaseBackend, MultipleReleaseProductsError
+from subscription_manager.release import ReleaseBackend, MultipleReleaseProductsError, get_consumer_release
 from subscription_manager.repolib import RepoActionInvoker
 from subscription_manager.utils import parse_baseurl_info
 import subscription_manager.injection as inj
@@ -60,11 +60,13 @@ def __init__(self):
         )
 
     def _get_consumer_release(self):
-        err_msg = _("Error: The 'release' command is not supported by the server.")
-        consumer = self.cp.getConsumer(self.identity.uuid)
-        if "releaseVer" not in consumer:
-            system_exit(os.EX_UNAVAILABLE, err_msg)
-        return consumer["releaseVer"]["releaseVer"]
+        release_ver = get_consumer_release(self.identity, self.cp)
+        if release_ver is None:
+            system_exit(
+                os.EX_UNAVAILABLE,
+                _("Error: The 'release' command is not supported by the server.")
+            )
+        return release_ver
 
     def show_current_release(self):
         release = self._get_consumer_release()

src/subscription_manager/release.py

@@ -33,17 +33,33 @@
 from subscription_manager.i18n import ugettext as _
 
 if TYPE_CHECKING:
-    from rhsm.connection import ContentConnection
+    from rhsm.connection import ContentConnection, UEPConnection
     from rhsm.certificate2 import Certificate, Content, EntitlementCertificate, Product, ProductCertificate
     from subscription_manager.cp_provider import CPProvider
     from subscription_manager.certdirectory import EntitlementDirectory, ProductDirectory
+    from subscription_manager.identity import Identity
 
 
 log = logging.getLogger(__name__)
 
 cfg = rhsm.config.get_config_parser()
 
 
+def get_consumer_release(identity: Identity = None, cp: UEPConnection = None) -> Optional[str]:
+    """
+    Try to get consumer release
+    """
+    if identity is None:
+        identity = inj.require(inj.IDENTITY)
+    if cp is None:
+        cp_provider: CPProvider = inj.require(inj.CP_PROVIDER)
+        cp = cp_provider.get_consumer_auth_cp()
+    consumer = cp.getConsumer(identity.uuid)
+    if "releaseVer" not in consumer:
+        return None
+    return consumer["releaseVer"]["releaseVer"]
+
+
 class MultipleReleaseProductsError(ValueError):
     def __init__(self, certificates: Iterable["Certificate"]):
         self.certificates = certificates