Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_ops_item method in localstack

Best Python code snippet using localstack_python

availableEBS.py

Source:availableEBS.py

...185        NoTagVols = list(filter(None, volumesWithoutTags))186        opsNoTagData["Value"] = buildOpsEntries(NoTagVols, rgn, acctID) 187        opsItemData = opsItemTemplate188        opsItemData["/aws/resources"] = opsNoTagData189        # ops_res = opscenter.create_ops_item(Description=description, Title=title, Priority=2, Notifications=snsArn, Source="EBS", OperationalData=opsItemData)190        ## process EBS volumes and create OpsItems191        flaggedVols.sort()192        print(f'Available volumes with the next tags: {os.environ["DELETE_TAGS"]} in region: {rgn} : {flaggedVols}') 193        splitList = splitter(flaggedVols, int(os.environ["BATCH_SIZE"]))   # BATCH_SIZE defined in environment variables194        for i in splitList:195            opsData = {}196            processedVols = list(filter(None, i))197            opsData["Value"] = buildOpsEntries(processedVols, rgn, acctID) 198            opsItemData = opsItemTemplate199            opsItemData["/aws/resources"] = opsData200            description = "Unused EBS volume identifier" 201            title = "EBS Volumes older than " + os.environ["IGNORE_WINDOW"] + " days."202            try:203                response = opscenter.create_ops_item(Description=description, Title=title, Priority=2, Notifications=snsArn, Source="EBS", OperationalData=opsItemData)204                # ops_id = response['OpsItemId'] see how response for more than 1 volume looks and parse to handle update205                print(response, i)206            except ClientError as err:207                if err.response['Error']['Code'] == "OpsItemAlreadyExistsException":208                    print("OpsItem already exists, skipping this batch.")209                else:210                    print(err)211            if(os.environ["DETAILED_NOTIFICATIONS"].upper() == "TRUE"):212                try:213                    print(detailedNotifier(processedVols))214                except ClientError as err:215                    print(err)216                try:217                    snapshotAndDelete(processedVols, rgn)...

risk_tracking.py

Source:risk_tracking.py

...75                            opsitem_description='Medium Risk Issue(MRI) identified'76                            opsitem_title='Medium Risk' + ' - ' + workload_name + ' - ' + risk_title77                            severity='3'78                        # Create an OpsItem for missing Best practice79                        create_opsitem = ssm.create_ops_item(80                                  Description=opsitem_description,81                                  OperationalData={82                                      '/aws/resources': {83                                          'Value': '[{\"arn\":\"' + workload_arn + '\"}]',84                                          'Type': 'SearchableString'85                                      },86                                      'WorkloadName': {87                                          'Value': workload_name,88                                          'Type': 'SearchableString'89                                      },90                                      'Best practices missing': {91                                          'Value': risk_title,92                                          'Type': 'SearchableString'93                                      },...

notifications_test.py

Source:notifications_test.py

...57        return res['SubscriptionArn']58    @staticmethod59    def creat_ops_item(id):60        print('creating ops item with test id "{}"'.format(id))61        res = ssm.create_ops_item(62            Description='Description-{}'.format(id),63            Source='test',64            Title="Title-{}".format(id),65        )66        return res['OpsItemId']67    @staticmethod68    def cleanup(subscription_arn, queue_url):69        print('unsubscribing from temp queue via "{}"'.format(subscription_arn))70        sns.unsubscribe(71            SubscriptionArn=subscription_arn,72        )73        print('deleting temp queue "{}"'.format(queue_url))74        sqs.delete_queue(75            QueueUrl=queue_url,...

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.