From f438b2e81cefaf05490ef4ebda2a3d59b764e59d Mon Sep 17 00:00:00 2001 From: rverdile Date: Tue, 1 Oct 2024 15:27:18 -0400 Subject: [PATCH] feat: add GetEnvironments method to DBus register * Card-ID: CCT-764 --- src/rhsm/connection.py | 4 +- src/rhsmlib/dbus/objects/register.py | 60 +++++++++++++++++++++++++ test/rhsmlib/dbus/test_register.py | 67 ++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+), 2 deletions(-) diff --git a/src/rhsm/connection.py b/src/rhsm/connection.py index 54302c0099..b7894e659e 100644 --- a/src/rhsm/connection.py +++ b/src/rhsm/connection.py @@ -2055,14 +2055,14 @@ def getServiceLevelList(self, owner_key: str) -> List[str]: results = self.conn.request_get(method, description=_("Fetching service levels")) return results - def getEnvironmentList(self, owner_key: str) -> List[dict]: + def getEnvironmentList(self, owner_key: str, list_all: bool = False) -> List[dict]: """ List the environments for a particular owner. Some servers may not support this and will error out. The caller can always check with supports_resource("environments"). """ - method = "/owners/%s/environments" % self.sanitize(owner_key) + method = "/owners/%s/environments?list_all=%r" % (self.sanitize(owner_key), list_all) 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 f22d4b2efd..61c9d61136 100644 --- a/src/rhsmlib/dbus/objects/register.py +++ b/src/rhsmlib/dbus/objects/register.py @@ -181,6 +181,32 @@ def get_organizations(self, options: dict) -> List[dict]: owners: List[dict] = uep.getOwnerList(options["username"]) return owners + def get_environments(self, options: dict) -> List[dict]: + """Get environments for every org belonging to user + + :param options: Connection options including the 'username', 'password', and 'org_id' keys + :return: List of environments + """ + uep: UEPConnection = self.build_uep(options, basic_auth_method=True) + status: dict = uep.getStatus() + + list_all_environments: bool = False + if "typed_environments" in status["managerCapabilities"]: + list_all_environments = True + + org_environments: List[dict] = uep.getEnvironmentList(options["org_id"], list_all_environments) + + 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 + def register_with_credentials( self, organization: Optional[str], register_options: dict, connection_options: dict ) -> dict: @@ -333,6 +359,40 @@ def __init__(self, conn=None, object_path=None, bus_name=None, sender=None, cmd_ self.sender = sender self.cmd_line = cmd_line + @dbus.service.method( + dbus_interface=constants.PRIVATE_REGISTER_INTERFACE, + in_signature="sssa{sv}s", + out_signature="s", + ) + @util.dbus_handle_exceptions + def GetEnvironments(self, username, password, org_id, connection_options, locale): + """ + This method tries to return list of environments in the given orgs. This method also uses + basic authentication method (using username and password). + + :param username: string with username used for connection to candlepin server + :param password: string with password + :param org_id: string with org id to list environments for + :param connection_options: dictionary with connection options + :param locale: string with locale + :return: string with json returned by candlepin server + """ + connection_options = dbus_utils.dbus_to_python(connection_options, expected_type=dict) + connection_options["username"] = dbus_utils.dbus_to_python(username, expected_type=str) + connection_options["password"] = dbus_utils.dbus_to_python(password, expected_type=str) + connection_options["org_id"] = dbus_utils.dbus_to_python(org_id, expected_type=str) + locale = dbus_utils.dbus_to_python(locale, expected_type=str) + + with DBusSender() as dbus_sender: + dbus_sender.set_cmd_line(sender=self.sender, cmd_line=self.cmd_line) + Locale.set(locale) + + owners = self.impl.get_environments(connection_options) + + dbus_sender.reset_cmd_line() + + return json.dumps(owners) + @dbus.service.method( dbus_interface=constants.PRIVATE_REGISTER_INTERFACE, in_signature="ssa{sv}s", diff --git a/test/rhsmlib/dbus/test_register.py b/test/rhsmlib/dbus/test_register.py index c9b5aceef7..7e6bea63a0 100644 --- a/test/rhsmlib/dbus/test_register.py +++ b/test/rhsmlib/dbus/test_register.py @@ -123,6 +123,56 @@ ] """ +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" + } +] +""" + class RegisterDBusObjectTest(SubManDBusFixture): socket_dir: Optional[tempfile.TemporaryDirectory] = None @@ -308,6 +358,23 @@ def test_GetOrgs(self): result = self.impl.get_organizations({"username": "username", "password": "password"}) self.assertEqual(expected, result) + def test_GetEnvironments(self): + self.patches["is_registered"].return_value = False + 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 + + expected = json.loads(ENVIRONMENTS_CONTENT_JSON) + result = self.impl.get_environments( + {"username": "username", "password": "password", "org_id": "org_id"} + ) + self.assertEqual(expected, result) + def test_RegisterWithActivationKeys(self): expected = json.loads(CONSUMER_CONTENT_JSON_SCA) self.patches["is_registered"].return_value = False