Skip to content

Commit

Permalink
fix: Disabling of content in D-Bus methods Register*()
Browse files Browse the repository at this point in the history
* CCT-1005
* When D-Bus method Register() or RegisterWithActivationKeys() is
  called with "enable_content": false, rhsm.service will not
  download and install SCA cerificate, but it is not enough,
  because later rhsmcertd or "subscription-manager status"
  will download SCA certificate and it will generated redhat.repo
  file
* For this reason, when "enable_content" is false, then
  manage_repos will be set to 0 to be sure that redhat.repo
  will not be generated later
* Fixed unit tests
* TODO: Do not download SCA certificate, when manage_repos is
  equal to "0", because it is useless activity and it
  consumes resources on candlepin server
  • Loading branch information
jirihnidek committed Jan 23, 2025
1 parent cb97911 commit 6a3f82d
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 1 deletion.
19 changes: 19 additions & 0 deletions src/rhsmlib/dbus/objects/register.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
from rhsmlib.services.unregister import UnregisterService
from rhsmlib.services.entitlement import EntitlementService
from rhsmlib.services.environment import EnvironmentService
from rhsmlib.services.config import Config
from rhsmlib.client_info import DBusSender
from subscription_manager.cp_provider import CPProvider

Expand All @@ -34,6 +35,7 @@

if TYPE_CHECKING:
from rhsm.connection import UEPConnection
from rhsm.config import RhsmConfigParser

log = logging.getLogger(__name__)

Expand Down Expand Up @@ -180,6 +182,9 @@ class DomainSocketRegisterDBusImplementation(base_object.BaseImplementation):
where a D-Bus signal is emitted.
"""

def __init__(self, parser: "RhsmConfigParser" = None):
self.config = Config(parser)

def get_organizations(self, options: dict) -> List[dict]:
"""Get user account organizations.
Expand Down Expand Up @@ -244,6 +249,8 @@ def register_with_credentials(
# When consumer is created, we can try to enable content, if requested.
if enable_content:
self._enable_content(uep, consumer)
else:
self._disable_content()

return consumer

Expand All @@ -268,6 +275,8 @@ def register_with_activation_keys(
# When consumer is created, we can try to enable content, if requested.
if enable_content is True:
self._enable_content(uep, consumer)
else:
self._disable_content()

return consumer

Expand Down Expand Up @@ -316,6 +325,16 @@ def _enable_content(uep: "UEPConnection", consumer: dict) -> None:
else:
log.error(f"Unable to enable content due to unsupported content access mode: '{content_access}'")

def _disable_content(self) -> None:
"""
Try to disable generating content from SCA certificates. This method
sets value manage_repos to 0 in rhsm.conf to avoid generating redhat.repo
file from SCA certificates by subscription-manager or rhsmcertd.
"""
log.info("Disabling generating content (redhat.repo) from SCA certificate in rhsm.conf.")
self.config["rhsm"]["manage_repos"] = "0"
self.config.persist()

def _check_force_handling(self, register_options: dict, connection_options: dict) -> None:
"""
Handles "force=true" in the registration options
Expand Down
24 changes: 23 additions & 1 deletion test/rhsmlib/dbus/test_register.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@
from rhsm import connection
import rhsmlib.dbus.exceptions
from rhsmlib.dbus.objects.register import DomainSocketRegisterDBusImplementation, RegisterDBusImplementation
from rhsm.config import RhsmConfigParser
from rhsmlib.dbus.objects.config import ConfigDBusImplementation

Check warning on line 24 in test/rhsmlib/dbus/test_register.py

View workflow job for this annotation

GitHub Actions / black & flake8 & rpmlint

F401 'rhsmlib.dbus.objects.config.ConfigDBusImplementation' imported but unused
from test.rhsmlib.services.test_config import TEST_CONFIG
from rhsmlib.dbus.server import DomainSocketServer

from unittest import mock
Expand Down Expand Up @@ -275,9 +278,28 @@ def test_Stop__not_running(self):


class DomainSocketRegisterDBusObjectTest(SubManDBusFixture):
config_file = None
"""Attribute referencing file containing test configuration text."""
parser = None
"""Attribute referencing configuration's parser object."""

@classmethod
def setUpClass(cls) -> None:
super().setUpClass()

cls.config_file = tempfile.NamedTemporaryFile()
with open(cls.config_file.name, "w") as handle:
handle.write(TEST_CONFIG)
cls.parser = RhsmConfigParser(cls.config_file.name)

@classmethod
def tearDownClass(cls) -> None:
cls.config_file = None
super().tearDownClass()

def setUp(self) -> None:
super().setUp()
self.impl = DomainSocketRegisterDBusImplementation()
self.impl = DomainSocketRegisterDBusImplementation(self.parser)

register_patch = mock.patch(
"rhsmlib.dbus.objects.register.RegisterService.register",
Expand Down

0 comments on commit 6a3f82d

Please sign in to comment.