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

How to use install_kinesalite method in localstack

Best Python code snippet using localstack_python

install.py

Source:install.py

...29        # install plugins30        # TODO remove31        # cmd = 'echo y | %s/bin/elasticsearch-plugin install x-pack'32        # run(cmd % INSTALL_DIR_ES)33def install_kinesalite():34    target_dir = '%s/kinesalite' % INSTALL_DIR_NPM35    if not os.path.exists(target_dir):36        LOGGER.info('Downloading and installing local Kinesis server. This may take some time.')37        run('cd "%s" && npm install' % ROOT_PATH)38def is_alpine():39    try:40        run('cat /etc/issue | grep Alpine', print_error=False)41        return True42    except Exception as e:43        return False44def install_dynamodb_local():45    if not os.path.exists(INSTALL_DIR_DDB):46        LOGGER.info('Downloading and installing local DynamoDB server. This may take some time.')47        run('mkdir -p %s' % INSTALL_DIR_DDB)48        if not os.path.exists(TMP_ARCHIVE_DDB):49            download(DYNAMODB_JAR_URL, TMP_ARCHIVE_DDB)50        cmd = 'cd %s && cp %s ddb.zip && unzip -q ddb.zip && rm ddb.zip'51        run(cmd % (INSTALL_DIR_DDB, TMP_ARCHIVE_DDB))52        # fix for Alpine, otherwise DynamoDBLocal fails with:53        # DynamoDBLocal_lib/libsqlite4java-linux-amd64.so: __memcpy_chk: symbol not found54        if is_alpine():55            patched_lib = ('https://rawgit.com/bhuisgen/docker-alpine/master/alpine-dynamodb/' +56                'rootfs/usr/local/dynamodb/DynamoDBLocal_lib/libsqlite4java-linux-amd64.so')57            patched_jar = ('https://rawgit.com/bhuisgen/docker-alpine/master/alpine-dynamodb/' +58                'rootfs/usr/local/dynamodb/DynamoDBLocal_lib/sqlite4java.jar')59            run("curl -L -o %s/DynamoDBLocal_lib/libsqlite4java-linux-amd64.so '%s'" % (INSTALL_DIR_DDB, patched_lib))60            run("curl -L -o %s/DynamoDBLocal_lib/sqlite4java.jar '%s'" % (INSTALL_DIR_DDB, patched_jar))61def install_component(name):62    if name == 'kinesis':63        install_kinesalite()64    elif name == 'dynamodb':65        # install_dynalite()66        install_dynamodb_local()67    elif name == 'es':68        install_elasticsearch()69def install_components(names):70    parallelize(install_component, names)71def install_all_components():72    install_components(DEFAULT_SERVICE_PORTS.keys())73if __name__ == '__main__':74    if len(sys.argv) > 1 and sys.argv[1] == 'run':75        print('Initializing installation.')76        logging.basicConfig(level=logging.INFO)77        install_all_components()...

kinesis_starter.py

Source:kinesis_starter.py

...9from localstack.services.install import ROOT_PATH10LOGGER = logging.getLogger(__name__)11def start_kinesis(port=None, asynchronous=False, shard_limit=100, update_listener=None):12    port = port or config.PORT_KINESIS13    install.install_kinesalite()14    backend_port = DEFAULT_PORT_KINESIS_BACKEND15    kinesis_data_dir_param = ''16    if config.DATA_DIR:17        kinesis_data_dir = '%s/kinesis' % config.DATA_DIR18        mkdir(kinesis_data_dir)19        kinesis_data_dir_param = '--path %s' % kinesis_data_dir20    cmd = ('%s/node_modules/kinesalite/cli.js --shardLimit %s --port %s %s' %21        (ROOT_PATH, shard_limit, backend_port, kinesis_data_dir_param))22    print('Starting mock Kinesis (%s port %s)...' % (get_service_protocol(), port))23    start_proxy_for_service('kinesis', port, backend_port, update_listener)24    return do_run(cmd, asynchronous)25def check_kinesis(expect_shutdown=False, print_error=False):26    out = None27    try:...

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.