# Copyright 2019-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 # Framework imports from shared.lib.utils.login_session import LoginSession from pyrock.lib.PyRockRun import get_pyrock_run from pyrock.lib.scheduler.tasks.OverseerTask import OverseerTask from pyrock.lib.scheduler.tasks.StepTask import StepTask from pyrock.shared.scripts.idm.lib_utils import OrgModel from pyrock.lib.report.json.tasks.ReportSimulation import ReportSimulation from pyrock.tasks.deployment.configuration_idm import PrepareWorkloadTask from pyrock.tasks.scenario.gatling import GatlingTask pyrock_run = get_pyrock_run() class PrepareOrgHierarchyTask(OverseerTask): @property def command(self): base_url = f"{self.target.base_url}/openidm/" am = pyrock_run.get_component("am") access_token = am.get_user_token(token_type=LoginSession.ACCESS_TOKEN) while access_token.startswith("-"): access_token = am.get_user_oauth2_headers(force_renew=True) return ( f"/lodestar/pyrock/shared/scripts/idm/prepare_org_hierarchy.py --base_url {base_url} " f"--access_token {access_token} {self.get_option()}" ) class OrgModelTask(StepTask): def __init__(self, **kwargs): super().__init__(**kwargs) self.org_model: OrgModel = None def pre(self): base_url = self.target.url + "/" am = pyrock_run.get_component("am") access_token = am.get_user_token(token_type=LoginSession.ACCESS_TOKEN) while access_token.startswith("-"): access_token = am.get_user_oauth2_headers(force_renew=True) self.org_model = OrgModel(base_url=base_url, access_token=access_token) def report_result(self, parsed_options, latency): report_simulation = ReportSimulation() report_simulation.tool_name = type(self).__name__ report_simulation.target_hostname = self.target.hostname report_simulation.options = parsed_options report_simulation.num_requests = 1 report_simulation.concurrency = "N/A" report_simulation.num_requests_percent_pass = 100 report_simulation.avg_num_of_requests_per_second = 1 / latency report_simulation.avg_response_time = latency report_simulation.min_response_time = latency report_simulation.max_response_time = latency return report_simulation class AddRootOrgTask(OrgModelTask): def pre(self): super().pre() # do not put default values for this params as it could cover some issue, rather check if they are provided self.is_option("new_root_id", required=True) self.is_option("old_root_id", required=True) def step1(self): self.org_model.create_org(self.get_option("new_root_id")) def step2(self): self.start_time = time.time() self.org_model.add_new_root_org(self.get_option("new_root_id"), self.get_option("old_root_id")) def step3(self): self.end_time = time.time() self.latency = self.end_time - self.start_time report_detail = ( "Latency of adding a parent reference reference to " f"{self.org_model.org_resource_path(self.get_option('old_root_id'))}" f" in {self.org_model.org_and_members_description()}: {self.latency} seconds" ) pyrock_run.log(report_detail) def step4(self): self.report_simulation = self.report_result(self.get_options_dict(), self.latency) class AddUserToOrgTask(OrgModelTask): def pre(self): super().pre() self.is_option("user_id", required=True) self.is_option("org_id", required=True) self.is_option("relationship_field", required=True) def step1(self): self.start_time = time.time() self.org_model.create_org_user_with_org_ref( self.get_option("user_id"), self.get_option("org_id"), self.get_option("relationship_field") ) def step2(self): self.end_time = time.time() self.latency = self.end_time - self.start_time report_detail = ( f"Latency of attaching user {self.get_option('user_id')} to {self.org_model.org_resource_path(self.get_option('org_id'))} " f"via {self.get_option('relationship_field')} in {self.org_model.org_and_members_description()}: " f"{self.latency} seconds" ) pyrock_run.log(report_detail) def step3(self): self.report_simulation = self.report_result(self.get_options_dict(), self.latency) class MoveOrgTask(OrgModelTask): def pre(self): super().pre() self.is_option("depth", required=True) def step1(self): self.start_time = time.time() # Move the org through Patch self.modified_org_index, self.new_parent_index = self.org_model.move_org_of_specified_depth( self.get_option("depth"), True, False ) def step2(self): self.end_time = time.time() self.latency1 = self.end_time - self.start_time report_detail = ( f"Latency of modifying an org of depth {self.get_option('depth')}, " f"{self.org_model.org_resource_path(self.modified_org_index)} " f"to point to new org parent {self.org_model.org_resource_path(self.new_parent_index)} via " f"patch in {self.org_model.org_and_members_description()}: {self.latency1} seconds" ) pyrock_run.log(report_detail) def step3(self): self.start_time = time.time() # Move the org back through delete and create self.modified_org_index, self.new_parent_index = self.org_model.move_org_of_specified_depth( self.get_option("depth"), False, True ) def step4(self): self.end_time = time.time() self.latency2 = self.end_time - self.start_time report_detail = ( f"Latency of modifying an org of depth {self.get_option('depth')}, " f"{self.org_model.org_resource_path(self.modified_org_index)} " f"to point to new org parent {self.org_model.org_resource_path(self.new_parent_index)} via " f"delete-create on the relationship endpoint in {self.org_model.org_and_members_description()}: " f"{self.latency2} seconds" ) pyrock_run.log(report_detail) def step5(self): self.report_simulation = self.report_result(self.get_options_dict(), self.latency1 + self.latency2)