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

How to use create_task_output_dir method in avocado

Best Python code snippet using avocado_python

manager.py

Source:manager.py

...19    """20    def __init__(self):21        self.task_counter = 022        self.tasks={}23        self.create_task_output_dir()24   25    def clear_contexts(self):26        """Delete all contexts27        """28        cmanager = ContextManager()29        for mid,m in self.iter_modules():30            context_id = self.get_context_id(mid)31            cmanager.remove(context_id)32    def clear(self):33        """Clear the list of modules and all associated tasks34        """35        # clear contexts36        self.clear_contexts()37        # clear tasks38        self.clear_tasks()39        # prune docker images and containers40        ContextManager().docker_prune()41    def clear_tasks(self):42        """Clear list of tasks43        """44        self.tasks = {}45        # remove all task outputs 46        os.system("rm -rf {}".format(os.path.join(config.task_directory,"*")))47    def iter_modules(self, status="installed", debug=False):48        """Module iterator, yields tuples (module_id, module_obj)49        """50        if debug:51            modules = Module.query.where(Module.status == status)52        else:53            modules = Module.query.where(Module.status == status).where(Module.debug_flag==False)54        for module in modules:55            yield module.id, module.info()56        57    def iter_tasks(self):58        """Task iterator, yields tuples (module_id, module_obj)59        """60        tasks = Task.query.all()61        for task in tasks:62            yield task.id, task.info()63    def module_from_row(self, row):64        """Create a ModuleInfo object from a row of the module info table65        """66        module_id = int(row[0])67        return ModuleInfo.from_json(row[1])68    def __contains__(self, module_id):69        """Check if a module is installed, the input can be either the modules id or a ModuleInfo object70        """71        return Module.query.get(module_id) is not None72    def __len__(self):73        """Get number of installed modules74        """75        return Module.query.count()76    def register_module(self, module, verbose=True):77        """Register a module, returns the module id78        """79        if verbose:80            sys.stdout.write("Registrating {}...".format(module))81            sys.stdout.flush()82        # create the context83        image_tag = "orchestra:{}".format(module.metadata["name"])84        context = module.get_context()85        context = ContextManager().build(context, tag = image_tag)86        module.metadata["build_log"] = context["log"]87        if "error" in context:88            # set the metadata89            module.metadata["status"] = "error"90            return module.metadata, None91        image = context["image"]92        return module.metadata, image.id93       94    def __getitem__(self, module_id):95        """Get a ModuleInfo object by id96        """97        return Module.query.get(module_id).info()98    def remove_module(self, module_id):99        """Remove a module100        """101        print(f"Remove moduel {module_id}")102        # delete related contexts first103        context_id = self.get_context_id(module_id)104        ContextManager().remove(context_id)105    def forcefully_remove_directory(self, path):106        """Forcefully remove a directory107        """108        cmd = "rm -rf {}".format(path)109        #cmd = "rm -rf {} &> /dev/null".format(path)110        os.system(cmd)111    def create_task_output_dir(self):112        """Create task output directory if needed113        """114        if not os.path.exists(config.task_directory):115            try:116                original_umask = os.umask(0)117                os.makedirs(config.task_directory, exist_ok=True, mode=0o777)118                os.system("chmod a+s {}".format(config.task_directory))119            except Exception as e:120                os.umask(original_umask)121                raise e122    def get_task_dir(self, task_id):123        """Get path of a task directory124        """125        return os.path.join(config.task_directory, "task_{}".format(task_id))...

process.py

Source:process.py

...18        if runtime_task.spawner_handle is None:19            return False20        return runtime_task.spawner_handle.returncode is None21    async def spawn_task(self, runtime_task):22        self.create_task_output_dir(runtime_task)23        task = runtime_task.task24        runner = task.runnable.runner_command()25        args = runner[1:] + ["task-run"] + task.get_command_args()26        runner = runner[0]27        # When running Avocado Python modules, the interpreter on the new28        # process needs to know where Avocado can be found.29        env = os.environ.copy()30        env["PYTHONPATH"] = ":".join(p for p in sys.path)31        # pylint: disable=E113332        try:33            runtime_task.spawner_handle = await asyncio.create_subprocess_exec(34                runner,35                *args,36                stdout=asyncio.subprocess.DEVNULL,37                stderr=asyncio.subprocess.DEVNULL,38                env=env39            )40        except (FileNotFoundError, PermissionError):41            return False42        asyncio.ensure_future(self._collect_task(runtime_task.spawner_handle))43        return True44    def create_task_output_dir(self, runtime_task):45        output_dir_path = self.task_output_dir(runtime_task)46        os.makedirs(output_dir_path, exist_ok=True)47        runtime_task.task.setup_output_dir(output_dir_path)48    @staticmethod49    async def wait_task(runtime_task):  # pylint: disable=W022150        await runtime_task.spawner_handle.wait()51    @staticmethod52    async def terminate_task(runtime_task):  # pylint: disable=W022153        runtime_task.spawner_handle.terminate()54    @staticmethod55    async def check_task_requirements(runtime_task):56        """Check the runtime task requirements needed to be able to run"""57        # right now, limit the check to the runner availability.58        if runtime_task.task.runnable.runner_command() is None:...

base_setting.py

Source:base_setting.py

...10    @task_configuration: task configuration parameters.11    """12    def __init__(self, task_configuration: dict):13        self.update_by_dict(task_configuration)14        self.create_task_output_dir()15    def update_by_dict(self, config_dict: dict):16        """Update class attributes by dict.17        18        @config_dict: configuration dictionary19        """20        for key, val in config_dict.items():21            setattr(self, key, val)22    def create_task_output_dir(self):23        """Create task output path.24        """25        # create task dir26        self.task_dir = os.path.join(self.exp_dir, self.task_name)27        if not os.path.exists(self.task_dir):28            os.makedirs(self.task_dir, exist_ok=True)29        # create model saving dir30        self.model_dir = os.path.join(self.task_dir, "Model")31        if not os.path.exists(self.model_dir):32            os.makedirs(self.model_dir, exist_ok=True)33            34        # create model result dir35        self.result_dir = os.path.join(self.task_dir, "Result")36        if not os.path.exists(self.result_dir):...

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.