feat: add GetEnvironments method to DBus register

    * Card-ID: CCT-764

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
 

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",

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