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

How to use terminate method in Airtest

Best Python code snippet using Airtest

JenkinsJob.py

Source:JenkinsJob.py

1from __future__ import print_function2import json3import re4import sys5import ConfigParser6import boto37import credstash8import requests9# asg lifecycle actions10LAUNCH_STR = 'autoscaling:EC2_INSTANCE_LAUNCHING'11TERMINATE_STR = 'autoscaling:EC2_INSTANCE_TERMINATING'12# aws clients and resources13asg_client = boto3.client('autoscaling')14s3_client = boto3.client('s3')15ec2_resource = boto3.resource('ec2')16# main entrypoint for lambda function17def handler(event, context):18    print("DEBUG: Received Event\n%s" % json.dumps(event))19    # parse the message and metadata out of the event20    message, metadata = parse_event(event)21    print("DEBUG: Message\n%s" % json.dumps(message))22    print("DEBUG: Metadata\n%s" % json.dumps(metadata))23    transition = message['LifecycleTransition']24    # build instance metadata to support parameter interpolation25    instance_metadata = get_instance_metadata(message['EC2InstanceId'])26    print("DEBUG: Instance Information\n%s" % json.dumps(instance_metadata))27    # determine the config file to use from either a local file or one28    # downloaded from an s3 bucket29    config_file = get_config_file(metadata)30    print("Reading settings from %s" % config_file)31    # load the config file32    settings = read_config(config_file, instance_metadata)33    # run on instance launch and when the user has call_create_job set to true34    if transition == LAUNCH_STR and settings['call_create_job']:35        print("Calling create job %s/job/%s" % (settings['url'],36                                                settings['create_job']))37        run_jenkins_job(settings['create_job'],38                        settings['create_job_params'],39                        settings['create_job_token'],40                        settings)41    # run on instance terminate and when the user has call_terminate_job set42    # to true43    elif transition == TERMINATE_STR and settings['call_terminate_job']:44        print("Calling terminate job %s/job/%s" % (settings['url'],45                                                   settings['terminate_job']))46        run_jenkins_job(settings['terminate_job'],47                        settings['terminate_job_params'],48                        settings['terminate_job_token'],49                        settings)50    # finish the asg lifecycle operation by sending a continue result51    print("Job queued for execution, finishing ASG action")52    response = asg_client.complete_lifecycle_action(53        LifecycleHookName=message['LifecycleHookName'],54        AutoScalingGroupName=message['AutoScalingGroupName'],55        LifecycleActionToken=message['LifecycleActionToken'],56        LifecycleActionResult='CONTINUE',57        InstanceId=instance_metadata['id']58    )59    print("ASG action complete:\n %s" % response)60# returns the message and metadata from an event object61def parse_event(event):62    metadata = {}63    message = json.loads(event['Records'][0]['Sns']['Message'])64    if 'NotificationMetadata' in message.keys():65        metadata = json.loads(message['NotificationMetadata'])66    return message, metadata67# builds and returns a metadata object from an ec2 instance id68def get_instance_metadata(instance_id):69    instance = ec2_resource.Instance(instance_id)70    metadata = {71        "id": instance.instance_id,72        "private_hostname": instance.private_dns_name,73        "private_ip": instance.private_ip_address,74        "public_hostname": instance.public_dns_name,75        "public_ip": instance.public_ip_address76    }77    for tag in instance.tags:78        metadata[tag['Key']] = tag['Value']79    return metadata80# when the user has supplied an s3 bucket and config file location in the81# message metadata, download the file to the tmp directory. otherwise a file82# provided with the function, config.ini83def get_config_file(metadata):84    config_file = 'config.ini'85    if 's3_config_file' in metadata.keys():86        config_file = "/tmp/%s" % metadata['s3_config_file']87        s3_client.download_file(metadata['s3_bucket'],88                                metadata['s3_config_file'],89                                config_file)90    return config_file91# get a csrf token and run the configured jenkins job92def run_jenkins_job(job, params, token, settings):93    job_url = "%s/job/%s/buildWithParameters" % (settings['url'], job)94    job_params = "token=%s&%s&cause=Lambda+ASG+Scale" % (token, params)95    auth = (settings['username'], settings['api_key'])96    # for most jenkins setups, we need a CSRF crumb to subsequently call the97    # jenkins api to trigger a job98    print("DEBUG: Getting CSRF crumb")99    response = requests.get("%s/crumbIssuer/api/json" % settings['url'],100                            auth=auth,101                            timeout=5,102                            verify=settings['verify_ssl'])103    # if we dont get a 2xx response, display the code and dump the response104    if re.match('^2[0-9]{2}$', str(response.status_code)) is None:105        print("Reponse from crumb was not 2xx: %s" % response.status_code)106        print(response.text)107        sys.exit(1)108    crumb = json.loads(response.text)109    headers = {crumb['crumbRequestField']: crumb['crumb']}110    # call the jenkins job api with the supplied parameters111    response = requests.post("%s?%s" % (job_url, job_params),112                             data={},113                             auth=auth,114                             timeout=5,115                             headers=headers,116                             verify=settings['verify_ssl'])117    # if we dont get a 2xx response, display the code and dump the response118    if re.match('^2[0-9]{2}$', str(response.status_code)) is None:119        print("Reponse from job was not 2xx: %s" % (job, response.status_code))120        print(response.text)121        sys.exit(1)122# parses a config file and returns a settings object123def read_config(config_file, instance_metadata):124    settings = {}125    # get user config126    config = ConfigParser.ConfigParser()127    config.read(config_file)128    # get global config settings129    use_credstash = config.getboolean('DEFAULT', 'use_credstash')130    settings['call_create_job'] = config.getboolean('DEFAULT',131                                                    'call_create_job')132    settings['call_terminate_job'] = config.getboolean('DEFAULT',133                                                       'call_terminate_job')134    # get jenkins settings135    settings['url'] = config.get('jenkins', 'url')136    settings['verify_ssl'] = config.getboolean('jenkins', 'verify_ssl')137    if settings['call_create_job']:138        settings['create_job'] = config.get('jenkins', 'create_job')139        settings['create_job_params'] = config.get('jenkins',140                                                   'create_job_params',141                                                   0, instance_metadata)142    if settings['call_terminate_job']:143        settings['terminate_job'] = config.get('jenkins', 'terminate_job')144        settings['terminate_job_params'] = config.get('jenkins',145                                                      'terminate_job_params',146                                                      0, instance_metadata)147    # get credstash settings148    if use_credstash:149        credstash_table = config.get('credstash', 'table')150        settings['username'] = credstash.getSecret(151            config.get('credstash', 'jenkins_username_key'),152            table=credstash_table153        )154        settings['api_key'] = credstash.getSecret(155            config.get('credstash', 'jenkins_user_token_key'),156            table=credstash_table157        )158        if settings['call_create_job']:159            settings['create_job_token'] = credstash.getSecret(160                config.get('credstash', 'jenkins_create_job_token_key'),161                table=credstash_table162            )163        if settings['call_terminate_job']:164            settings['terminate_job_token'] = credstash.getSecret(165                config.get('credstash', 'jenkins_terminate_job_token_key'),166                table=credstash_table167            )168    else:169        settings['username'] = config.get('jenkins', 'username')170        settings['api_key'] = config.get('jenkins', 'api_key')171        if settings['call_create_job']:172            settings['create_job_token'] = config.get('jenkins',173                                                      'create_job_token')174        if settings['call_terminate_job']:175            settings['terminate_job_token'] = config.get('jenkins',176                                                         'terminate_job_token')...

test_process_control.py

Source:test_process_control.py

...41        self.returncode = exit_status42        self.completed_event.set()43class ProcessControlTests(unittest.TestCase):44    @classmethod45    def _suppress_soft_terminate(cls, command):46        # Do the right thing for your platform here.47        # Right now only POSIX-y systems are reporting48        # soft terminate support, so this is set up for49        # those.50        helper = process_control.ProcessHelper.process_helper()51        signals = helper.soft_terminate_signals()52        if signals is not None:53            for signum in helper.soft_terminate_signals():54                command.extend(["--ignore-signal", str(signum)])55    @classmethod56    def inferior_command(57            cls,58            ignore_soft_terminate=False,59            options=None):60        # Base command.61        script_name = "{}/inferior.py".format(os.path.dirname(__file__))62        if not os.path.exists(script_name):63            raise Exception(64                "test inferior python script not found: {}".format(script_name))65        command = ([sys.executable, script_name])66        if ignore_soft_terminate:67            cls._suppress_soft_terminate(command)68        # Handle options as string or list.69        if isinstance(options, str):70            command.extend(options.split())71        elif isinstance(options, list):72            command.extend(options)73        # Return full command.74        return command75class ProcessControlNoTimeoutTests(ProcessControlTests):76    """Tests the process_control module."""77    def test_run_completes(self):78        """Test that running completes and gets expected stdout/stderr."""79        driver = TestInferiorDriver()80        driver.run_command(self.inferior_command())81        self.assertTrue(82            driver.completed_event.wait(5), "process failed to complete")83        self.assertEqual(driver.returncode, 0, "return code does not match")84    def test_run_completes_with_code(self):85        """Test that running completes and gets expected stdout/stderr."""86        driver = TestInferiorDriver()87        driver.run_command(self.inferior_command(options="-r10"))88        self.assertTrue(89            driver.completed_event.wait(5), "process failed to complete")90        self.assertEqual(driver.returncode, 10, "return code does not match")91class ProcessControlTimeoutTests(ProcessControlTests):92    def test_run_completes(self):93        """Test that running completes and gets expected return code."""94        driver = TestInferiorDriver()95        timeout_seconds = 596        driver.run_command_with_timeout(97            self.inferior_command(),98            "{}s".format(timeout_seconds),99            False)100        self.assertTrue(101            driver.completed_event.wait(2 * timeout_seconds),102            "process failed to complete")103        self.assertEqual(driver.returncode, 0)104    def _soft_terminate_works(self, with_core):105        # Skip this test if the platform doesn't support soft ti106        helper = process_control.ProcessHelper.process_helper()107        if not helper.supports_soft_terminate():108            self.skipTest("soft terminate not supported by platform")109        driver = TestInferiorDriver()110        timeout_seconds = 5111        driver.run_command_with_timeout(112            # Sleep twice as long as the timeout interval.  This113            # should force a timeout.114            self.inferior_command(115                options="--sleep {}".format(timeout_seconds * 2)),116            "{}s".format(timeout_seconds),117            with_core)118        # We should complete, albeit with a timeout.119        self.assertTrue(120            driver.completed_event.wait(2 * timeout_seconds),121            "process failed to complete")122        # Ensure we received a timeout.123        self.assertTrue(driver.was_timeout, "expected to end with a timeout")124        self.assertTrue(125            helper.was_soft_terminate(driver.returncode, with_core),126            ("timeout didn't return expected returncode "127             "for soft terminate with core: {}").format(driver.returncode))128    def test_soft_terminate_works_core(self):129        """Driver uses soft terminate (with core request) when process times out.130        """131        self._soft_terminate_works(True)132    def test_soft_terminate_works_no_core(self):133        """Driver uses soft terminate (no core request) when process times out.134        """135        self._soft_terminate_works(False)136    def test_hard_terminate_works(self):137        """Driver falls back to hard terminate when soft terminate is ignored.138        """139        driver = TestInferiorDriver(soft_terminate_timeout=2.0)140        timeout_seconds = 1141        driver.run_command_with_timeout(142            # Sleep much longer than the timeout interval,forcing a143            # timeout.  Do whatever is needed to have the inferior144            # ignore soft terminate calls.145            self.inferior_command(146                ignore_soft_terminate=True,147                options="--never-return"),148            "{}s".format(timeout_seconds),149            True)150        # We should complete, albeit with a timeout.151        self.assertTrue(152            driver.completed_event.wait(60),153            "process failed to complete")154        # Ensure we received a timeout.155        self.assertTrue(driver.was_timeout, "expected to end with a timeout")156        helper = process_control.ProcessHelper.process_helper()157        self.assertTrue(158            helper.was_hard_terminate(driver.returncode),159            ("timeout didn't return expected returncode "160             "for hard teriminate: {} ({})").format(161                 driver.returncode,162                 driver.output))163    def test_inferior_exits_with_live_child_shared_handles(self):164        """inferior exit detected when inferior children are live with shared165        stdout/stderr handles.166        """167        # Requires review D13362 or equivalent to be implemented.168        self.skipTest("http://reviews.llvm.org/D13362")169        driver = TestInferiorDriver()170        # Create the inferior (I1), and instruct it to create a child (C1)171        # that shares the stdout/stderr handles with the inferior.172        # C1 will then loop forever....

tests_terminate_codes.py

Source:tests_terminate_codes.py

1from django.conf import settings2from django.test.utils import override_settings3from rest_framework import status4from rest_framework.test import APITestCase5from base.factories import ManagerFactory6from merchant.factories import MerchantFactory7from merchant.models import Merchant8from tasks.models.terminate_code import SUCCESS_CODES_DISABLED_MSG, TerminateCode9TEST_TERMINATE_CODES = {10    'error': {11        'STARTING': 501,12        'OTHER': 505,13        'MAX_COUNT': 5,14        'DEFAULT_CODES': (15            {'code': 501, 'name': 'Test error #1'},16            {'code': 502, 'name': 'Test error #2'},17            {'code': 503, 'name': 'Test error #3'},18            {'code': 505, 'name': 'Other', 'is_comment_necessary': True}19        )20    },21    'success': {22        'STARTING': 201,23        'OTHER': 205,24        'MAX_COUNT': 5,25        'DEFAULT_CODES': (26            {'code': 201, 'name': 'Test success #1'},27            {'code': 202, 'name': 'Test success #2'},28            {'code': 203, 'name': 'Test success #3'},29            {'code': 205, 'name': 'Other', 'is_comment_necessary': True}30        )31    }32}33class TerminateCodesAPITestCase(APITestCase):34    @classmethod35    def setUpTestData(cls):36        super(TerminateCodesAPITestCase, cls).setUpTestData()37        cls.merchant = MerchantFactory(advanced_completion=Merchant.ADVANCED_COMPLETION_DISABLED)38        cls.manager = ManagerFactory(merchant=cls.merchant)39    def setUp(self):40        self.client.force_authenticate(self.manager)41    def test_merchant_have_error_codes(self):42        resp = self.client.get('/api/terminate-codes/', {'type': TerminateCode.TYPE_ERROR})43        self.assertEqual(resp.json().get('count'), len(settings.TERMINATE_CODES['error']['DEFAULT_CODES']))44    def test_merchant_have_success_codes(self):45        resp = self.client.get('/api/terminate-codes/', {'type': TerminateCode.TYPE_SUCCESS})46        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)47        self.assertIn(SUCCESS_CODES_DISABLED_MSG, resp.json().get('errors'))48        merchant_with_success_codes = MerchantFactory(advanced_completion=Merchant.ADVANCED_COMPLETION_OPTIONAL)49        manager = ManagerFactory(merchant=merchant_with_success_codes)50        self.client.force_authenticate(manager)51        resp = self.client.get('/api/terminate-codes/', {'type': TerminateCode.TYPE_SUCCESS})52        self.assertEqual(resp.status_code, status.HTTP_200_OK)53        self.assertEqual(resp.json().get('count'), len(settings.TERMINATE_CODES['success']['DEFAULT_CODES']))54    @override_settings(TERMINATE_CODES=TEST_TERMINATE_CODES)55    def test_create_error_codes(self):56        resp = self.client.post('/api/terminate-codes/', {'name': 'Error code', 'type': TerminateCode.TYPE_ERROR})57        self.assertEqual(resp.status_code, status.HTTP_201_CREATED)58        self.assertTrue(self.merchant.terminate_codes.filter(type=TerminateCode.TYPE_ERROR,59                                                             name='Error code', code=504).exists())60        resp = self.client.post('/api/terminate-codes/', {'name': 'One more error code'})61        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)62        self.assertFalse(self.merchant.terminate_codes.filter(name='One more error code').exists())63    @override_settings(TERMINATE_CODES=TEST_TERMINATE_CODES)64    def test_create_success_codes(self):65        resp = self.client.post('/api/terminate-codes/', {'name': 'Success code', 'type': TerminateCode.TYPE_SUCCESS})66        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)67        self.assertIn(SUCCESS_CODES_DISABLED_MSG, resp.json().get('errors'))68        merchant_with_success_codes = MerchantFactory(advanced_completion=Merchant.ADVANCED_COMPLETION_OPTIONAL)69        manager = ManagerFactory(merchant=merchant_with_success_codes)70        self.client.force_authenticate(manager)71        resp = self.client.post('/api/terminate-codes/', {'name': 'Success code', 'type': TerminateCode.TYPE_SUCCESS})72        self.assertEqual(resp.status_code, status.HTTP_201_CREATED)73        self.assertTrue(merchant_with_success_codes.terminate_codes.filter(type=TerminateCode.TYPE_SUCCESS,74                                                                           name='Success code', code=204).exists())75        resp = self.client.post('/api/terminate-codes/', {'name': 'One more success code'})76        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)77        self.assertFalse(merchant_with_success_codes.terminate_codes.filter(name='One more success code').exists())78    def test_code_deleting(self):79        error_code = self.merchant.terminate_codes.get(code=501)80        resp = self.client.delete('/api/terminate-codes/%s/' % error_code.id)81        self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)82        error_other_code = self.merchant.terminate_codes.get(code=settings.TERMINATE_CODES['error']['OTHER'])83        resp = self.client.delete('/api/terminate-codes/%s/' % error_other_code.id)84        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)85    def test_code_name_change(self):86        code = self.merchant.terminate_codes.get(code=502)87        resp = self.client.patch('/api/terminate-codes/%s/' % code.id, {'name': 'new name'})88        self.assertEqual(resp.status_code, status.HTTP_200_OK)89        self.assertTrue(self.merchant.terminate_codes.filter(name='new name', code=502).exists())90        other_code = self.merchant.terminate_codes.get(code=settings.TERMINATE_CODES['error']['OTHER'])91        resp = self.client.patch('/api/terminate-codes/%s/' % other_code.id, {'name': 'test name'})...

TerminateOperationalBindingResult.ta.ts

Source:TerminateOperationalBindingResult.ta.ts

1/* eslint-disable */2import {3    ASN1Element as _Element,4    ASN1TagClass as _TagClass,5    NULL,6} from "asn1-ts";7import * as $ from "asn1-ts/dist/node/functional";8import {9    OPTIONALLY_PROTECTED_SEQ,10    _get_decoder_for_OPTIONALLY_PROTECTED_SEQ,11    _get_encoder_for_OPTIONALLY_PROTECTED_SEQ,12} from "../EnhancedSecurity/OPTIONALLY-PROTECTED-SEQ.ta";13import {14    TerminateOperationalBindingResultData,15    _decode_TerminateOperationalBindingResultData,16    _encode_TerminateOperationalBindingResultData,17} from "../OperationalBindingManagement/TerminateOperationalBindingResultData.ta";18export {19    OPTIONALLY_PROTECTED_SEQ,20    _get_decoder_for_OPTIONALLY_PROTECTED_SEQ,21    _get_encoder_for_OPTIONALLY_PROTECTED_SEQ,22} from "../EnhancedSecurity/OPTIONALLY-PROTECTED-SEQ.ta";23export {24    TerminateOperationalBindingResultData,25    _decode_TerminateOperationalBindingResultData,26    _encode_TerminateOperationalBindingResultData,27} from "../OperationalBindingManagement/TerminateOperationalBindingResultData.ta";2829/* START_OF_SYMBOL_DEFINITION TerminateOperationalBindingResult */30/**31 * @summary TerminateOperationalBindingResult32 * @description33 *34 * ### ASN.1 Definition:35 *36 * ```asn137 * TerminateOperationalBindingResult  ::=  CHOICE {38 *   null            NULL,39 *   protected  [1]  OPTIONALLY-PROTECTED-SEQ{ TerminateOperationalBindingResultData },40 *   ... }41 * ```42 */43export type TerminateOperationalBindingResult =44    | { null_: NULL } /* CHOICE_ALT_ROOT */45    | {46          protected_: OPTIONALLY_PROTECTED_SEQ<TerminateOperationalBindingResultData>;47      } /* CHOICE_ALT_ROOT */48    | _Element /* CHOICE_ALT_UNRECOGNIZED_EXT */;49/* END_OF_SYMBOL_DEFINITION TerminateOperationalBindingResult */5051/* START_OF_SYMBOL_DEFINITION _cached_decoder_for_TerminateOperationalBindingResult */52let _cached_decoder_for_TerminateOperationalBindingResult: $.ASN1Decoder<TerminateOperationalBindingResult> | null = null;53/* END_OF_SYMBOL_DEFINITION _cached_decoder_for_TerminateOperationalBindingResult */5455/* START_OF_SYMBOL_DEFINITION _decode_TerminateOperationalBindingResult */56/**57 * @summary Decodes an ASN.1 element into a(n) TerminateOperationalBindingResult58 * @function59 * @param {_Element} el The element being decoded.60 * @returns {TerminateOperationalBindingResult} The decoded data structure.61 */62export function _decode_TerminateOperationalBindingResult(el: _Element) {63    if (!_cached_decoder_for_TerminateOperationalBindingResult) {64        _cached_decoder_for_TerminateOperationalBindingResult = $._decode_extensible_choice<TerminateOperationalBindingResult>(65            {66                "UNIVERSAL 5": ["null_", $._decodeNull],67                "CONTEXT 1": [68                    "protected_",69                    $._decode_explicit<70                        OPTIONALLY_PROTECTED_SEQ<TerminateOperationalBindingResultData>71                    >(() =>72                        _get_decoder_for_OPTIONALLY_PROTECTED_SEQ<TerminateOperationalBindingResultData>(73                            _decode_TerminateOperationalBindingResultData74                        )75                    ),76                ],77            }78        );79    }80    return _cached_decoder_for_TerminateOperationalBindingResult(el);81}82/* END_OF_SYMBOL_DEFINITION _decode_TerminateOperationalBindingResult */8384/* START_OF_SYMBOL_DEFINITION _cached_encoder_for_TerminateOperationalBindingResult */85let _cached_encoder_for_TerminateOperationalBindingResult: $.ASN1Encoder<TerminateOperationalBindingResult> | null = null;86/* END_OF_SYMBOL_DEFINITION _cached_encoder_for_TerminateOperationalBindingResult */8788/* START_OF_SYMBOL_DEFINITION _encode_TerminateOperationalBindingResult */89/**90 * @summary Encodes a(n) TerminateOperationalBindingResult into an ASN.1 Element.91 * @function92 * @param {value} el The element being decoded.93 * @param elGetter A function that can be used to get new ASN.1 elements.94 * @returns {_Element} The TerminateOperationalBindingResult, encoded as an ASN.1 Element.95 */96export function _encode_TerminateOperationalBindingResult(97    value: TerminateOperationalBindingResult,98    elGetter: $.ASN1Encoder<TerminateOperationalBindingResult>99) {100    if (!_cached_encoder_for_TerminateOperationalBindingResult) {101        _cached_encoder_for_TerminateOperationalBindingResult = $._encode_choice<TerminateOperationalBindingResult>(102            {103                null_: $._encodeNull,104                protected_: $._encode_explicit(105                    _TagClass.context,106                    1,107                    () =>108                        _get_encoder_for_OPTIONALLY_PROTECTED_SEQ<TerminateOperationalBindingResultData>(109                            _encode_TerminateOperationalBindingResultData110                        ),111                    $.BER112                ),113            },114            $.BER115        );116    }117    return _cached_encoder_for_TerminateOperationalBindingResult(118        value,119        elGetter120    );121}122123/* END_OF_SYMBOL_DEFINITION _encode_TerminateOperationalBindingResult */124
...

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.