KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_policy_state method in localstack

Best Python code snippet using localstack_python

contact_tracing_policy.py

Source:contact_tracing_policy.py

...193            agents: Collection of :class:`~episimmer.agent.Agent` objects194        """195        if CTPolicy.reduce_flag:196            for agent in agents.values():197                agent_ct_state = agent.get_policy_state('Contact_Tracing')198                if agent_ct_state is not None:199                    for policy_index in agent_ct_state.keys():200                        CTPolicy.reduce_agent_schedule_time(201                            agent_ct_state, policy_index)202            CTPolicy.reduce_flag = False203    @staticmethod204    def get_policy_index_list(agent: Agent) -> Union[List[int], None]:205        """206        Gets the list of policy indices corresponding to contact tracing policies207        Args:208            agent: Current agent209        Returns:210            List of policy indices (if they exist)211        """212        agent_ct_state = agent.get_policy_state('Contact_Tracing')213        if agent_ct_state is not None:214            return list(agent_ct_state.keys())215        return None216    @staticmethod217    def get_contact_list(agent: Agent, policy_index: int) -> List[str]:218        """219        Gets the contacts saved in an agent's contact tracing policy state for a particular policy index.220        Args:221            agent: Agent whose contacts are to be returned222            policy_index: Policy index of contact tracing policy223        Returns:224            All the contacts of the agent saved across all contact tracing policies225        """226        agent_ct_state = agent.get_policy_state('Contact_Tracing')227        contact_list = ContactList()228        if agent_ct_state is not None:229            if 'contact_deque' in agent_ct_state[policy_index].keys():230                contacts_deque = agent_ct_state[policy_index]['contact_deque']231                for contact_list_obj in contacts_deque:232                    contact_list = contact_list.union_contacts(233                        contact_list_obj)234        return contact_list.contacts235    @staticmethod236    def set_contacts_schedule_time(agents: Dict[str, Agent],237                                   contacts: List[str], policy_index: int,238                                   time_period: int) -> None:239        """240        Sets the time needed to lockdown for contacts of a positive agent to a fixed period. If the scheduled time of241        the contact is not 0 (contact currently in lockdown), then the time needed to lockdown is not reset.242        Args:243            agents: Collection of :class:`~episimmer.agent.Agent` objects244            contacts: List of contacts of a positively tested agent245            policy_index: Policy index of contact tracing policy246            time_period: Time period of lockdown247        """248        for contact_index in contacts:249            contact_agent = agents[contact_index]250            contact_ct_state = contact_agent.get_policy_state(251                'Contact_Tracing')252            if contact_ct_state[policy_index]['schedule_time'] == 0:253                contact_ct_state[policy_index]['schedule_time'] = time_period254    @staticmethod255    def get_max_schedule_time(agent: Agent) -> int:256        """257        Gets the maximum scheduled time left across contact tracing policies.258        Args:259            agent: Current agent260        Returns:261            Maximum schedule time left for lockdown for an agent262        """263        max_schedule_time = 0264        agent_ct_policy = agent.get_policy_state('Contact_Tracing')265        if agent_ct_policy is not None:266            for policy_index in agent_ct_policy.keys():267                agent_ct_scheduled_time = agent_ct_policy[policy_index][268                    'schedule_time']269                if agent_ct_scheduled_time is not None:270                    max_schedule_time = max(max_schedule_time,271                                            agent_ct_scheduled_time)...

iam.py

Source:iam.py

...15        return "AWS::IAM::ManagedPolicy"16    def get_physical_resource_id(self, attribute=None, **kwargs):17        return aws_stack.policy_arn(self.props["ManagedPolicyName"])18    def fetch_state(self, stack_name, resources):19        return IAMPolicy.get_policy_state(self, stack_name, resources, managed_policy=True)20    @classmethod21    def get_deploy_templates(cls):22        def _create(resource_id, resources, resource_type, func, stack_name, *args, **kwargs):23            iam = aws_stack.connect_to_service("iam")24            resource = resources[resource_id]25            props = resource["Properties"]26            cls.resolve_refs_recursively(stack_name, props, resources)27            policy_doc = json.dumps(props["PolicyDocument"])28            policy = iam.create_policy(29                PolicyName=props["ManagedPolicyName"], PolicyDocument=policy_doc30            )31            policy_arn = policy["Policy"]["Arn"]32            for role in resource.get("Roles", []):33                iam.attach_role_policy(RoleName=role, PolicyArn=policy_arn)34            for user in resource.get("Users", []):35                iam.attach_user_policy(UserName=user, PolicyArn=policy_arn)36            for group in resource.get("Groups", []):37                iam.attach_group_policy(GroupName=group, PolicyArn=policy_arn)38            return {}39        return {"create": {"function": _create}}40class IAMUser(GenericBaseModel):41    @staticmethod42    def cloudformation_type():43        return "AWS::IAM::User"44    def get_physical_resource_id(self, attribute=None, **kwargs):45        return self.props.get("UserName")46    def get_resource_name(self):47        return self.props.get("UserName")48    def fetch_state(self, stack_name, resources):49        user_name = self.resolve_refs_recursively(stack_name, self.props.get("UserName"), resources)50        return aws_stack.connect_to_service("iam").get_user(UserName=user_name)["User"]51    def update_resource(self, new_resource, stack_name, resources):52        props = new_resource["Properties"]53        client = aws_stack.connect_to_service("iam")54        return client.update_user(55            UserName=props.get("UserName"),56            NewPath=props.get("NewPath") or "",57            NewUserName=props.get("NewUserName") or "",58        )59    @staticmethod60    def get_deploy_templates():61        return {62            "create": {63                "function": "create_user",64                "parameters": ["Path", "UserName", "PermissionsBoundary", "Tags"],65            },66            "delete": {67                "function": "delete_user",68                "parameters": ["UserName"],69            },70        }71class IAMRole(GenericBaseModel, MotoRole):72    @staticmethod73    def cloudformation_type():74        return "AWS::IAM::Role"75    def get_resource_name(self):76        return self.props.get("RoleName")77    def fetch_state(self, stack_name, resources):78        role_name = self.resolve_refs_recursively(stack_name, self.props.get("RoleName"), resources)79        return aws_stack.connect_to_service("iam").get_role(RoleName=role_name)["Role"]80    def update_resource(self, new_resource, stack_name, resources):81        props = new_resource["Properties"]82        client = aws_stack.connect_to_service("iam")83        return client.update_role(84            RoleName=props.get("RoleName"), Description=props.get("Description") or ""85        )86    @staticmethod87    def get_deploy_templates():88        return {89            "create": {90                "function": "create_role",91                "parameters": param_defaults(92                    dump_json_params(93                        select_parameters(94                            "Path",95                            "RoleName",96                            "AssumeRolePolicyDocument",97                            "Description",98                            "MaxSessionDuration",99                            "PermissionsBoundary",100                            "Tags",101                        ),102                        "AssumeRolePolicyDocument",103                    ),104                    {"RoleName": PLACEHOLDER_RESOURCE_NAME},105                ),106            },107            "delete": {"function": "delete_role", "parameters": {"RoleName": "RoleName"}},108        }109class IAMPolicy(GenericBaseModel):110    @staticmethod111    def cloudformation_type():112        return "AWS::IAM::Policy"113    def fetch_state(self, stack_name, resources):114        return IAMPolicy.get_policy_state(self, stack_name, resources, managed_policy=False)115    @classmethod116    def get_deploy_templates(cls):117        def _create(resource_id, resources, resource_type, func, stack_name, *args, **kwargs):118            iam = aws_stack.connect_to_service("iam")119            props = resources[resource_id]["Properties"]120            cls.resolve_refs_recursively(stack_name, props, resources)121            policy_doc = json.dumps(remove_none_values(props["PolicyDocument"]))122            policy_name = props["PolicyName"]123            for role in props.get("Roles", []):124                iam.put_role_policy(125                    RoleName=role, PolicyName=policy_name, PolicyDocument=policy_doc126                )127            for user in props.get("Users", []):128                iam.put_user_policy(129                    UserName=user, PolicyName=policy_name, PolicyDocument=policy_doc130                )131            for group in props.get("Groups", []):132                iam.put_group_policy(133                    GroupName=group, PolicyName=policy_name, PolicyDocument=policy_doc134                )135            return {}136        return {"create": {"function": _create}}137    @staticmethod138    def get_policy_state(obj, stack_name, resources, managed_policy=False):139        def _filter(pols):140            return [p for p in pols["AttachedPolicies"] if p["PolicyName"] == policy_name]141        iam = aws_stack.connect_to_service("iam")142        props = obj.props143        policy_name = props.get("PolicyName") or props.get("ManagedPolicyName")144        result = {}145        roles = props.get("Roles", [])146        users = props.get("Users", [])147        groups = props.get("Groups", [])148        if managed_policy:149            result["policy"] = iam.get_policy(PolicyArn=aws_stack.policy_arn(policy_name))150        for role in roles:151            role = obj.resolve_refs_recursively(stack_name, role, resources)152            policies = (...

rollout_sampler_test.py

Source:rollout_sampler_test.py

...4from rollout.simple_parallel_rollout import SimpleParallelPolicyRoller5import parameters.parameter_server as P6def test_rollout_sampler():7    policy, _ = load_model("pvn_full_bidomain")8    policy_state = policy.get_policy_state()9    from visualization import Presenter10    #roller = SimplePolicyRoller(policy_factory)11    roller =  SimpleParallelPolicyRoller("pvn_full_bidomain", num_workers=4)12    rollout_sampler = RolloutSampler(roller)13    # TODO: Load some policy14    print("Sampling once")15    rollouts = rollout_sampler.sample_n_rollouts(12, policy_state)16    print("Sampling twice")17    rollouts += rollout_sampler.sample_n_rollouts(12, policy_state)18    print("Sampling thrice")19    rollouts += rollout_sampler.sample_n_rollouts(12, policy_state)20    for rollout in rollouts:21        print("Visualizing rollout")22        for sample in rollout:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.