Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Really disable content #3500

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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:
self._enable_content(uep, consumer)
else:
self._disable_content()

return consumer

Expand Down Expand Up @@ -313,6 +322,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
23 changes: 22 additions & 1 deletion test/rhsmlib/dbus/test_register.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
from rhsm import connection
import rhsmlib.dbus.exceptions
from rhsmlib.dbus.objects.register import DomainSocketRegisterDBusImplementation, RegisterDBusImplementation
from rhsm.config import RhsmConfigParser
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 +277,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
Loading