Add EnvironmentService class to list environments

Address other PR feedback

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
 

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,

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

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

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)