Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use untar_pkg method in autotest

Best Python code snippet using autotest_python
base_packages.py
Source:base_packages.py
...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                    utils.run('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                    utils.run("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                    utils.run('scp %s %s:%s' % (local_path, hostline,484                                                remote_path))485                    ssh_path = os.path.join(remote_path, "*")486                    utils.run("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                utils.run("cp %s %s " % (local_path, upload_path))492                up_path = os.path.join(upload_path, "*")493                utils.run("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                utils.run("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')...
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.