How to use setup_dep method in autotest

Best Python code snippet using autotest_python

task.py

Source:task.py Github

copy

Full Screen

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...

Full Screen

Full Screen

install_deps.py

Source:install_deps.py Github

copy

Full Screen

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:...

Full Screen

Full Screen

main.py

Source:main.py Github

copy

Full Screen

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")...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful