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

How to use add_file_to_jar method in localstack

Best Python code snippet using localstack_python

install.py

Source:install.py

...384        SFN_PATCH_FILE_METAINF,385    ]386    for patch_class in classes:387        patch_url = f"{SFN_PATCH_URL_PREFIX}/{patch_class}"388        add_file_to_jar(patch_class, patch_url, target_jar=INSTALL_PATH_STEPFUNCTIONS_JAR)389    # special case for Manifest file - extract first, replace content, then update in JAR file390    manifest_file = os.path.join(INSTALL_DIR_STEPFUNCTIONS, "META-INF", "MANIFEST.MF")391    if not os.path.exists(manifest_file):392        content = run(["unzip", "-p", INSTALL_PATH_STEPFUNCTIONS_JAR, "META-INF/MANIFEST.MF"])393        content = re.sub(394            "Main-Class: .+", "Main-Class: cloud.localstack.StepFunctionsStarter", content395        )396        classpath = " ".join([os.path.basename(jar) for jar in JAR_URLS])397        content = re.sub(r"Class-Path: \. ", f"Class-Path: {classpath} . ", content)398        save_file(manifest_file, content)399        run(400            ["zip", INSTALL_PATH_STEPFUNCTIONS_JAR, "META-INF/MANIFEST.MF"],401            cwd=INSTALL_DIR_STEPFUNCTIONS,402        )403    # download additional jar libs404    for jar_url in JAR_URLS:405        target = os.path.join(INSTALL_DIR_STEPFUNCTIONS, os.path.basename(jar_url))406        if not file_exists_not_empty(target):407            download(jar_url, target)408    # download aws-sdk lambda handler409    target = os.path.join(INSTALL_DIR_STEPFUNCTIONS, "localstack-internal-awssdk", "awssdk.zip")410    if not file_exists_not_empty(target):411        download(SFN_AWS_SDK_LAMBDA_ZIP_FILE, target)412def add_file_to_jar(class_file, class_url, target_jar, base_dir=None):413    base_dir = base_dir or os.path.dirname(target_jar)414    patch_class_file = os.path.join(base_dir, class_file)415    if not os.path.exists(patch_class_file):416        download(class_url, patch_class_file)417        run(["zip", target_jar, class_file], cwd=base_dir)418def install_dynamodb_local():419    if not os.path.exists(INSTALL_PATH_DDB_JAR):420        log_install_msg("DynamoDB")421        # download and extract archive422        tmp_archive = os.path.join(tempfile.gettempdir(), "localstack.ddb.zip")423        download_and_extract_with_retry(DYNAMODB_JAR_URL, tmp_archive, INSTALL_DIR_DDB)424    # fix logging configuration for DynamoDBLocal425    log4j2_config = """<Configuration status="WARN">426      <Appenders>...

package.py

Source:package.py

...16def add_to_jar(jar, name, data):17    path = 'resources/%s' % name18    print('Adding {}'.format(path))19    jar.writestr(path, data)20def add_file_to_jar(jar, directory, script=None, required=True, strip_dir=True):21    if script is not None:22        path = os.path.join(directory, script)23    else:24        path = directory25    26    # Use glob() to allow for wildcards, e.g. in manifest.txt.27    path_list = glob.glob(path)28    if len(path_list) == 0 and required:29        raise ValueError('No files found matching: %s' % path)30    #elif len(path_list) > 1:31    #    raise ValueError("Wildcard '%s' matches multiple files: %s" % (path, ', '.join(path_list)))32    for this_path in path_list:33        with open(this_path, 'rb') as f:34            if strip_dir:35                # Drop the path when adding to the jar.36                name = os.path.basename(this_path)37            else:38                name = os.path.relpath(this_path)39            add_to_jar(jar, name, f.read())40def add_dir_to_jar(jar, directory, required=True):41    dir_path_list = glob.glob(directory)42    if len(dir_path_list) == 0 and required:43        raise ValueError('No directory found matching: %s' % directory)44    for dir_path in dir_path_list:45        for dirpath, dirnames, filenames in os.walk(dir_path):46            for filename in filenames:47                add_file_to_jar(jar, dirpath, filename, strip_dir=False)48def add_item_to_jar(jar, item):49    path_list = glob.glob(item)50    for this_path in path_list:51        if os.path.isdir(this_path):52            add_dir_to_jar(jar, this_path)53        elif os.path.isfile(this_path):54            add_file_to_jar(jar, this_path)55        else:56            raise ValueError("No file or directory found matching: %s" % this_path)57def build_jar(source_jar_path, dest_jar_path, config, venv=None, definition=None, logdir=None):58    """Build a StormTopology .jar which encapsulates the topology defined in59    topology_dir. Optionally override the module and function names. This60    feature supports the definition of multiple topologies in a single61    directory."""62    if definition is None:63        definition = 'create.create'64    # Prepare data we'll use later for configuring parallelism.65    config_yaml = read_yaml(config)66    parallelism = dict((k.split('.')[-1], v) for k, v in six.iteritems(config_yaml)67        if k.startswith('petrel.parallelism'))68    pip_options = config_yaml.get('petrel.pip_options', '')69    module_name, dummy, function_name = definition.rpartition('.')70    71    topology_dir = os.getcwd()72    # Make a copy of the input "jvmpetrel" jar. This jar acts as a generic73    # starting point for all Petrel topologies.74    source_jar_path = os.path.abspath(source_jar_path)75    dest_jar_path = os.path.abspath(dest_jar_path)76    if source_jar_path == dest_jar_path:77        raise ValueError("Error: Destination and source path are the same.")78    shutil.copy(source_jar_path, dest_jar_path)79    jar = zipfile.ZipFile(dest_jar_path, 'a', compression=zipfile.ZIP_DEFLATED)80    81    added_path_entry = False82    try:83        # Add the files listed in manifest.txt to the jar.84        with open(os.path.join(topology_dir, MANIFEST), 'r') as f:85            for fn in f.readlines():86                # Ignore blank and comment lines.87                fn = fn.strip()88                if len(fn) and not fn.startswith('#'):89                    add_item_to_jar(jar, os.path.expandvars(fn.strip()))90        # Add user and machine information to the jar.91        add_to_jar(jar, '__submitter__.yaml', '''92petrel.user: %s93petrel.host: %s94''' % (getpass.getuser(),socket.gethostname()))95        96        # Also add the topology configuration to the jar.97        with open(config, 'r') as f:98            config_text = f.read()99        add_to_jar(jar, '__topology__.yaml', config_text)100    101        # Call module_name/function_name to populate a Thrift topology object.102        builder = TopologyBuilder()103        module_dir = os.path.abspath(topology_dir)104        if module_dir not in sys.path:105            sys.path[:0] = [ module_dir ]106            added_path_entry = True107        module = __import__(module_name)108        getattr(module, function_name)(builder)109        # Add the spout and bolt Python scripts to the jar. Create a110        # setup_<script>.sh for each Python script.111        # Add Python scripts and any other per-script resources.112        for k, v in chain(six.iteritems(builder._spouts), six.iteritems(builder._bolts)):113            add_file_to_jar(jar, topology_dir, v.script)114            # Create a bootstrap script.115            if venv is not None:116                # Allow overriding the execution command from the "petrel"117                # command line. This is handy if the server already has a118                # virtualenv set up with the necessary libraries.119                v.execution_command = os.path.join(venv, 'bin/python')120            # If a parallelism value was specified in the configuration YAML,121            # override any setting provided in the topology definition script.122            if k in parallelism:123                builder._commons[k].parallelism_hint = int(parallelism.pop(k))124            v.execution_command, v.script = \125                intercept(venv, v.execution_command, os.path.splitext(v.script)[0],126                          jar, pip_options, logdir)127        if len(parallelism):...

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.