From b4030d04c67e4c1e4b41e0b3b8544821052045e4 Mon Sep 17 00:00:00 2001 From: rverdile Date: Wed, 9 Oct 2024 16:06:39 -0400 Subject: [PATCH] Add EnvironmentService class to list environments Address other PR feedback --- src/rhsm/connection.py | 5 +- src/rhsmlib/dbus/objects/register.py | 32 ++++---- src/rhsmlib/services/environment.py | 55 +++++++++++++ test/rhsmlib/dbus/test_register.py | 55 +++---------- test/rhsmlib/services/test_environment.py | 98 +++++++++++++++++++++++ 5 files changed, 184 insertions(+), 61 deletions(-) create mode 100644 src/rhsmlib/services/environment.py create mode 100644 test/rhsmlib/services/test_environment.py diff --git a/src/rhsm/connection.py b/src/rhsm/connection.py index e9671d4ce2..163b8ef055 100644 --- a/src/rhsm/connection.py +++ b/src/rhsm/connection.py @@ -2066,7 +2066,10 @@ def getEnvironmentList(self, owner_key: str, list_all: bool = False) -> List[dic Some servers may not support this and will error out. The caller can always check with supports_resource("environments"). """ - method = "/owners/%s/environments?list_all=%r" % (self.sanitize(owner_key), list_all) + if list_all and self.has_capability("typed_environments"): + method = "/owners/%s/environments?list_all=%r" % (self.sanitize(owner_key), list_all) + else: + method = "/owners/%s/environments" % (self.sanitize(owner_key)) results = self.conn.request_get(method, description=_("Fetching environments")) return results diff --git a/src/rhsmlib/dbus/objects/register.py b/src/rhsmlib/dbus/objects/register.py index 61c9d61136..0209059a3c 100644 --- a/src/rhsmlib/dbus/objects/register.py +++ b/src/rhsmlib/dbus/objects/register.py @@ -24,6 +24,7 @@ from rhsmlib.services.register import RegisterService from rhsmlib.services.unregister import UnregisterService from rhsmlib.services.entitlement import EntitlementService +from rhsmlib.services.environment import EnvironmentService from rhsmlib.client_info import DBusSender from subscription_manager.cp_provider import CPProvider @@ -188,24 +189,21 @@ def get_environments(self, options: dict) -> List[dict]: :return: List of environments """ uep: UEPConnection = self.build_uep(options, basic_auth_method=True) - status: dict = uep.getStatus() + environment_service: EnvironmentService = EnvironmentService(uep) - list_all_environments: bool = False - if "typed_environments" in status["managerCapabilities"]: - list_all_environments = True + environments = environment_service.list(options["org_id"], typed_environments=True) - org_environments: List[dict] = uep.getEnvironmentList(options["org_id"], list_all_environments) + result: List[dict] = [] + for environment in environments: + result.append( + { + "id": environment["id"], + "name": environment["name"], + "description": environment["description"], + "type": environment["type"], + }) - env: dict = {} - environments = [] - for e in org_environments: - env["id"] = e["id"] - env["name"] = e["name"] - env["description"] = e["description"] - env["type"] = e["type"] - environments.append(env) - - return environments + return result def register_with_credentials( self, organization: Optional[str], register_options: dict, connection_options: dict @@ -387,11 +385,11 @@ def GetEnvironments(self, username, password, org_id, connection_options, locale dbus_sender.set_cmd_line(sender=self.sender, cmd_line=self.cmd_line) Locale.set(locale) - owners = self.impl.get_environments(connection_options) + environments = self.impl.get_environments(connection_options) dbus_sender.reset_cmd_line() - return json.dumps(owners) + return json.dumps(environments) @dbus.service.method( dbus_interface=constants.PRIVATE_REGISTER_INTERFACE, diff --git a/src/rhsmlib/services/environment.py b/src/rhsmlib/services/environment.py new file mode 100644 index 0000000000..1af6c04b06 --- /dev/null +++ b/src/rhsmlib/services/environment.py @@ -0,0 +1,55 @@ +# Copyright (c) 2017 Red Hat, Inc. +# +# This software is licensed to you under the GNU General Public License, +# version 2 (GPLv2). There is NO WARRANTY for this software, express or +# implied, including the implied warranties of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 +# along with this software; if not, see +# http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. +# +# Red Hat trademarks are not licensed under GPLv2. No permission is +# granted to use or replicate Red Hat trademarks that are incorporated +# in this software or its documentation. +import logging +from typing import List + + +from rhsm.connection import UEPConnection + +from subscription_manager import injection as inj + +log = logging.getLogger(__name__) + + +class EnvironmentService: + """ + Class for listing environments + """ + + def __init__(self, cp: UEPConnection) -> None: + """ + Initialization of EnvironmentService instance + :param cp: instance of connection? + """ + self.plugin_manager = inj.require(inj.PLUGIN_MANAGER) + self.cp = cp + + def list(self, org_id: str, typed_environments: bool = True) -> List[dict]: + """ + Method for listing environments + :param org_id: organization to list environments for + :param typed_environments: Whether output should include typed environments + :return: List of environments. + """ + + if typed_environments: + has_typed_environments = self.cp.has_capability("typed_environments") + + if not has_typed_environments: + log.debug("candlepin does not have typed_environments capability") + + environments = self.cp.getEnvironmentList(org_id, list_all=has_typed_environments) + else: + environments = self.cp.getEnvironmentList(org_id) + + return environments diff --git a/test/rhsmlib/dbus/test_register.py b/test/rhsmlib/dbus/test_register.py index d97560fc5f..b6dda48105 100644 --- a/test/rhsmlib/dbus/test_register.py +++ b/test/rhsmlib/dbus/test_register.py @@ -25,7 +25,6 @@ from unittest import mock from test.rhsmlib.base import SubManDBusFixture - CONSUMER_CONTENT_JSON_SCA = """{"hypervisorId": null, "serviceLevel": "", "autoheal": true, @@ -123,52 +122,24 @@ ] """ -STATUS_CONTENT_JSON = """ -{ - "mode": "NORMAL", - "modeReason": null, - "modeChangeTime": null, - "result": true, - "version": "4.4.17", - "release": "1", - "standalone": true, - "timeUTC": "2024-10-01T14:05:05+0000", - "rulesSource": "database", - "rulesVersion": "5.44", - "managerCapabilities": [ - "instance_multiplier", - "derived_product", - "vcpu", - "cert_v3", - "hypervisors_heartbeat", - "remove_by_pool_id", - "syspurpose", - "storage_band", - "cores", - "multi_environment", - "hypervisors_async", - "org_level_content_access", - "typed_environments", - "guest_limit", - "ram", - "batch_bind" - ], - "keycloakRealm": null, - "keycloakAuthUrl": null, - "keycloakResource": null, - "deviceAuthRealm": null, - "deviceAuthUrl": null, - "deviceAuthClientId": null, - "deviceAuthScope": null -} -""" - ENVIRONMENTS_CONTENT_JSON = """[ { "id": "fake-id", "name": "test-environment", "description": "test description", "type": "content-template" + }, + { + "id": "fake-id-2", + "name": "test-environment-2", + "description": "test description", + "type": "content-template" + }, + { + "id": "fake-id-3", + "name": "test-environment-3", + "description": "test description", + "type": "content-template" } ] """ @@ -363,8 +334,6 @@ def test_GetEnvironments(self): mock_cp = mock.Mock(spec=connection.UEPConnection, name="UEPConnection") mock_cp.username = "username" mock_cp.password = "password" - mock_cp.getStatus = mock.Mock() - mock_cp.getStatus.return_value = json.loads(STATUS_CONTENT_JSON) mock_cp.getEnvironmentList = mock.Mock() mock_cp.getEnvironmentList.return_value = json.loads(ENVIRONMENTS_CONTENT_JSON) self.patches["build_uep"].return_value = mock_cp diff --git a/test/rhsmlib/services/test_environment.py b/test/rhsmlib/services/test_environment.py new file mode 100644 index 0000000000..5aa4b57038 --- /dev/null +++ b/test/rhsmlib/services/test_environment.py @@ -0,0 +1,98 @@ +# Copyright (c) 2017 Red Hat, Inc. +# +# This software is licensed to you under the GNU General Public License, +# version 2 (GPLv2). There is NO WARRANTY for this software, express or +# implied, including the implied warranties of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 +# along with this software; if not, see +# http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. +# +# Red Hat trademarks are not licensed under GPLv2. No permission is +# granted to use or replicate Red Hat trademarks that are incorporated +# in this software or its documentation. +from unittest import mock + +from test.rhsmlib.base import InjectionMockingTest + +from rhsm import connection + +from rhsmlib.services import environment + + +ENVIRONMENTS_JSON = [ + { + "created": "2024-10-03T18:12:56+0000", + "updated": "2024-10-03T18:12:56+0000", + "id": "8bdf14cf9e534119a1fe617c03304768", + "name": "template 1", + "type": "content-template", + "description": "my template", + "owner": { + "id": "8ad980939253781c01925378340e0002", + "key": "content-sources-test", + "displayName": "ContentSourcesTest", + "href": "/owners/content-sources-test", + "contentAccessMode": "org_environment" + }, + "environmentContent": [ + { + "contentId": "11055", + "enabled": True + }, + { + "contentId": "56a3a98c76ea4e16bd68424a2c9cc1c1", + "enabled": True + }, + { + "contentId": "11049", + "enabled": True + }, + ] + }, + { + "created": "2024-10-09T19:08:14+0000", + "updated": "2024-10-09T19:08:14+0000", + "id": "6c62889601be41128fe2fece53141fd4", + "name": "template 2", + "type": "content-template", + "description": "my template", + "owner": { + "id": "8ad980939253781c01925378340e0002", + "key": "content-sources-test", + "displayName": "ContentSourcesTest", + "href": "/owners/content-sources-test", + "contentAccessMode": "org_environment" + }, + "environmentContent": [ + { + "contentId": "11055", + "enabled": True + }, + { + "contentId": "11049", + "enabled": True + }, + ] + } +] + + +class TestEnvironmentService(InjectionMockingTest): + def setUp(self): + super(TestEnvironmentService, self).setUp() + self.mock_cp = mock.Mock(spec=connection.UEPConnection, name="UEPConnection") + + def injection_definitions(self, *args, **kwargs): + return None + + def test_list_environments(self): + self.mock_cp.getEnvironmentList.return_value = ENVIRONMENTS_JSON + + result = environment.EnvironmentService(self.mock_cp).list("org") + self.assertEqual(ENVIRONMENTS_JSON, result) + + def test_list_environments_without_typed_environments(self): + self.mock_cp.getEnvironmentList.return_value = ENVIRONMENTS_JSON + + result = environment.EnvironmentService(self.mock_cp).list("org", typed_environments=False) + self.assertEqual(ENVIRONMENTS_JSON, result)