How to use install_pkg_post method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...183 :param dest_path: Destination path to download the file to.184 :raise PackageFetchError: if the fetch failed185 """186 raise NotImplementedError()187 def install_pkg_post(self, filename, fetch_dir,188 install_dir, preserve_install_dir=False):189 """190 Fetcher specific post install191 :type filename: string192 :param filename: The filename of the package to install193 :type fetch_dir: string194 :param fetch_dir: The fetched path of the package195 :type install_dir: string196 :param install_dir: The path to install the package to197 :type preserve_install_dir: boolean198 @preserve_install_dir: Preserve the install directory199 """200 # check to see if the install_dir exists and if it does201 # then check to see if the .checksum file is the latest202 install_dir_exists = False203 try:204 self.pkgmgr._run_command("ls %s" % install_dir)205 install_dir_exists = True206 except (error.CmdError, error.AutoservRunError):207 pass208 fetch_path = os.path.join(fetch_dir, re.sub("/", "_", filename))209 if (install_dir_exists and210 not self.pkgmgr.untar_required(fetch_path, install_dir)):211 return212 # untar the package into install_dir and213 # update the checksum in that directory214 if not preserve_install_dir:215 # Make sure we clean up the install_dir216 self.pkgmgr._run_command('rm -rf %s' % install_dir)217 self.pkgmgr._run_command('mkdir -p %s' % install_dir)218 self.pkgmgr.untar_pkg(fetch_path, install_dir)219class HttpFetcher(RepositoryFetcher):220 '''221 Repository Fetcher using HTTP222 '''223 #224 # parameters: url, destination file path225 #226 wget_cmd_pattern = 'wget --connect-timeout=15 -nv %s -O %s'227 def _quick_http_test(self):228 """229 Runs a wget command with a 30s timeout230 This checks that the repository is reachable, and avoids the need to231 wait for a full 10min timeout.232 """233 # just make a temp file to write a test fetch into234 mktemp = 'mktemp -u /tmp/tmp.XXXXXX'235 dest_file_path = self.run_command(mktemp).stdout.strip()236 try:237 # build up a wget command238 http_cmd = self.wget_cmd_pattern % (self.url, dest_file_path)239 try:240 self.run_command(http_cmd, _run_command_dargs={'timeout': 30})241 except Exception, e:242 msg = 'HTTP test failed, unable to contact %s: %s'243 raise error.PackageFetchError(msg % (self.url, e))244 finally:245 self.run_command('rm -rf %s' % dest_file_path)246 def fetch_pkg_file(self, filename, dest_path):247 """248 Fetch a package file from a package repository.249 :type filename: string250 :param filename: The filename of the package file to fetch.251 :type dest_path: string252 :param dest_path: Destination path to download the file to.253 :raise PackageFetchError: if the fetch failed254 """255'Fetching %s from %s to %s', filename, self.url,256 dest_path)257 # do a quick test to verify the repo is reachable258 self._quick_http_test()259 # try to retrieve the package via http260 package_url = os.path.join(self.url, filename)261 try:262 cmd = self.wget_cmd_pattern % (package_url, dest_path)263 result = self.run_command(cmd)264 file_exists = self.run_command(265 'ls %s' % dest_path,266 _run_command_dargs={'ignore_status': True}).exit_status == 0267 if not file_exists:268 logging.error('wget failed: %s', result)269 raise error.CmdError(cmd, result)270 logging.debug('Successfully fetched %s from %s', filename,271 package_url)272 except error.CmdError:273 # remove whatever junk was retrieved when the get failed274 self.run_command('rm -f %s' % dest_path)275 raise error.PackageFetchError('%s not found in %s' % (filename,276 package_url))277class GitFetcher(RepositoryFetcher):278 """279 A git based repository fetcher280 """281 #282 # parameters: url, destination file path, <branch>:<file name>283 #284 git_archive_cmd_pattern = 'git archive --remote=%s -o %s %s'285 def __init__(self, package_manager, repository_url):286 """287 Initializes a new GitFetcher288 :type package_manager: BasePackageManager class289 :param package_manager: and instance of BasePackageManager class290 :type repository_url: string291 :param repository_url: The base URL of the git repository292 """293 super(GitFetcher, self).__init__(package_manager, repository_url)294 self._set_repo_url_branch(repository_url)295 logging.debug('GitFetcher initialized with repo=%s and branch=%s',296 self.url, self.branch)297 def _set_repo_url_branch(self, repository_url):298 '''299 Parse the url, look for a branch and set it accordingly300 :type repository_url: string301 :param repository_url: The base URL of the git repository302 '''303 # do we have branch info in the repoistory_url?304 branch = "master"305 match = repository_url.split(":")306 if len(match) > 2:307 # we have a branch308 branch = match[2]309 repository_url = re.sub(":" + branch, "", repository_url)310 self.branch = branch311 def fetch_pkg_file(self, filename, dest_path):312 """313 Fetch a package file and save it to the given destination path314 git is an SCM, you can download the test directly. No need to fetch315 a bz2'd tarball file. However 'filename' is <type>-<name>.tar.bz2316 break this up and only fetch <name>.317 :type filename: string318 :param filename: The filename of the package file to fetch.319 :type dest_path: string320 :param dest_path: Destination path to download the file to.321 """322'Fetching %s from %s to %s', filename, self.url,323 dest_path)324 name, _ = self.pkgmgr.parse_tarball_name(filename)325 package_path = self.branch + " " + name326 try:327 cmd = self.git_archive_cmd_pattern % (self.url, dest_path, package_path)328 result = self.run_command(cmd)329 file_exists = self.run_command(330 'ls %s' % dest_path,331 _run_command_dargs={'ignore_status': True}).exit_status == 0332 if not file_exists:333 logging.error('git archive failed: %s', result)334 raise error.CmdError(cmd, result)335 logging.debug('Successfully fetched %s from %s', package_path,336 self.url)337 except error.CmdError:338 raise error.PackageFetchError('%s not found in %s' % (name,339 package_path))340 def install_pkg_post(self, filename, fetch_dir, install_dir,341 preserve_install_dir=False):342 os_dep.command("tar")343 filename, _ = self.pkgmgr.parse_tarball_name(filename)344 install_path = re.sub(filename, "", install_dir)345 for suffix in ['', '.tar', '.tar.bz2']:346 pkg_name = "%s%s" % (suffix, re.sub("/", "_", filename))347 fetch_path = os.path.join(fetch_dir, pkg_name)348 if os.path.exists(fetch_path):349 self.pkgmgr._run_command('tar -xf %s -C %s' % (fetch_path,350 install_path))351class LocalFilesystemFetcher(RepositoryFetcher):352 def fetch_pkg_file(self, filename, dest_path):353'Fetching %s from %s to %s', filename, self.url,354 dest_path)355 local_path = os.path.join(self.url, filename)356 try:357 self.run_command('cp %s %s' % (local_path, dest_path))358 logging.debug('Successfully fetched %s from %s', filename,359 local_path)360 except error.CmdError, e:361 raise error.PackageFetchError(362 'Package %s could not be fetched from %s'363 % (filename, self.url), e)364class BasePackageManager(object):365 def __init__(self, pkgmgr_dir, hostname=None, repo_urls=None,366 upload_paths=None, do_locking=True,,367 run_function_args=[], run_function_dargs={}):368 '''369 Initializes a new BasePackageManager instance370 One of most used interfaces on this class is the _run_command(), which371 is controlled by the run_function parameter. It defaults to but a custom method (if provided) should be of the same schema as373 It should return a CmdResult object and throw a CmdError374 exception. The reason for using a separate function to run the commands375 is that the same code can be run to fetch a package on the local376 machine or on a remote machine (in which case ssh_host's run function377 is passed in for run_function).378 :type pkgmgr_dir: string379 :param pkgmgr_dir: A directory that can be used by the package manager380 to dump stuff (like checksum files of the repositories etc)381 :type hostname: string382 :param hostname: hostname from where to fetch a list of package repos383 :type repo_urls: list of strings384 :param repo_urls: The list of the repository urls which is consulted385 whilst fetching the package386 :type upload_paths: list of strings387 :param upload_paths: The list of the upload of repositories to which388 the package is uploaded to389 :type do_locking: boolean390 :param do_locking: Enable locking when the packages are installed.391 :type run_function: function392 :param run_function: function used to execute commands.393 :type run_function_args: tuple394 :param run_function_args: positional (tuple-like) arguments to395 run_function396 :param run_function_dargs: dictionary397 :param run_function_dargs: named (dictionary-like) arguments to398 run_function399 '''400 # In memory dictionary that stores the checksum's of packages401 self._checksum_dict = {}402 self.pkgmgr_dir = pkgmgr_dir403 self.do_locking = do_locking404 self.hostname = hostname405 self.repositories = []406 # Create an internal function that is a simple wrapper of407 # run_function and takes in the args and dargs as arguments408 def _run_command(command, _run_command_args=run_function_args,409 _run_command_dargs={}):410 '''411 Special internal function that takes in a command as412 argument and passes it on to run_function (if specified).413 The _run_command_dargs are merged into run_function_dargs414 with the former having more precedence than the latter.415 '''416 new_dargs = dict(run_function_dargs)417 new_dargs.update(_run_command_dargs)418 # avoid polluting logs with extremely verbose packaging output419 new_dargs.update({'stdout_tee': None})420 return run_function(command, *_run_command_args,421 **new_dargs)422 self._run_command = _run_command423 # Process the repository URLs424 if not repo_urls:425 repo_urls = []426 elif hostname:427 repo_urls = self.get_mirror_list(repo_urls)428 for url in repo_urls:429 self.add_repository(url)430 # Process the upload URLs431 if not upload_paths:432 self.upload_paths = []433 else:434 self.upload_paths = list(upload_paths)435 def add_repository(self, repo):436 if isinstance(repo, basestring):437 self.repositories.append(self.get_fetcher(repo))438 elif isinstance(repo, RepositoryFetcher):439 self.repositories.append(repo)440 else:441 raise TypeError("repo must be RepositoryFetcher or url string")442 def get_fetcher(self, url):443 if url.startswith('http://'):444 return HttpFetcher(self, url)445 elif url.startswith('git://'):446 return GitFetcher(self, url)447 else:448 return LocalFilesystemFetcher(self, url)449 def repo_check(self, repo):450 '''451 Check to make sure the repo is in a sane state:452 ensure we have at least XX amount of free space453 Make sure we can write to the repo454 '''455 if not repo.startswith('/') and not repo.startswith('ssh:'):456 return457 try:458 create_directory(repo)459 check_diskspace(repo)460 check_write(repo)461 except (error.RepoWriteError, error.RepoUnknownError,462 error.RepoDiskFullError), e:463 raise error.RepoError("ERROR: Repo %s: %s" % (repo, e))464 def upkeep(self, custom_repos=None):465 '''466 Clean up custom upload/download areas467 '''468 from autotest.server import subcommand469 if not custom_repos:470 # Not all package types necessarily require or allow custom repos471 try:472 custom_repos = settings.get_value('PACKAGES',473 'custom_upload_location').split(',')474 except SettingsError:475 custom_repos = []476 try:477 custom_download = settings.get_value('PACKAGES',478 'custom_download_location')479 custom_repos += [custom_download]480 except SettingsError:481 pass482 if not custom_repos:483 return484 subcommand.parallel_simple(trim_custom_directories, custom_repos,485 log=False)486 def install_pkg(self, name, pkg_type, fetch_dir, install_dir,487 preserve_install_dir=False, repo_url=None):488 '''489 Remove install_dir if it already exists and then recreate it unless490 preserve_install_dir is specified as True.491 Fetch the package into the pkg_dir. Untar the package into install_dir492 The assumption is that packages are of the form :493 <pkg_type>.<pkg_name>.tar.bz2494 name : name of the package495 type : type of the package496 fetch_dir : The directory into which the package tarball will be497 fetched to.498 install_dir : the directory where the package files will be untarred to499 repo_url : the url of the repository to fetch the package from.500 '''501 # do_locking flag is on by default unless you disable it (typically502 # in the cases where packages are directly installed from the server503 # onto the client in which case fcntl stuff wont work as the code504 # will run on the server in that case..505 if self.do_locking:506 lockfile_name = '.%s-%s-lock' % (re.sub("/", "_", name), pkg_type)507 lockfile = open(os.path.join(self.pkgmgr_dir, lockfile_name), 'w')508 try:509 if self.do_locking:510 fcntl.flock(lockfile, fcntl.LOCK_EX)511 self._run_command('mkdir -p %s' % fetch_dir)512 pkg_name = self.get_tarball_name(name, pkg_type)513 try:514 # Fetch the package into fetch_dir515 fetcher = self.fetch_pkg(pkg_name, fetch_dir, use_checksum=True,516 repo_url=repo_url, install=True)517 fetcher.install_pkg_post(pkg_name, fetch_dir, install_dir, preserve_install_dir)518 except error.PackageFetchError, why:519 raise error.PackageInstallError(520 'Installation of %s(type:%s) failed : %s'521 % (name, pkg_type, why))522 finally:523 if self.do_locking:524 fcntl.flock(lockfile, fcntl.LOCK_UN)525 lockfile.close()526 def fetch_pkg(self, pkg_name, dest_path, repo_url=None, use_checksum=False, install=False):527 '''528 Fetch the package into dest_dir from repo_url. By default repo_url529 is None and the package is looked in all the repositories specified.530 Otherwise it fetches it from the specific repo_url.531 pkg_name : name of the package (ex: test-sleeptest.tar.bz2,...

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:


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?