How to use untar_pkg method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...313 if not preserve_install_dir:314 # Make sure we clean up the install_dir315 self._run_command('rm -rf %s' % install_dir)316 self._run_command('mkdir -p %s' % install_dir)317 self.untar_pkg(fetch_path, install_dir)318 except error.PackageFetchError, why:319 raise error.PackageInstallError(320 'Installation of %s(type:%s) failed : %s'321 % (name, pkg_type, why))322 finally:323 if self.do_locking:324 fcntl.flock(lockfile, fcntl.LOCK_UN)325 lockfile.close()326 def fetch_pkg(self, pkg_name, dest_path, repo_url=None, use_checksum=False):327 '''328 Fetch the package into dest_dir from repo_url. By default repo_url329 is None and the package is looked in all the repositories specified.330 Otherwise it fetches it from the specific repo_url.331 pkg_name : name of the package (ex: test-sleeptest.tar.bz2,332 dep-gcc.tar.bz2, kernel.1-1.rpm)333 repo_url : the URL of the repository where the package is located.334 dest_path : complete path of where the package will be fetched to.335 use_checksum : This is set to False to fetch the packages.checksum file336 so that the checksum comparison is bypassed for the337 checksum file itself. This is used internally by the338 packaging system. It should be ignored by externals339 callers of this method who use it fetch custom packages.340 '''341 try:342 self._run_command("ls %s" % os.path.dirname(dest_path))343 except (error.CmdError, error.AutoservRunError):344 raise error.PackageFetchError("Please provide a valid "345 "destination: %s " % dest_path)346 # See if the package was already fetched earlier, if so347 # the checksums need to be compared and the package is now348 # fetched only if they differ.349 pkg_exists = False350 try:351 self._run_command("ls %s" % dest_path)352 pkg_exists = True353 except (error.CmdError, error.AutoservRunError):354 pass355 # if a repository location is explicitly provided, fetch the package356 # from there and return357 if repo_url:358 repositories = [self.get_fetcher(repo_url)]359 elif self.repositories:360 repositories = self.repositories361 else:362 raise error.PackageFetchError("No repository urls specified")363 # install the package from the package repos, try the repos in364 # reverse order, assuming that the 'newest' repos are most desirable365 for fetcher in reversed(repositories):366 try:367 # Fetch the package if it is not there, the checksum does368 # not match, or checksums are disabled entirely369 need_to_fetch = (370 not use_checksum or not pkg_exists371 or not self.compare_checksum(dest_path))372 if need_to_fetch:373 fetcher.fetch_pkg_file(pkg_name, dest_path)374 # update checksum so we won't refetch next time.375 if use_checksum:376 self.update_checksum(dest_path)377 return378 except (error.PackageFetchError, error.AutoservRunError) as e:379 # The package could not be found in this repo, continue looking380 logging.debug(e)381 repo_url_list = [repo.url for repo in repositories]382 message = ('%s could not be fetched from any of the repos %s' %383 (pkg_name, repo_url_list))384 logging.error(message)385 # if we got here then that means the package is not found386 # in any of the repositories.387 raise error.PackageFetchError(message)388 def upload_pkg(self, pkg_path, upload_path=None, update_checksum=False,389 timeout=300):390 from autotest_lib.server import subcommand391 if upload_path:392 upload_path_list = [upload_path]393 self.upkeep(upload_path_list)394 elif len(self.upload_paths) > 0:395 self.upkeep()396 upload_path_list = self.upload_paths397 else:398 raise error.PackageUploadError("Invalid Upload Path specified")399 if update_checksum:400 # get the packages' checksum file and update it with the current401 # package's checksum402 self.update_checksum(pkg_path)403 commands = []404 for path in upload_path_list:405 commands.append(subcommand.subcommand(self.upload_pkg_parallel,406 (pkg_path, path,407 update_checksum)))408 results = subcommand.parallel(commands, timeout, return_results=True)409 for result in results:410 if result:411 print str(result)412 # TODO(aganti): Fix the bug with the current checksum logic where413 # packages' checksums that are not present consistently in all the414 # repositories are not handled properly. This is a corner case though415 # but the ideal solution is to make the checksum file repository specific416 # and then maintain it.417 def upload_pkg_parallel(self, pkg_path, upload_path, update_checksum=False):418 '''419 Uploads to a specified upload_path or to all the repos.420 Also uploads the checksum file to all the repos.421 pkg_path : The complete path to the package file422 upload_path : the absolute path where the files are copied to.423 if set to 'None' assumes 'all' repos424 update_checksum : If set to False, the checksum file is not425 going to be updated which happens by default.426 This is necessary for custom427 packages (like custom kernels and custom tests)428 that get uploaded which do not need to be part of429 the checksum file and bloat it.430 '''431 self.repo_check(upload_path)432 # upload the package433 if os.path.isdir(pkg_path):434 self.upload_pkg_dir(pkg_path, upload_path)435 else:436 self.upload_pkg_file(pkg_path, upload_path)437 if update_checksum:438 self.upload_pkg_file(self._get_checksum_file_path(),439 upload_path)440 def upload_pkg_file(self, file_path, upload_path):441 '''442 Upload a single file. Depending on the upload path, the appropriate443 method for that protocol is called. Currently this simply copies the444 file to the target directory (but can be extended for other protocols)445 This assumes that the web server is running on the same machine where446 the method is being called from. The upload_path's files are447 basically served by that web server.448 '''449 try:450 if upload_path.startswith('ssh://'):451 # parse ssh://user@host/usr/local/autotest/packages452 hostline, remote_path = parse_ssh_path(upload_path)453 try:454'scp %s %s:%s' % (file_path, hostline,455 remote_path))456 r_path = os.path.join(remote_path,457 os.path.basename(file_path))458"ssh %s 'chmod 644 %s'" % (hostline, r_path))459 except error.CmdError:460 logging.error("Error uploading to repository %s",461 upload_path)462 else:463 shutil.copy(file_path, upload_path)464 os.chmod(os.path.join(upload_path,465 os.path.basename(file_path)), 0644)466 except (IOError, os.error), why:467 logging.error("Upload of %s to %s failed: %s", file_path,468 upload_path, why)469 def upload_pkg_dir(self, dir_path, upload_path):470 '''471 Upload a full directory. Depending on the upload path, the appropriate472 method for that protocol is called. Currently this copies the whole473 tmp package directory to the target directory.474 This assumes that the web server is running on the same machine where475 the method is being called from. The upload_path's files are476 basically served by that web server.477 '''478 local_path = os.path.join(dir_path, "*")479 try:480 if upload_path.startswith('ssh://'):481 hostline, remote_path = parse_ssh_path(upload_path)482 try:483'scp %s %s:%s' % (local_path, hostline,484 remote_path))485 ssh_path = os.path.join(remote_path, "*")486"ssh %s 'chmod 644 %s'" % (hostline, ssh_path))487 except error.CmdError:488 logging.error("Error uploading to repository: %s",489 upload_path)490 else:491"cp %s %s " % (local_path, upload_path))492 up_path = os.path.join(upload_path, "*")493"chmod 644 %s" % up_path)494 except (IOError, os.error), why:495 raise error.PackageUploadError("Upload of %s to %s failed: %s"496 % (dir_path, upload_path, why))497 def remove_pkg(self, pkg_name, remove_path=None, remove_checksum=False):498 '''499 Remove the package from the specified remove_path500 pkg_name : name of the package (ex: test-sleeptest.tar.bz2,501 dep-gcc.tar.bz2)502 remove_path : the location to remove the package from.503 '''504 if remove_path:505 remove_path_list = [remove_path]506 elif len(self.upload_paths) > 0:507 remove_path_list = self.upload_paths508 else:509 raise error.PackageRemoveError(510 "Invalid path to remove the pkg from")511 checksum_path = self._get_checksum_file_path()512 if remove_checksum:513 self.remove_checksum(pkg_name)514 # remove the package and upload the checksum file to the repos515 for path in remove_path_list:516 self.remove_pkg_file(pkg_name, path)517 self.upload_pkg_file(checksum_path, path)518 def remove_pkg_file(self, filename, pkg_dir):519 '''520 Remove the file named filename from pkg_dir521 '''522 try:523 # Remove the file524 if pkg_dir.startswith('ssh://'):525 hostline, remote_path = parse_ssh_path(pkg_dir)526 path = os.path.join(remote_path, filename)527"ssh %s 'rm -rf %s/%s'" % (hostline, remote_path,528 path))529 else:530 os.remove(os.path.join(pkg_dir, filename))531 except (IOError, os.error), why:532 raise error.PackageRemoveError("Could not remove %s from %s: %s "533 % (filename, pkg_dir, why))534 def get_mirror_list(self, repo_urls):535 '''536 Stub function for site specific mirrors.537 Returns:538 Priority ordered list539 '''540 return repo_urls541 def _get_checksum_file_path(self):542 '''543 Return the complete path of the checksum file (assumed to be stored544 in self.pkgmgr_dir545 '''546 return os.path.join(self.pkgmgr_dir, CHECKSUM_FILE)547 def _get_checksum_dict(self):548 '''549 Fetch the checksum file if not already fetched. If the checksum file550 cannot be fetched from the repos then a new file is created with551 the current package's (specified in pkg_path) checksum value in it.552 Populate the local checksum dictionary with the values read from553 the checksum file.554 The checksum file is assumed to be present in self.pkgmgr_dir555 '''556 checksum_path = self._get_checksum_file_path()557 if not self._checksum_dict:558 # Fetch the checksum file559 try:560 try:561 self._run_command("ls %s" % checksum_path)562 except (error.CmdError, error.AutoservRunError):563 # The packages checksum file does not exist locally.564 # See if it is present in the repositories.565 self.fetch_pkg(CHECKSUM_FILE, checksum_path)566 except error.PackageFetchError:567 # This should not happen whilst fetching a package..if a568 # package is present in the repository, the corresponding569 # checksum file should also be automatically present. This570 # case happens only when a package571 # is being uploaded and if it is the first package to be572 # uploaded to the repos (hence no checksum file created yet)573 # Return an empty dictionary in that case574 return {}575 # Read the checksum file into memory576 checksum_file_contents = self._run_command('cat '577 + checksum_path).stdout578 # Return {} if we have an empty checksum file present579 if not checksum_file_contents.strip():580 return {}581 # Parse the checksum file contents into self._checksum_dict582 for line in checksum_file_contents.splitlines():583 checksum, package_name = line.split(None, 1)584 self._checksum_dict[package_name] = checksum585 return self._checksum_dict586 def _save_checksum_dict(self, checksum_dict):587 '''588 Save the checksum dictionary onto the checksum file. Update the589 local _checksum_dict variable with this new set of values.590 checksum_dict : New checksum dictionary591 checksum_dir : The directory in which to store the checksum file to.592 '''593 checksum_path = self._get_checksum_file_path()594 self._checksum_dict = checksum_dict.copy()595 checksum_contents = '\n'.join(checksum + ' ' + pkg_name596 for pkg_name, checksum in597 checksum_dict.iteritems())598 # Write the checksum file back to disk599 self._run_command('echo "%s" > %s' % (checksum_contents,600 checksum_path),601 _run_command_dargs={'verbose': False})602 def compute_checksum(self, pkg_path):603 '''604 Compute the MD5 checksum for the package file and return it.605 pkg_path : The complete path for the package file606 '''607 md5sum_output = self._run_command("md5sum %s " % pkg_path).stdout608 return md5sum_output.split()[0]609 def update_checksum(self, pkg_path):610 '''611 Update the checksum of the package in the packages' checksum612 file. This method is called whenever a package is fetched just613 to be sure that the checksums in the local file are the latest.614 pkg_path : The complete path to the package file.615 '''616 # Compute the new checksum617 new_checksum = self.compute_checksum(pkg_path)618 checksum_dict = self._get_checksum_dict()619 checksum_dict[os.path.basename(pkg_path)] = new_checksum620 self._save_checksum_dict(checksum_dict)621 def remove_checksum(self, pkg_name):622 '''623 Remove the checksum of the package from the packages checksum file.624 This method is called whenever a package is removed from the625 repositories in order clean its corresponding checksum.626 pkg_name : The name of the package to be removed627 '''628 checksum_dict = self._get_checksum_dict()629 if pkg_name in checksum_dict:630 del checksum_dict[pkg_name]631 self._save_checksum_dict(checksum_dict)632 def compare_checksum(self, pkg_path):633 '''634 Calculate the checksum of the file specified in pkg_path and635 compare it with the checksum in the checksum file636 Return True if both match else return False.637 pkg_path : The full path to the package file for which the638 checksum is being compared639 '''640 checksum_dict = self._get_checksum_dict()641 package_name = os.path.basename(pkg_path)642 if not checksum_dict or package_name not in checksum_dict:643 return False644 repository_checksum = checksum_dict[package_name]645 local_checksum = self.compute_checksum(pkg_path)646 return (local_checksum == repository_checksum)647 def tar_package(self, pkg_name, src_dir, dest_dir, exclude_string=None):648 '''649 Create a tar.bz2 file with the name 'pkg_name' say test-blah.tar.bz2.650 Excludes the directories specified in exclude_string while tarring651 the source. Returns the tarball path.652 '''653 tarball_path = os.path.join(dest_dir, pkg_name)654 temp_path = tarball_path + '.tmp'655 cmd_list = ['tar', '-cf', temp_path, '-C', src_dir]656 if _PBZIP2_AVAILABLE:657 cmd_list.append('--use-compress-prog=pbzip2')658 else:659 cmd_list.append('-j')660 if exclude_string is not None:661 cmd_list.append(exclude_string)662 try:663 utils.system(' '.join(cmd_list))664 except:665 os.unlink(temp_path)666 raise667 os.rename(temp_path, tarball_path)668 return tarball_path669 def untar_required(self, tarball_path, dest_dir):670 '''671 Compare the checksum of the tarball_path with the .checksum file672 in the dest_dir and return False if it matches. The untar673 of the package happens only if the checksums do not match.674 '''675 checksum_path = os.path.join(dest_dir, '.checksum')676 try:677 existing_checksum = self._run_command('cat ' + checksum_path).stdout678 except (error.CmdError, error.AutoservRunError):679 # If the .checksum file is not present (generally, this should680 # not be the case) then return True so that the untar happens681 return True682 new_checksum = self.compute_checksum(tarball_path)683 return (new_checksum.strip() != existing_checksum.strip())684 def untar_pkg(self, tarball_path, dest_dir):685 '''686 Untar the package present in the tarball_path and put a687 ".checksum" file in the dest_dir containing the checksum688 of the tarball. This method689 assumes that the package to be untarred is of the form690 <name>.tar.bz2691 '''692 self._run_command('tar --no-same-owner -xjf %s -C %s' %693 (tarball_path, dest_dir))694 # Put the .checksum file in the install_dir to note695 # where the package came from696 pkg_checksum = self.compute_checksum(tarball_path)697 pkg_checksum_path = os.path.join(dest_dir,698 '.checksum')...

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?