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

How to use list_partner_event_sources method in localstack

Best Python code snippet using localstack_python

client.pyi

Source:client.pyi

...430        account ID that a particular partner event source name is associated with.431        [Show boto3 documentation](https://boto3.amazonaws.com/v1/documentation/api/1.24.58/reference/services/events.html#EventBridge.Client.list_partner_event_source_accounts)432        [Show boto3-stubs documentation](https://vemel.github.io/boto3_stubs_docs/mypy_boto3_events/client.html#list_partner_event_source_accounts)433        """434    def list_partner_event_sources(435        self, *, NamePrefix: str, NextToken: str = None, Limit: int = None436    ) -> ListPartnerEventSourcesResponseTypeDef:437        """438        An SaaS partner can use this operation to list all the partner event source439        names that they have created.440        [Show boto3 documentation](https://boto3.amazonaws.com/v1/documentation/api/1.24.58/reference/services/events.html#EventBridge.Client.list_partner_event_sources)441        [Show boto3-stubs documentation](https://vemel.github.io/boto3_stubs_docs/mypy_boto3_events/client.html#list_partner_event_sources)442        """443    def list_replays(444        self,445        *,446        NamePrefix: str = None,447        State: ReplayStateType = None,448        EventSourceArn: str = None,...

aws_events_info.py

Source:aws_events_info.py

...175                return paginator.paginate(176                    NamePrefix=module.params['name_prefix'],177                ), True178            else:179                return client.list_partner_event_sources(180                    NamePrefix=module.params['name_prefix'],181                ), False182        elif module.params['list_replays']:183            if client.can_paginate('list_replays'):184                paginator = client.get_paginator('list_replays')185                return paginator.paginate(186                    NamePrefix=module.params['name_prefix'],187                ), True188            else:189                return client.list_replays(190                    NamePrefix=module.params['name_prefix'],191                ), False192        elif module.params['list_rules']:193            if client.can_paginate('list_rules'):...

events_client.py

Source:events_client.py

1"""2AWS client to handle service API requests.3"""4from horey.aws_api.aws_clients.boto3_client import Boto3Client5from horey.aws_api.aws_services_entities.event_bridge_rule import EventBridgeRule6from horey.aws_api.aws_services_entities.event_bridge_target import EventBridgeTarget7from horey.aws_api.base_entities.aws_account import AWSAccount8from horey.h_logger import get_logger9logger = get_logger()10import pdb11class EventsClient(Boto3Client):12    """13    Client to handle specific aws service API calls.14    """15    def __init__(self):16        client_name = "events"17        super().__init__(client_name)18    def get_all_rules(self, region=None, full_information=True):19        """20        Get all rules in all regions.21        :return:22        """23        if region is not None:24            return self.get_region_rules(region, full_information=full_information)25        final_result = list()26        for region in AWSAccount.get_aws_account().regions.values():27            final_result += self.get_region_rules(region, full_information=full_information)28        return final_result29    def get_region_rules(self, region, full_information=True, custom_filter=None):30        final_result = list()31        AWSAccount.set_aws_region(region)32        for dict_src in self.execute(self.client.list_rules, "Rules", filters_req=custom_filter):33            obj = EventBridgeRule(dict_src)34            final_result.append(obj)35            if full_information:36                self.update_rule_targets(obj)37                self.update_rule_tags(obj)38                pass39        return final_result40    def update_rule_targets(self, rule):41        filters_req = {"Rule": rule.name}42        rule.targets = []43        for dict_src in self.execute(self.client.list_targets_by_rule, "Targets", filters_req=filters_req):44            obj = EventBridgeTarget(dict_src)45            rule.targets.append(obj)46    def update_rule_tags(self, rule):47        filters_req = {"ResourceARN": rule.arn}48        rule.tags = []49        for dict_src in self.execute(self.client.list_tags_for_resource, "Tags", filters_req=filters_req):50            rule.targets.append(dict_src)51    def get_samples(self):52        ret = list(self.execute(self.client.list_connections, None, raw_data=True))53        ret = list(self.execute(self.client.list_event_buses, None, raw_data=True))54        ret = list(self.execute(self.client.list_event_sources, None, raw_data=True))55        ret = list(self.execute(self.client.list_partner_event_source_accounts, None, raw_data=True))56        ret = list(self.execute(self.client.list_partner_event_sources, None, raw_data=True))57        ret = list(self.execute(self.client.list_replays, None, raw_data=True))58        ret = list(self.execute(self.client.list_rule_names_by_target, None, raw_data=True))59        ret = list(self.execute(self.client.list_targets_by_rule, None, raw_data=True, filters_req={"Rule":"initiator-production-InitiatorEventsRuleSchedule1-MWYJ6917GUN9"}))60    def update_rule_information(self, rule):61        region_rules = self.get_region_rules(rule.region, custom_filter={"NamePrefix": rule.name})62        if len(region_rules) == 1:63            rule.update_from_raw_response(region_rules[0].dict_src)64            return65        if len(region_rules) > 1:66            raise RuntimeError(region_rules)67    def provision_rule(self, rule: EventBridgeRule):68        self.update_rule_information(rule)69        AWSAccount.set_aws_region(rule.region)70        if rule.arn is None:71            response = self.provision_rule_raw(rule.generate_create_request())72            del response["ResponseMetadata"]73            rule.update_from_raw_response(response)74        put_targets_request = rule.generate_put_targets_request()75        if put_targets_request is not None:76            self.put_targets_raw(put_targets_request)77    def provision_rule_raw(self, request_dict):78        logger.info(f"Creating rule: {request_dict}")79        for response in self.execute(self.client.put_rule, None, raw_data=True,80                                     filters_req=request_dict):81            return response82    def put_targets_raw(self, request_dict):83        logger.info(f"Putting targets: {request_dict}")84        for response in self.execute(self.client.put_targets, None, raw_data=True, filters_req=request_dict):85            if response.get("FailedEntryCount") != 0:86                raise RuntimeError(response)...

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.