How to use temp_dir_prefix method in avocado

Best Python code snippet using avocado_python

local_workflow_decorator.py

Source:local_workflow_decorator.py

1#########2# Copyright (c) 2015 GigaSpaces Technologies Ltd. All rights reserved3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#        http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12#  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13#  * See the License for the specific language governing permissions and14#  * limitations under the License.15import sys16import shutil17import tempfile18from os import path, listdir, makedirs19from functools import wraps20from cloudify.workflows import local21PLUGIN_YAML_NAME = 'plugin.yaml'22IGNORED_LOCAL_WORKFLOW_MODULES = (23    'cloudify_agent.operations',24    'cloudify_agent.installer.operations',25    # maintained for backward compatibily with < 3.3 blueprints26    'worker_installer.tasks',27    'plugin_installer.tasks',28    'windows_agent_installer.tasks',29    'windows_plugin_installer.tasks'30)31def _find_plugin_yaml(original_path):32    """33    Tries to find the plugin.yaml file automatically (by traversing up the34    directory tree).35    :param original_path: The path to start the search from36    :return: The absolute path of the plugin.yaml file (if found, o.w. raises37     an Error)38    """39    running_path = original_path40    while PLUGIN_YAML_NAME not in listdir(running_path):41        level_up_path = path.dirname(running_path)42        if level_up_path == running_path:43            msg = 'Traversing up the folder tree from {0}, failed to find {1}.'44            raise IOError(msg.format(original_path, PLUGIN_YAML_NAME))45        else:46            running_path = level_up_path47    return path.abspath(path.join(running_path, PLUGIN_YAML_NAME))48def _assure_path_exists(dest_path):49    """50    Creates a the destination path (if not exists)51    :param dest_path:52    :return:53    """54    if not path.exists(dest_path):55        makedirs(dest_path)56def _expand_dictionary(inputs, func_self, func_args, func_kwargs):57    func_to_call = None58    if callable(inputs):59        func_to_call = inputs60    elif isinstance(inputs, basestring):61        if func_self is None:62            raise ValueError("You cannot supply 'string' "63                             "references to 'self' object in "64                             "contextmanager mode.")65        else:66            func_to_call = getattr(func_self, inputs)67    if func_to_call:68        return func_to_call(*func_args, **func_kwargs)69    return inputs70def _copy_resources(test_source_path, resources, default_dest_path):71    """72    Copies a list of resources to the dest_path73    :param test_source_path: the default destination path74    :param resources: a list of resources to be copied - can contain source75    path only, or a tuple of source and destination path.76    :return: None77    """78    for resource in resources:79        # Setting resource relative destination path and temp source path80        if isinstance(resource, tuple):81            resource_source_path, relative_dest_path = resource82            relative_dest_path = path.join(relative_dest_path,83                                           path.basename(resource_source_path))84        else:85            resource_source_path = resource86            relative_dest_path = path.basename(resource_source_path)87        # Setting resource source path88        if test_source_path:89            if not path.isabs(resource_source_path):90                resource_source_path = path.join(test_source_path,91                                                 resource_source_path)92        # Setting absolute destination path93        resource_dest_path = path.join(default_dest_path, relative_dest_path)94        _assure_path_exists(path.dirname(resource_dest_path))95        shutil.copyfile(resource_source_path, resource_dest_path)96class WorkflowTestDecorator(object):97    def __init__(self,98                 blueprint_path,99                 copy_plugin_yaml=False,100                 resources_to_copy=None,101                 temp_dir_prefix=None,102                 init_args=None,103                 inputs=None,104                 input_func_args=None,105                 input_func_kwargs=None):106        """107        Sets the required parameters for future env init. passes the108        environment to the cfy_local argument.109        :param blueprint_path: The relative path to the blueprint110        :param copy_plugin_yaml: Tries to find and copy plugin.yaml (optional)111        :param resources_to_copy: Paths to resources to copy (optional)112        :param temp_dir_prefix: prefix for the resources (optional)113        :param init_args: arguments to pass to the environment init (optional).114        :param inputs: directs inputs assignments into init_args0 (optional).115        :param input_func_args: if you pass a function name into the inputs,116               you can use this arg to specify the args to the function.117        :param input_func_kwargs: if you pass a function name into the inputs,118               you can use this arg to specify the kwargs to the function.119        """120        # blueprint to run121        self.blueprint_path = blueprint_path122        self.temp_blueprint_path = None123        # Plugin path and name124        self.resources_to_copy = resources_to_copy if resources_to_copy else []125        self.copy_plugin_yaml = copy_plugin_yaml126        if self.copy_plugin_yaml:127            self.plugin_yaml_filename = PLUGIN_YAML_NAME128        # Set prefix for resources129        self.temp_dir_prefix = temp_dir_prefix130        self.temp_dir = None131        # set init args132        if init_args:133            self.init_args = init_args134            if 'ignored_modules' not in init_args:135                self.init_args['ignored_modules'] = \136                    IGNORED_LOCAL_WORKFLOW_MODULES137        else:138            self.init_args = {139                'ignored_modules': IGNORED_LOCAL_WORKFLOW_MODULES140            }141        # set the inputs (if set)142        if inputs and 'inputs' in self.init_args.keys():143            raise ValueError("You've supplied 'inputs' inside init_args and as"144                             " a keyword. You cannot have more than "145                             "1 'inputs' source is needed.")146        else:147            if inputs:148                self.init_args['inputs'] = inputs149        self.input_func_args = input_func_args or []150        self.input_func_kwargs = input_func_kwargs or {}151    def set_up(self, func_self=None):152        """153        Sets up the enviroment variables needed for this test.154        :param local_file_path: the path of the test file.155        :param test_method_name: the name of the test method.156        :return: The test env which is a wrapped Environment.157        """158        if func_self:159            local_file_path = \160                sys.modules[func_self.__class__.__module__].__file__161            test_method_name = func_self._testMethodName162        else:163            local_file_path, test_method_name = None, None164        # Creating a temp dir165        if self.temp_dir_prefix:166            self.temp_dir = tempfile.mkdtemp(prefix=self.temp_dir_prefix)167        elif test_method_name:168            self.temp_dir = tempfile.mkdtemp(prefix=test_method_name)169        else:170            self.temp_dir = tempfile.mkdtemp()171        # Adding blueprint to the resources to copy172        self.resources_to_copy.append(self.blueprint_path)173        # Finding and adding the plugin174        if func_self is not None and self.copy_plugin_yaml:175            self.resources_to_copy.append(176                _find_plugin_yaml(path.dirname(local_file_path)))177        elif self.copy_plugin_yaml:178            raise StandardError("You cannot use copy_plugin_yaml in "179                                "contextmanager mode.")180        # Copying resources181        _copy_resources(path.dirname(local_file_path)182                        if local_file_path else None,183                        self.resources_to_copy, self.temp_dir)184        # Updating the test_method_name (if not manually set)185        if self.init_args and not self.init_args.get('name'):186            self.init_args['name'] = test_method_name187        # Expand inputs dictionary188        if 'inputs' in self.init_args.keys():189            self.input_func_kwargs['decorator_kwargs'] = vars(self)190            self.init_args['inputs'] = \191                _expand_dictionary(self.init_args['inputs'],192                                   func_self,193                                   self.input_func_args,194                                   self.input_func_kwargs)195        # Init env with supplied args196        temp_blueprint_path = path.join(self.temp_dir,197                                        path.basename(self.blueprint_path))198        test_env = local.init_env(temp_blueprint_path, **self.init_args)199        return test_env200    def tear_down(self):201        """202        Deletes the allocated temp dir203        :return: None204        """205        if path.exists(self.temp_dir):206            shutil.rmtree(self.temp_dir)207    def __call__(self, test):208        @wraps(test)209        def wrapped_test(func_self, *args, **kwargs):210            """211            The wrapper function itself.212            :param func: the function of which this test has been called from213            :return:214            """215            test_env = self.set_up(func_self)216            try:217                test(func_self, test_env, *args, **kwargs)218            finally:219                self.tear_down()220        return wrapped_test221    # Support for context manager222    def __enter__(self):223        return self.set_up()224    def __exit__(self, exc_type, exc_val, exc_tb):...

utils.py

Source:utils.py

1# -*- coding: utf-8 -*-2"""3The tests utils module functionality.4"""5import configparser6import os7import shutil8import tempfile9import textwrap10from contextlib import contextmanager11from typing import TYPE_CHECKING12from pyxform import file_utils13from pyxform.builder import create_survey, create_survey_from_path14from tests import example_xls15if TYPE_CHECKING:16    from typing import Tuple17def path_to_text_fixture(filename):18    return os.path.join(example_xls.PATH, filename)19def build_survey(filename):20    path = path_to_text_fixture(filename)21    return create_survey_from_path(path)22def create_survey_from_fixture(fixture_name, filetype="xls", include_directory=False):23    fixture_path = path_to_text_fixture("%s.%s" % (fixture_name, filetype))24    noop, section_dict = file_utils.load_file_to_dict(fixture_path)25    pkg = {"main_section": section_dict}26    if include_directory:27        directory, noop = os.path.split(fixture_path)28        pkg["sections"] = file_utils.collect_compatible_files_in_directory(directory)29    return create_survey(**pkg)30def prep_class_config(cls, test_dir="tests"):31    cls.config = configparser.ConfigParser()32    here = os.path.dirname(__file__)33    root = os.path.dirname(here)34    strings = os.path.join(root, test_dir, "fixtures", "strings.ini")35    cls.config.read(strings)36    cls.cls_name = cls.__name__37def prep_for_xml_contains(text: str) -> "Tuple[str]":38    """Prep string for finding an exact match to formatted XML text."""39    # noinspection PyRedundantParentheses40    return (textwrap.indent(textwrap.dedent(text), "    "),)41@contextmanager42def get_temp_file():43    temp_file = tempfile.NamedTemporaryFile(delete=False)44    temp_file.close()45    try:46        yield temp_file.name47    finally:48        temp_file.close()49        if os.path.exists(temp_file.name):50            os.remove(temp_file.name)51@contextmanager52def get_temp_dir():53    temp_dir_prefix = "pyxform_tmp_"54    if os.name == "nt":55        cleanup_pyxform_temp_files(prefix=temp_dir_prefix)56    temp_dir = tempfile.mkdtemp(prefix=temp_dir_prefix)57    try:58        yield temp_dir59    finally:60        try:61            if os.path.exists(temp_dir):62                shutil.rmtree(temp_dir)63        except PermissionError:64            truncate_temp_files(temp_dir=temp_dir)65def truncate_temp_files(temp_dir):66    """67    Truncate files in a folder, recursing into directories.68    """69    # If we can't delete, at least the files can be truncated,70    # so that they don't take up disk space until next cleanup.71    # Seems to be a Windows-specific error for newly-created files.72    temp_root = tempfile.gettempdir()73    if os.path.exists(temp_dir):74        for f in os.scandir(temp_dir):75            if os.path.isdir(f.path):76                truncate_temp_files(f.path)77            else:78                # Check still in temp directory79                if f.path.startswith(temp_root):80                    with open(f.path, mode="w") as _:81                        pass82def cleanup_pyxform_temp_files(prefix: str):83    """84    Try to clean up temp pyxform files from previous test runs.85    """86    temp_root = tempfile.gettempdir()87    if os.path.exists(temp_root):88        for f in os.scandir(temp_root):89            if os.path.isdir(f.path):90                if f.name.startswith(prefix) and f.path.startswith(temp_root):91                    try:92                        shutil.rmtree(f.path)93                    except PermissionError:...

tar.py

Source:tar.py

1import logging2import tempfile3import tarfile4from urllib.parse import urljoin5from urllib.request import pathname2url6from gi.repository import GLib7from os import makedirs, path, sep as path_separator8from .base import BaseImporter9from .util import find_repo, open_repository10class TarImporter(BaseImporter):11    MIME_TYPE = 'application/x-tar'12    TEMP_DIR_PREFIX = path.abspath(path.join(path_separator,13                                             'var',14                                             'tmp',15                                             'ostree-upload-server',16                                             'tar'))17    def _import_commit(self, commit, src_path_obj, target_repo):18        if not self._source_repo_path:19            raise RuntimeError("Cannot invoke _import_commit without calling "20                               "import_to_repo first")21        if not commit:22            raise RuntimeError("Cannot invoke _import_commit without valid "23                               "commit")24        logging.debug("Importing %s@%s into %s...", self._source_repo_path,25                      commit,26                      target_repo.get_path().get_uri())27        options = GLib.Variant('a{sv}', {28            'refs': GLib.Variant('as', [commit]),29            'inherit-transaction': GLib.Variant('b', True),30        })31        source_repo_uri = urljoin('file:',32                                  pathname2url(self._source_repo_path))33        target_repo.pull_with_options(source_repo_uri,34                                      options,35                                      None, None)36        logging.info("Importing complete.")37    def import_to_repo(self):38        logging.info('Trying to use %s extractor...', self.__class__.__name__)39        if not path.isdir(self.__class__.TEMP_DIR_PREFIX):40            makedirs(self.__class__.TEMP_DIR_PREFIX, 0o0755)41        with tempfile.TemporaryDirectory(42                prefix=self.__class__.__name__,43                dir=self.__class__.TEMP_DIR_PREFIX) as dest_path:44            logging.info('Extracting \'%s\' to a temp dir in %s...',45                         self._src_path, dest_path)46            with tarfile.open(self._src_path) as tar_archive:47                tar_archive.extractall(path=dest_path)48            self._source_repo_path = find_repo(dest_path)49            source_repo = open_repository(self._source_repo_path)50            refs = source_repo.list_refs().out_all_refs51            logging.debug("Refs: %r", refs)52            if not refs:53                logging.error("Could not find any refs in the source repo!")54                raise RuntimeError("Missing refs in Tar ostree repository!")55            # We only process the first ref in the repo provided - all56            # other variations are currently unsupported.57            if len(refs) > 1:58                error_msg = ("Multiple refs ({}) found in Tar archive!"59                             .format(refs))60                logging.error(error_msg)61                raise RuntimeError(error_msg)62            ref, commit = refs.popitem()63            logging.info("Ref: %s", ref)64            logging.info("Commit: %s", commit)65            self._apply_commit_to_repo(commit, ref)66        logging.info('Import complete of \'%s\' into %s!', self._src_path,67                     self._repo_path)68# This works with the same code so we just override the mimetype69class TgzImporter(TarImporter):...

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.