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

How to use install_amazon_kinesis_client_libs method in localstack

Best Python code snippet using localstack_python

install.py

Source:install.py

...67                'rootfs/usr/local/dynamodb/DynamoDBLocal_lib/sqlite4java.jar')68            run("curl -L -o %s/libsqlite4java-linux-amd64.so '%s'" % (ddb_libs_dir, patched_lib))69            run("curl -L -o %s/sqlite4java.jar '%s'" % (ddb_libs_dir, patched_jar))70            save_file(patched_marker, '')71def install_amazon_kinesis_client_libs():72    # install KCL/STS JAR files73    if not os.path.exists(INSTALL_DIR_KCL):74        mkdir(INSTALL_DIR_KCL)75        if not os.path.exists(TMP_ARCHIVE_STS):76            download(URL_STS_JAR, TMP_ARCHIVE_STS)77        shutil.copy(TMP_ARCHIVE_STS, INSTALL_DIR_KCL)78    # Compile Java files79    from localstack.utils.kinesis import kclipy_helper80    classpath = kclipy_helper.get_kcl_classpath()81    java_files = '%s/utils/kinesis/java/com/atlassian/*.java' % ROOT_PATH82    class_files = '%s/utils/kinesis/java/com/atlassian/*.class' % ROOT_PATH83    if not glob.glob(class_files):84        run('javac -cp "%s" %s' % (classpath, java_files))85def install_lambda_java_libs():86    # install LocalStack "fat" JAR file (contains all dependencies)87    if not os.path.exists(INSTALL_PATH_LOCALSTACK_FAT_JAR):88        download(URL_LOCALSTACK_FAT_JAR, INSTALL_PATH_LOCALSTACK_FAT_JAR)89def install_component(name):90    if name == 'kinesis':91        install_kinesalite()92    elif name == 'dynamodb':93        install_dynamodb_local()94    elif name == 'es':95        install_elasticsearch()96    elif name == 'sqs':97        install_elasticmq()98def install_components(names):99    parallelize(install_component, names)100    install_lambda_java_libs()101def install_all_components():102    install_components(DEFAULT_SERVICE_PORTS.keys())103# -----------------104# HELPER FUNCTIONS105# -----------------106def is_alpine():107    try:108        run('cat /etc/issue | grep Alpine', print_error=False)109        return True110    except Exception:111        return False112def download_and_extract_with_retry(archive_url, tmp_archive, target_dir):113    def download_and_extract():114        if not os.path.exists(tmp_archive):115            download(archive_url, tmp_archive)116        unzip(tmp_archive, target_dir)117    try:118        download_and_extract()119    except Exception:120        # try deleting and re-downloading the zip file121        LOGGER.info('Unable to extract file, re-downloading ZIP archive: %s' % tmp_archive)122        rm_rf(tmp_archive)123        download_and_extract()124if __name__ == '__main__':125    if len(sys.argv) > 1:126        if sys.argv[1] == 'libs':127            print('Initializing installation.')128            logging.basicConfig(level=logging.INFO)129            logging.getLogger('requests').setLevel(logging.WARNING)130            install_all_components()131            print('Done.')132        elif sys.argv[1] == 'testlibs':133            # Install additional libraries for testing...

34068_install.py

Source:34068_install.py

...67                'rootfs/usr/local/dynamodb/DynamoDBLocal_lib/sqlite4java.jar')68            run("curl -L -o %s/libsqlite4java-linux-amd64.so '%s'" % (ddb_libs_dir, patched_lib))69            run("curl -L -o %s/sqlite4java.jar '%s'" % (ddb_libs_dir, patched_jar))70            save_file(patched_marker, '')71def install_amazon_kinesis_client_libs():72    # install KCL/STS JAR files73    if not os.path.exists(INSTALL_DIR_KCL):74        mkdir(INSTALL_DIR_KCL)75        if not os.path.exists(TMP_ARCHIVE_STS):76            download(URL_STS_JAR, TMP_ARCHIVE_STS)77        shutil.copy(TMP_ARCHIVE_STS, INSTALL_DIR_KCL)78    # Compile Java files79    from localstack.utils.kinesis import kclipy_helper80    classpath = kclipy_helper.get_kcl_classpath()81    java_files = '%s/utils/kinesis/java/com/atlassian/*.java' % ROOT_PATH82    class_files = '%s/utils/kinesis/java/com/atlassian/*.class' % ROOT_PATH83    if not glob.glob(class_files):84        run('javac -cp "%s" %s' % (classpath, java_files))85def install_lambda_java_libs():86    # install LocalStack "fat" JAR file (contains all dependencies)87    if not os.path.exists(INSTALL_PATH_LOCALSTACK_FAT_JAR):88        download(URL_LOCALSTACK_FAT_JAR, INSTALL_PATH_LOCALSTACK_FAT_JAR)89def install_component(name):90    if name == 'kinesis':91        install_kinesalite()92    elif name == 'dynamodb':93        install_dynamodb_local()94    elif name == 'es':95        install_elasticsearch()96    elif name == 'sqs':97        install_elasticmq()98def install_components(names):99    parallelize(install_component, names)100    install_lambda_java_libs()101def install_all_components():102    install_components(DEFAULT_SERVICE_PORTS.keys())103# -----------------104# HELPER FUNCTIONS105# -----------------106def is_alpine():107    try:108        run('cat /etc/issue | grep Alpine', print_error=False)109        return True110    except Exception:111        return False112def download_and_extract_with_retry(archive_url, tmp_archive, target_dir):113    def download_and_extract():114        if not os.path.exists(tmp_archive):115            download(archive_url, tmp_archive)116        unzip(tmp_archive, target_dir)117    try:118        download_and_extract()119    except Exception:120        # try deleting and re-downloading the zip file121        LOGGER.info('Unable to extract file, re-downloading ZIP archive: %s' % tmp_archive)122        rm_rf(tmp_archive)123        download_and_extract()124if __name__ == '__main__':125    if len(sys.argv) > 1:126        if sys.argv[1] == 'libs':127            print('Initializing installation.')128            logging.basicConfig(level=logging.INFO)129            logging.getLogger('requests').setLevel(logging.WARNING)130            install_all_components()131            print('Done.')132        elif sys.argv[1] == 'testlibs':133            # Install additional libraries for testing...

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.