Skip to content

Commit

Permalink
feat: add GetEnvironments method to DBus register
Browse files Browse the repository at this point in the history
    * Card-ID: CCT-764
  • Loading branch information
rverdile committed Oct 1, 2024
1 parent 0d27a59 commit 2ff93a3
Show file tree
Hide file tree
Showing 2 changed files with 62 additions and 2 deletions.
4 changes: 2 additions & 2 deletions src/rhsm/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
60 changes: 60 additions & 0 deletions src/rhsmlib/dbus/objects/register.py
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,40 @@ 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' and 'password' keys
:return: List of environments
"""

orgs: List[dict] = self.get_organizations(options)

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

environments = []
for org in orgs:
org_environments: List[dict] = uep.getEnvironmentList(org["key"], list_all_environments)
return org_environments
if not org_environments:
continue

env: dict = {}
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:
Expand Down Expand Up @@ -339,6 +373,32 @@ def __init__(self, conn=None, object_path=None, bus_name=None, sender=None, cmd_
out_signature="s",
)
@util.dbus_handle_exceptions
def GetEnvironments(self, username, password, connection_options, locale):
"""
This method tries to return list of organization user belongs to. 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 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)
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)
@util.dbus_handle_exceptions
def GetOrgs(self, username, password, connection_options, locale):
"""
This method tries to return list of organization user belongs to. This method also uses
Expand Down

0 comments on commit 2ff93a3

Please sign in to comment.