# Copyright 2022-2025 Ping Identity Corporation. All Rights Reserved # # This code is to be used exclusively in connection with Ping Identity # Corporation software or services. Ping Identity Corporation only offers # such software or services to legal entities who have entered into a # binding license agreement with Ping Identity Corporation. # -*- coding: utf-8 -*- # Python imports import time # load tasks used by configuration file from pyrock.tasks.deployment.datainit import MakeUsersTask from pyrock.tasks.idm.recon import TestIdmConnector, TestRCSConnection, ReconTask, ReconResultTask from pyrock.tasks.idm.general import GenerateUserGroups, DumpIDMIDWithAPITask from pyrock.tasks.scenario.gatling import GatlingTask from pyrock.tasks.scenario.ds_sdk import DSLdapModifyTask, DSModRateTask from pyrock.tasks.deployment.validation import ValidationNumUsers from pyrock.tasks.deployment.installation import DeployOverseerTask from shared.lib.components.am import AM from pyrock.lib.scheduler.tasks.StepTask import StepTask from pyrock.lib.PyRockRun import get_pyrock_run from pyrock.lib.report.json.tasks.ReportSimulation import ReportSimulation from shared.lib.platform_utils import PlatformUtils from shared.lib.utils.constants import IS_TENANT from shared.lib.utils.exception import FailException pyrock_run = get_pyrock_run() class UpdateRcsClientSecretTask(StepTask): def pre(self): if self.source != "controller": raise FailException("Task must be executed on controller") if not isinstance(self.target, AM): raise ValueError(f"Target {self.target} must be an AM component") @staticmethod def step1(): pyrock_run.log("Reset RCSClient password") PlatformUtils(components=pyrock_run.get_components()).set_rcs_client_secret() class CreateConditionalRoles(StepTask): def __init__(self, **kwargs): super().__init__(**kwargs) self.conditions = [] self.authorization_header = None self.idm = None self.departments = None def pre(self): """ Initialise the test """ self.idm = pyrock_run.get_component("idm") self.authorization_header = pyrock_run.get_component("am").get_user_oauth2_headers() self.departments = self.get_option().split(":") def step1(self): """ Generate department manager roles """ if self.departments is not None: for department in self.departments: condition = f"(/frIndexedInteger2 eq {department} and /frIndexedInteger4 eq 1)" pyrock_run.log(f"Generating condition: {condition}") self.idm.create_managed_role( headers=self.authorization_header, condition=condition, role_name=f"DepartmentManagerRole{department}", ) def step2(self): """ Generate a role for all users in a department """ if self.departments is not None: for department in self.departments: condition = f"/frIndexedInteger2 eq {department}" pyrock_run.log(f"Generating condition: {condition}") self.idm.create_managed_role( headers=self.authorization_header, condition=condition, role_name=f"DepartmentRole{department}" ) def step3(self): """ Generate department roles per employment type """ if self.departments is not None: for department in self.departments: for employeeType in ["parttime", "fulltime", "contractor"]: condition = f'(/frIndexedInteger2 eq {department} and /frIndexedMultivalued1 co "{employeeType}")' pyrock_run.log(f"Generating condition: {condition}") self.idm.create_managed_role( headers=self.authorization_header, condition=condition, role_name=f"Dpt{department}{employeeType}Role", ) class DeleteRoles(StepTask): def __init__(self, **kwargs): super().__init__(**kwargs) self.authorization_header = None self.idm = None self.role_list = None self.only_conditional = False def pre(self): """ Initialise the test """ self.idm = pyrock_run.get_component("idm") self.authorization_header = pyrock_run.get_component("am").get_user_oauth2_headers() self.only_conditional = bool(self.get_option("onlyConditional")) def step1(self): """ Get the list of existing roles """ self.role_list = self.idm.get_managed_roles(headers=self.authorization_header).json() def step2(self): """ Delete all roles from the previous step """ role_id = None if "resultCount" in self.role_list: for index in range(self.role_list["resultCount"]): if self.only_conditional and "condition" in self.role_list["result"][index]: role_id = self.role_list["result"][index]["_id"] self.idm.delete_role(headers=self.authorization_header, role_path=f"managed/alpha_role/{role_id}") elif self.only_conditional is False: self.idm.delete_role(headers=self.authorization_header, role_path=f"managed/alpha_role/{role_id}") else: pyrock_run.log("Nothing to delete") class ManagedApplicationStepTask(StepTask): def __init__(self, **kwargs): super().__init__(**kwargs) self.idm = None self.authorization_header = None self.app_id = None # The application uuid (from conf.yaml) self.role_info = None # The application to role mapping (from conf.yaml) # for report self.num_members = 0 # Number of members in a role self.elapsed_time = 0 # Time it took for the REST call to assign role to application self.num_requests = 0 # Number of HTTP (API) requests self.ok_items = 0 # How many HTTP (API) requests were processed successfully def pre(self): """ Initialise the test """ self.idm = pyrock_run.get_component("idm") self.authorization_header = pyrock_run.get_component("am").get_user_oauth2_headers(force_renew=True) self.app_id = self.get_option("application_uuid") self.role_info = self.get_option("role_info") def build_report(self, num_members=None): """ Build the results for reporting """ self.report_simulation = ReportSimulation() self.report_simulation.target_hostname = pyrock_run.get_component(self.target_name).hostname self.report_simulation.tool_name = "-" self.report_simulation.stats = {"global": {}} global_stats = self.report_simulation.stats["global"] # i.e. processed entries (name kept from Gatling-based results) global_stats["numberOfRequests"] = {"ok": self.ok_items} # one request assigns multiple members at once global_stats["duration"] = {"total": self.elapsed_time} # Here we intentionally don't use num_requests because we are interested in getting the # average throughput of members per role assigned to an application avg_throughput = num_members / self.elapsed_time avg_response_time = round((1000 / avg_throughput), ndigits=3) avg_throughput = round(avg_throughput, ndigits=3) global_stats["meanNumberOfRequestsPerSecond"] = {"total": avg_throughput} global_stats["meanResponseTime"] = {"total": avg_response_time} # "generic results" section # Here we have to use num_requests as num_members does not represent a request self.report_simulation.num_requests = self.num_requests self.report_simulation.num_requests_pass = self.ok_items self.report_simulation.num_requests_percent_pass = round(self.ok_items * 100.0 / self.num_requests, ndigits=3) self.report_simulation.avg_num_of_requests_per_second = avg_throughput self.report_simulation.avg_response_time = avg_response_time class CreateApplicationTask(ManagedApplicationStepTask): def step1(self): """ Create Managed Application """ if self.app_id is None: pyrock_run.log(f"Application uuid not found in configuration") self.set_result_fail() # Create Application for id in self.app_id: result = self.idm.create_managed_application( application_id=id, headers=self.authorization_header, template_name="ds.ldap", counter=self.app_id.index(id) + 1, ) if not result: self.set_result_fail() class DeleteApplicationTask(ManagedApplicationStepTask): def step1(self): """ Delete Managed Application """ # Clean up only in IDC if IS_TENANT: result = self.idm.delete_all_managed_applications(headers=self.authorization_header) if not result: self.set_result_fail() class AssignRoleToApplicationTask(ManagedApplicationStepTask): def step1(self): """ Assign Role(s) to Application(s) """ for app_id in self.role_info: for role_name in self.role_info[app_id]: self.num_requests += 1 self.start_time = time.time() result = self.idm.assign_application_role( application_id=app_id, role_name=role_name, headers=self.authorization_header ) # Get the cumulative elapsed time and member count because "avg_throughput" is being # calculated for all applications current_role_elapsed_time = time.time() - self.start_time self.elapsed_time += current_role_elapsed_time current_role_members_count = self.idm.get_role_member_count( role_name=role_name, headers=self.authorization_header ) self.num_members += current_role_members_count pyrock_run.log( f"Number of members in role '{role_name}' for '{app_id}' app is {current_role_members_count}, " f"assigned with mean throughput {(current_role_members_count/current_role_elapsed_time):.2f} " f"(current role) / {(self.num_members/self.elapsed_time):.2f} (total)" ) if result: self.ok_items += 1 else: self.set_result_fail() def step2(self): """build report""" if self.num_members > 0: self.build_report(self.num_members) else: pyrock_run.log(f"Skipping Result Generation because number of role members is {self.num_members}")