How to use _validate_path method in tempest

Best Python code snippet using tempest_python

s3.py

Source:s3.py

...68            key.delete()69    def delete(self):70        self.bucket.delete()71        self.bucket = None72    def _validate_path(self, path):73        if path is None:74            path = ''75        elif isinstance(path, list):76            path = '/'.join(path)77        path = path.replace('\\', '/')78        path = path.strip('/')79        return path80    def exists(self, path):81        path = self._validate_path(path)82        objs = list(self.bucket.objects.filter(Prefix=path))83        if len(objs) > 0 and objs[0].key == path:84            return True85        else:86            return False87    def has_folder(self, path):88        path = self._validate_path(path)89        if path == '':90            raise ValueError('folder needed')91        path += '/'92        objs = list(self.bucket.objects.filter(Prefix=path))93        return len(objs) > 094    def iterate(self, path=None):95        path = self._validate_path(path)96        if path != '':97            path += '/'98        s3_paginator = self.client.get_paginator('list_objects_v2')99        for page in s3_paginator.paginate(Bucket=self.name, Prefix=path):100            for content in page.get('Contents', ()):101                yield content['Key']102    def listdir(self, path=None):103        path = self._validate_path(path)104        if path != '':105            path += '/'106        files = list(self.iterate(path))107        files = [f[len(path):] for f in files]108        folders = list(set([f.split('/')[0] for f in files if '/' in f]))109        files = [f for f in files if '/' not in f]110        return folders, files111    def save(self, obj, path):112        path = self._validate_path(path)113        if isinstance(obj, dict) or isinstance(obj, list):114            self.save_json(obj, path)115        elif isinstance(obj, io.StringIO) or isinstance(obj, io.BytesIO):116            self.bucket.put_object(Key=path, Body=obj.getvalue())117        else:118            self.bucket.put_object(Key=path, Body=obj)119    def save_json(self, obj, path):120        def default(obj):121            if isinstance(obj, (datetime.date, datetime.datetime)):122                return obj.isoformat()123        obj = json.dumps(obj, default=default, indent=4)124        self.save(obj, path)125    def save_pickle(self, obj, path, zip=False):126        buffer = io.BytesIO()127        if not zip:128            pickle.dump(obj, buffer, protocol=pickle.HIGHEST_PROTOCOL)129        else:130            with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zip:131                zip.writestr('object.pkl', pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL))132        self.save(buffer, path)133    def upload(self, filename, path):134        path = self._validate_path(path)135        self.bucket.upload_file(filename, path)136    def load(self, path):137        path = self._validate_path(path)138        if not self.exists(path):139            return FileNotFoundError('file {} does not exists in the bucket'.format(path))140        obj = self.bucket.Object(path)141        return io.BytesIO(obj.get()['Body'].read())142    def load_json(self, path):143        with self.load(path) as f:144            return json.load(f)145    def load_pickle(self, path, zip=False):146        buffer = self.load(path)147        if not zip:148            return pickle.loads(buffer.getvalue())149        else:150            with zipfile.ZipFile(buffer, 'r') as zip:151                return pickle.loads(zip.read('object.pkl'))152    def walk(self, path):153        raise NotImplementedError()154    def remove(self, path):155        path = self._validate_path(path)156        if self.exists(path):157            self.bucket.delete_objects(Delete={'Objects': [{'Key': path}]})158    def save_multiparts(self, buffer, path):159        buffer.seek(0)160        path = self._validate_path(path)161        config = TransferConfig(162            multipart_threshold=100 * MB,163            multipart_chunksize=100 * MB,164            max_concurrency=10,165            use_threads=True166        )167        self.client.upload_fileobj(168            Fileobj=buffer,169            Bucket=self.name,170            Key=path,171            Config=config172            #Callback=ProgressPercentage(path, size=buffer.__sizeof__())173        )174    def save_pickle_multiparts(self, obj, path):...

flags.py

Source:flags.py

...98  def _check_path_exists(flag_path, flag_name):99    if not file_exists_fn(flag_path):100      raise ValueError('Flag `%s` at %s does not exist.' %101                       (flag_name, flag_path))102  def _validate_path(flag_path, flag_name):103    if not flag_path:104      raise ValueError('Flag `%s` must be provided in mode %s.' %105                       (flag_name, flags_obj.mode))106    _check_path_exists(flag_path, flag_name)107  if 'train' in flags_obj.mode:108    _validate_path(flags_obj.train_input_path, 'train_input_path')109    _validate_path(flags_obj.input_meta_data_path, 'input_meta_data_path')110    if flags_obj.gin_file:111      for gin_file in flags_obj.gin_file:112        _check_path_exists(gin_file, 'gin_file')113    if flags_obj.config_file:114      for config_file in flags_obj.config_file:115        _check_path_exists(config_file, 'config_file')116  if 'eval' in flags_obj.mode:117    _validate_path(flags_obj.validation_input_path, 'validation_input_path')118  if flags_obj.mode == 'predict':119    # model_dir is only needed strictly in 'predict' mode.120    _validate_path(flags_obj.model_dir, 'model_dir')121  if 'predict' in flags_obj.mode:122    _validate_path(flags_obj.test_input_path, 'test_input_path')123  if not flags_obj.config_file and flags_obj.mode != 'predict':124    if flags_obj.hub_module_url:125      if flags_obj.init_checkpoint or flags_obj.model_config_file:126        raise ValueError(127            'When `hub_module_url` is specified, `init_checkpoint` and '128            '`model_config_file` should be empty.')129      logging.info('Using the pretrained tf.hub from %s',130                   flags_obj.hub_module_url)131    else:132      if not (flags_obj.init_checkpoint and flags_obj.model_config_file):133        raise ValueError('Both `init_checkpoint` and `model_config_file` '134                         'should be specified if `config_file` is not '135                         'specified.')136      _validate_path(flags_obj.model_config_file, 'model_config_file')137      logging.info(138          'Using the pretrained checkpoint from %s and model_config_file from '...

path_validator.py

Source:path_validator.py

...14    Validates existence of the resolved path of a `PathDdv`.15    """16    def __init__(self, path_sdv: PathSdv):17        self._path_sdv = path_sdv18    def _validate_path(self, path: DescribedPath) -> Optional[TextRenderer]:19        """20        :return: Error message iff validation was applicable and validation failed.21        """22        raise NotImplementedError('abstract method')23    def validate_pre_sds_if_applicable(self, environment: PathResolvingEnvironmentPreSds) -> Optional[TextRenderer]:24        path_ddv = self._path_sdv.resolve(environment.symbols)25        if path_ddv.exists_pre_sds():26            return self._validate_path(path_ddv.value_pre_sds__d(environment.hds))27        return None28    def validate_post_sds_if_applicable(self, environment: PathResolvingEnvironmentPostSds) -> Optional[TextRenderer]:29        described_path_value = self._path_sdv.resolve(environment.symbols)30        if not described_path_value.exists_pre_sds():31            return self._validate_path(described_path_value.value_post_sds__d(environment.sds))32        return None33class PathDdvValidatorBase(DdvValidator, ABC):34    """35    Validates existence of the resolved path of a `PathDdv`.36    """37    def __init__(self, path_ddv: PathDdv):38        self._path_ddv = path_ddv39    @abstractmethod40    def _validate_path(self, path: DescribedPath) -> Optional[TextRenderer]:41        """42        :return: Error message iff validation was applicable and validation failed.43        """44        raise NotImplementedError('abstract method')45    def validate_pre_sds_if_applicable(self, hds: HomeDs) -> Optional[TextRenderer]:46        p = self._path_ddv47        if p.exists_pre_sds():48            return self._validate_path(p.value_pre_sds__d(hds))49        return None50    def validate_post_sds_if_applicable(self, tcds: TestCaseDs) -> Optional[TextRenderer]:51        p = self._path_ddv52        if not p.exists_pre_sds():53            return self._validate_path(p.value_post_sds__d(tcds.sds))...

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.