KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _has_valid_hash method in avocado

Best Python code snippet using avocado_python

asset.py

Source:asset.py

...264        if time.monotonic() > expire_time:265            return True266        return False267    @classmethod268    def _has_valid_hash(cls, asset_path, asset_hash=None):269        """Checks if a file has a valid hash based on the hash parameter.270        If asset_hash is None then will consider a valid asset.271        """272        if asset_hash is None:273            LOG.debug("No hash provided. Cannot check the asset file"274                      " integrity.")275            return True276        hash_path = cls._get_hash_file(asset_path)277        _, hash_from_file = cls.read_hash_from_file(hash_path)278        if hash_from_file == asset_hash:279            return True280        return False281    def _verify_hash(self, asset_path):282        """283        Verify if the `asset_path` hash matches the hash in the hash file.284        :param asset_path: full path of the asset file.285        :returns: True when self.asset_hash is None or when it has the same286        value as the hash of the asset_file, otherwise return False.287        :rtype: bool288        """289        return self._has_valid_hash(asset_path, self.asset_hash)290    def fetch(self, timeout=None):291        """Try to fetch the current asset.292        First tries to find the asset on the provided cache_dirs list.293        Then tries to download the asset from the locations list294        provided.295        :param timeout: timeout in seconds. Default is296                        :data:`afutils.asset.DOWNLOAD_TIMEOUT`.297        :raise OSError: When it fails to fetch the asset298        :returns: The path for the file on the cache directory.299        :rtype: str300        """301        # First let's search for the file in each one of the cache locations302        asset_file = None303        error = "Can't fetch: 'urls' is not defined."304        timeout = timeout or DOWNLOAD_TIMEOUT305        LOG.info("Fetching asset %s", self.name)306        try:307            return self.find_asset_file(create_metadata=True)308        except OSError:309            LOG.info("Asset not in cache, fetching it.")310        # If we get to this point, we have to download it from a location.311        # A writable cache directory is then needed. The first available312        # writable cache directory will be used.313        cache_dir = self._get_writable_cache_dir()314        # Now we have a writable cache_dir. Let's get the asset.315        for url in self.urls:316            if url is None:317                continue318            urlobj = urlparse(url)319            if urlobj.scheme in ['http', 'https', 'ftp']:320                fetch = self._download321            elif urlobj.scheme == 'file':322                fetch = self._get_local_file323            # We are assuming that everything starting with './' or '/' are a324            # file too.325            elif url.startswith(('/', './')):326                fetch = self._get_local_file327            else:328                raise UnsupportedProtocolError("Unsupported protocol"329                                               ": %s" % urlobj.scheme)330            asset_file = os.path.join(cache_dir,331                                      self.relative_dir)332            dirname = os.path.dirname(asset_file)333            if not os.path.isdir(dirname):334                os.makedirs(dirname, exist_ok=True)335            try:336                if fetch(urlobj, asset_file, timeout):337                    LOG.info("Asset downloaded.")338                    if self.metadata is not None:339                        self._create_metadata_file(asset_file)340                    return asset_file341            except Exception:  # pylint: disable=W0703342                exc_type, exc_value = sys.exc_info()[:2]343                LOG.error('%s: %s', exc_type.__name__, exc_value)344                error = exc_value345        raise OSError("Failed to fetch %s (%s)." % (self.asset_name, error))346    def find_asset_file(self, create_metadata=False):347        """348        Search for the asset file in each one of the cache locations349        :param bool create_metadata: Should this method create the350                                     metadata in case asset file found351                                     and metadata is not found?  Default352                                     is False.353        :return: asset path, if it exists in the cache354        :rtype: str355        :raises: OSError356        """357        for cache_dir in self.cache_dirs:358            cache_dir = os.path.expanduser(cache_dir)359            asset_file = os.path.join(cache_dir, self.relative_dir)360            # Ignore non-files361            if not os.path.isfile(asset_file):362                continue363            # Ignore expired asset files364            if self._is_expired(asset_file, self.expire):365                continue366            # Ignore mismatch hash367            if not self._has_valid_hash(asset_file, self.asset_hash):368                continue369            if create_metadata:370                self._create_metadata_file(asset_file)371            LOG.info("Asset already exists in cache.")372            return asset_file373        raise OSError("File %s not found in the cache." % self.asset_name)374    def get_metadata(self):375        """376        Returns metadata of the asset if it exists or None.377        :return: metadata378        :rtype: dict or None379        """380        try:381            asset_file = self.find_asset_file()382        except OSError:383            raise OSError("Metadata not available.")384        basename = os.path.splitext(asset_file)[0]385        metadata_file = "%s_metadata.json" % basename386        if os.path.isfile(metadata_file):387            with open(metadata_file, "r", encoding='utf-8') as f:388                metadata = json.load(f)389                return metadata390    @property391    def asset_name(self):392        if self.parsed_name.query:393            return self.parsed_name.query394        return os.path.basename(self.parsed_name.path)395    @classmethod396    def get_all_assets(cls, cache_dirs, sort=True):397        """Returns all assets stored in all cache dirs."""398        assets = []399        for cache_dir in cache_dirs:400            expanded = os.path.expanduser(cache_dir)401            for root, _, files in os.walk(expanded):402                for f in files:403                    if not f.endswith('-CHECKSUM') and \404                       not f.endswith('_metadata.json'):405                        assets.append(os.path.join(root, f))406        if sort:407            assets = {a: os.stat(a).st_atime for a in assets}408            return [a[0] for a in sorted(assets.items(),409                                         key=lambda x: x[1],410                                         reverse=True)]411        return assets412    @classmethod413    def get_asset_by_name(cls, name, cache_dirs, expire=None, asset_hash=None):414        """This method will return a cached asset based on name if exists.415        You don't have to instantiate an object of Asset class. Just use this416        method.417        To be improved soon: cache_dirs should be not necessary.418        :param name: the asset filename used during registration.419        :param cache_dirs: list of directories to use during the search.420        :param expire: time in seconds for the asset to expire. Expired assets421                       will not be returned.422        :param asset_hash: asset hash.423        :return: asset path, if it exists in the cache.424        :rtype: str425        :raises: OSError426        """427        for cache_dir in cache_dirs:428            asset_file = os.path.join(os.path.expanduser(cache_dir),429                                      'by_name',430                                      name)431            # Ignore non-files432            if not os.path.isfile(asset_file):433                continue434            # Ignore expired asset files435            if cls._is_expired(asset_file, expire):436                continue437            # Ignore mismatch hash438            if not cls._has_valid_hash(asset_file, asset_hash):439                continue440            return asset_file441        raise OSError("File %s not found in the cache." % name)442    @classmethod443    def get_assets_unused_for_days(cls, days, cache_dirs):444        """Return a list of all assets in cache based on the access time.445        This will check if the file's data wasn't modified N days ago.446        :param days: how many days ago will be the threshold. Ex: "10" will447                     return the assets files that *was not* accessed during448                     the last 10 days.449        :param cache_dirs: list of directories to use during the search.450        """451        result = []452        for file_path in cls.get_all_assets(cache_dirs):...

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.