How to use _get_temp_dir method in autotest

Best Python code snippet using autotest_python

stats_gen_lib_test.py

Source:stats_gen_lib_test.py

...37        num_rank_histogram_buckets=2,38        num_values_histogram_buckets=2,39        num_histogram_buckets=2,40        num_quantiles_histogram_buckets=2)41  def _get_temp_dir(self):42    return tempfile.mkdtemp()43  def _make_example(self, feature_name_to_type_values_tuple_map):44    """Makes a tensorflow example.45    Args:46      feature_name_to_type_values_tuple_map: A map of feature name to47        [feature_type, feature_value_list] tuples. The feature type is one of48        'bytes'/'float'/'int'.49    Raises:50      ValueError: input feature type is invalid.51    Returns:52      A tf.Example.53    """54    result = tf.train.Example()55    for feature_name in feature_name_to_type_values_tuple_map:56      (feature_type, feature_values) = (57          feature_name_to_type_values_tuple_map[feature_name])58      if feature_type == 'bytes':59        result.features.feature[60            feature_name].bytes_list.value[:] = feature_values61      elif feature_type == 'float':62        result.features.feature[63            feature_name].float_list.value[:] = feature_values64      elif feature_type == 'int':65        result.features.feature[66            feature_name].int64_list.value[:] = feature_values67      else:68        raise ValueError('Invalid feature type: ' + feature_type)69    return result70  def _write_tfexamples_to_tfrecords(self, examples, compression_type):71    filename = 'input_data.tfrecord'72    if compression_type == tf.compat.v1.python_io.TFRecordCompressionType.GZIP:73      filename += '.gz'74    data_location = os.path.join(self._get_temp_dir(), filename)75    with tf.io.TFRecordWriter(76        data_location, options=compression_type) as writer:77      for example in examples:78        writer.write(example.SerializeToString())79    return data_location80  _BEAM_COMPRESSION_TYPES = [81      {82          'testcase_name': 'no_compression',83          'compression_type': CompressionTypes.AUTO84      },85      {86          'testcase_name': 'gzip_compression',87          'compression_type': CompressionTypes.GZIP88      },89  ]90  @parameterized.named_parameters(*_BEAM_COMPRESSION_TYPES)91  def test_stats_gen_with_tfrecords_of_tfexamples(self, compression_type):92    examples = [93        self._make_example({94            'a': ('float', [1.0, 2.0]),95            'b': ('bytes', [b'a', b'b', b'c', b'e'])96        }),97        self._make_example({98            'a': ('float', [3.0, 4.0, float('nan'), 5.0]),99            'b': ('bytes', [b'a', b'c', b'd', b'a'])100        }),101        self._make_example({102            'a': ('float', [1.0]),103            'b': ('bytes', [b'a', b'b', b'c', b'd'])104        })105    ]106    tf_compression_lookup = {107        CompressionTypes.AUTO:108            tf.compat.v1.python_io.TFRecordCompressionType.NONE,109        CompressionTypes.GZIP:110            tf.compat.v1.python_io.TFRecordCompressionType.GZIP111    }112    input_data_path = self._write_tfexamples_to_tfrecords(113        examples, tf_compression_lookup[compression_type])114    expected_result = text_format.Parse(115        """116    datasets {117      num_examples: 3118      features {119        path {120          step: "a"121        }122        type: FLOAT123        num_stats {124          common_stats {125            num_non_missing: 3126            num_missing: 0127            min_num_values: 1128            max_num_values: 4129            avg_num_values: 2.33333333130            tot_num_values: 7131            num_values_histogram {132              buckets {133                low_value: 1.0134                high_value: 2.0135                sample_count: 1.5136              }137              buckets {138                low_value: 2.0139                high_value: 4.0140                sample_count: 1.5141              }142              type: QUANTILES143            }144          }145          mean: 2.66666666146          std_dev: 1.49071198147          num_zeros: 0148          min: 1.0149          max: 5.0150          median: 3.0151          histograms {152            num_nan: 1153            buckets {154              low_value: 1.0155              high_value: 3.0156              sample_count: 3.0157            }158            buckets {159              low_value: 3.0160              high_value: 5.0161              sample_count: 3.0162            }163            type: STANDARD164          }165          histograms {166            num_nan: 1167            buckets {168              low_value: 1.0169              high_value: 3.0170              sample_count: 3.0171            }172            buckets {173              low_value: 3.0174              high_value: 5.0175              sample_count: 3.0176            }177            type: QUANTILES178          }179        }180      }181      features {182        path {183          step: "b"184        }185        type: STRING186        string_stats {187          common_stats {188            num_non_missing: 3189            min_num_values: 4190            max_num_values: 4191            avg_num_values: 4.0192            tot_num_values: 12193            num_values_histogram {194              buckets {195                low_value: 4.0196                high_value: 4.0197                sample_count: 1.5198              }199              buckets {200                low_value: 4.0201                high_value: 4.0202                sample_count: 1.5203              }204              type: QUANTILES205            }206          }207          unique: 5208          top_values {209            value: "a"210            frequency: 4.0211          }212          top_values {213            value: "c"214            frequency: 3.0215          }216          avg_length: 1.0217          rank_histogram {218            buckets {219              low_rank: 0220              high_rank: 0221              label: "a"222              sample_count: 4.0223            }224            buckets {225              low_rank: 1226              high_rank: 1227              label: "c"228              sample_count: 3.0229            }230          }231        }232      }233    }234    """, statistics_pb2.DatasetFeatureStatisticsList())235    result = stats_gen_lib.generate_statistics_from_tfrecord(236        data_location=input_data_path,237        stats_options=self._default_stats_options)238    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(239        self, expected_result)240    compare_fn([result])241  def _write_records_to_csv(self, records, tmp_dir, filename,242                            compression_type=''):243    data_location = os.path.join(tmp_dir, filename)244    if compression_type == 'gzip':245      with gzip.GzipFile(data_location, 'wb') as writer:246        writer.write('\n'.join(records).encode('utf-8'))247    else:248      with open(data_location, 'w') as writer:249        writer.write('\n'.join(records))250    return data_location251  def _get_csv_test(self, delimiter=',', with_header=False):252    fields = [['feature1', 'feature2'], ['1.0', 'aa'], ['2.0', 'bb'],253              ['3.0', 'cc'], ['4.0', 'dd'], ['5.0', 'ee'], ['6.0', 'ff'],254              ['7.0', 'gg'], ['', '']]255    records = []256    for row in fields:257      records.append(delimiter.join(row))258    expected_result = text_format.Parse(259        """260    datasets {261  num_examples: 8262  features {263    path {264      step: "feature1"265    }266    type: FLOAT267    num_stats {268      common_stats {269        num_non_missing: 7270        num_missing: 1271        min_num_values: 1272        max_num_values: 1273        avg_num_values: 1.0274        num_values_histogram {275          buckets {276            low_value: 1.0277            high_value: 1.0278            sample_count: 3.5279          }280          buckets {281            low_value: 1.0282            high_value: 1.0283            sample_count: 3.5284          }285          type: QUANTILES286        }287        tot_num_values: 7288      }289      mean: 4.0290      std_dev: 2.0291      min: 1.0292      max: 7.0293      median: 4.0294      histograms {295        buckets {296          low_value: 1.0297          high_value: 4.0298          sample_count: 3.01299        }300        buckets {301          low_value: 4.0302          high_value: 7.0303          sample_count: 3.99304        }305      }306      histograms {307        buckets {308          low_value: 1.0309          high_value: 4.0310          sample_count: 3.5311        }312        buckets {313          low_value: 4.0314          high_value: 7.0315          sample_count: 3.5316        }317        type: QUANTILES318      }319    }320  }321  features {322    path {323      step: "feature2"324    }325    type: STRING326    string_stats {327      common_stats {328        num_non_missing: 7329        num_missing: 1330        min_num_values: 1331        max_num_values: 1332        avg_num_values: 1.0333        num_values_histogram {334          buckets {335            low_value: 1.0336            high_value: 1.0337            sample_count: 3.5338          }339          buckets {340            low_value: 1.0341            high_value: 1.0342            sample_count: 3.5343          }344          type: QUANTILES345        }346        tot_num_values: 7347      }348      unique: 7349      top_values {350        value: "gg"351        frequency: 1.0352      }353      top_values {354        value: "ff"355        frequency: 1.0356      }357      avg_length: 2.0358      rank_histogram {359        buckets {360          label: "gg"361          sample_count: 1.0362        }363        buckets {364          low_rank: 1365          high_rank: 1366          label: "ff"367          sample_count: 1.0368        }369      }370    }371  }372    }373    """, statistics_pb2.DatasetFeatureStatisticsList())374    if with_header:375      return (records, None, expected_result)376    return (records[1:], records[0].split(delimiter), expected_result)377  @parameterized.named_parameters(*_BEAM_COMPRESSION_TYPES)378  def test_stats_gen_with_csv_no_header_in_file(self, compression_type):379    records, header, expected_result = self._get_csv_test(delimiter=',',380                                                          with_header=False)381    compression_type_lookup = {382        CompressionTypes.AUTO: '',383        CompressionTypes.GZIP: 'gzip'384    }385    input_data_path = self._write_records_to_csv(386        records, self._get_temp_dir(), 'input_data.csv',387        compression_type=compression_type_lookup[compression_type])388    result = stats_gen_lib.generate_statistics_from_csv(389        data_location=input_data_path,390        column_names=header,391        delimiter=',',392        stats_options=self._default_stats_options,393        compression_type=compression_type)394    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(395        self, expected_result)396    compare_fn([result])397  def test_stats_gen_with_csv_header_in_file(self):398    records, header, expected_result = self._get_csv_test(delimiter=',',399                                                          with_header=True)400    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),401                                                 'input_data.csv')402    result = stats_gen_lib.generate_statistics_from_csv(403        data_location=input_data_path,404        column_names=header,405        delimiter=',',406        stats_options=self._default_stats_options)407    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(408        self, expected_result)409    compare_fn([result])410  def test_stats_gen_with_csv_tab_delimiter_no_header_in_file(self):411    records, header, expected_result = self._get_csv_test(delimiter='\t',412                                                          with_header=False)413    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),414                                                 'input_data.tsv')415    result = stats_gen_lib.generate_statistics_from_csv(416        data_location=input_data_path,417        column_names=header,418        delimiter='\t',419        stats_options=self._default_stats_options)420    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(421        self, expected_result)422    compare_fn([result])423  def test_stats_gen_with_csv_header_in_multiple_files(self):424    records, _, expected_result = self._get_csv_test(delimiter=',',425                                                     with_header=True)426    header = records.pop(0)427    # Split the records into two subsets and write to separate files.428    records1 = [header] + records[0:3]429    records2 = [header] + records[3:]430    tmp_dir = self._get_temp_dir()431    self._write_records_to_csv(records1, tmp_dir, 'input_data1.csv')432    self._write_records_to_csv(records2, tmp_dir, 'input_data2.csv')433    input_data_path = os.path.join(tmp_dir, 'input_data*')434    result = stats_gen_lib.generate_statistics_from_csv(435        data_location=input_data_path,436        column_names=None,437        delimiter=',',438        stats_options=self._default_stats_options)439    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(440        self, expected_result)441    compare_fn([result])442  def test_stats_gen_with_csv_with_schema(self):443    records = ['feature1', '1']444    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),445                                                 'input_data.csv')446    schema = text_format.Parse(447        """448        feature { name: "feature1" type: BYTES }449        """, schema_pb2.Schema())450    expected_result = text_format.Parse(451        """452    datasets {453  num_examples: 1454  features {455    path {456      step: "feature1"457    }458    type: STRING459    string_stats {460      common_stats {461        num_non_missing: 1462        min_num_values: 1463        max_num_values: 1464        avg_num_values: 1.0465        num_values_histogram {466          buckets {467            low_value: 1.0468            high_value: 1.0469            sample_count: 0.5470          }471          buckets {472            low_value: 1.0473            high_value: 1.0474            sample_count: 0.5475          }476          type: QUANTILES477        }478        tot_num_values: 1479      }480      unique: 1481      top_values {482        value: "1"483        frequency: 1.0484      }485      avg_length: 1.0486      rank_histogram {487        buckets {488          label: "1"489          sample_count: 1.0490        }491      }492    }493  }494    }495    """, statistics_pb2.DatasetFeatureStatisticsList())496    self._default_stats_options.schema = schema497    self._default_stats_options.infer_type_from_schema = True498    result = stats_gen_lib.generate_statistics_from_csv(499        data_location=input_data_path,500        delimiter=',',501        stats_options=self._default_stats_options)502    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(503        self, expected_result)504    compare_fn([result])505  def test_stats_gen_with_invalid_csv_header_in_multiple_files(self):506    records, _, _ = self._get_csv_test(delimiter=',',507                                       with_header=True)508    header = records.pop(0)509    # Split the records into two subsets and write to separate files.510    records1 = [header] + records[0:3]511    records2 = ['random,header'] + records[3:]512    tmp_dir = self._get_temp_dir()513    self._write_records_to_csv(records1, tmp_dir, 'input_data1.csv')514    self._write_records_to_csv(records2, tmp_dir, 'input_data2.csv')515    input_data_path = os.path.join(tmp_dir, 'input_data*')516    with self.assertRaisesRegexp(517        ValueError, 'Files have different headers.'):518      _ = stats_gen_lib.generate_statistics_from_csv(519          data_location=input_data_path, column_names=None, delimiter=',')520  def test_stats_gen_with_csv_missing_column(self):521    records = [',', ',']522    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),523                                                 'input_data.csv')524    expected_result = text_format.Parse(525        """526        datasets {527          num_examples: 2528          features {529            path {530              step: "feature1"531            }532            type: STRING533            string_stats {534              common_stats {535                num_missing: 2536              }537            }538          }539          features {540            path {541              step: "feature2"542            }543            type: STRING544            string_stats {545              common_stats {546                num_missing: 2547              }548            }549          }550        }551        """, statistics_pb2.DatasetFeatureStatisticsList())552    result = stats_gen_lib.generate_statistics_from_csv(553        data_location=input_data_path,554        column_names=['feature1', 'feature2'],555        delimiter=',',556        stats_options=self._default_stats_options)557    compare_fn = test_util.make_dataset_feature_stats_list_proto_equal_fn(558        self, expected_result)559    compare_fn([result])560  def test_stats_gen_with_header_in_empty_csv_file(self):561    input_data_path = self._write_records_to_csv([], self._get_temp_dir(),562                                                 'input_data.csv')563    with self.assertRaisesRegexp(564        ValueError, 'Found empty file when reading the header.*'):565      _ = stats_gen_lib.generate_statistics_from_csv(566          data_location=input_data_path, column_names=None, delimiter=',')567  def test_stats_gen_with_dataframe(self):568    records, _, expected_result = self._get_csv_test(delimiter=',',569                                                     with_header=True)570    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),571                                                 'input_data.csv')572    dataframe = pd.read_csv(input_data_path)573    result = stats_gen_lib.generate_statistics_from_dataframe(574        dataframe=dataframe,575        stats_options=self._default_stats_options, n_jobs=1)576    self.assertLen(result.datasets, 1)577    test_util.assert_dataset_feature_stats_proto_equal(578        self, result.datasets[0], expected_result.datasets[0])579  def test_stats_gen_with_dataframe_feature_allowlist(self):580    records, _, expected_result = self._get_csv_test(delimiter=',',581                                                     with_header=True)582    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),583                                                 'input_data.csv')584    dataframe = pd.read_csv(input_data_path)585    stats_options_allowlist = self._default_stats_options586    stats_options_allowlist.feature_allowlist = list(dataframe.columns)587    dataframe['to_be_removed_column'] = [588        [1, 2], [], None, [1], None, [3, 4], [], None]589    result = stats_gen_lib.generate_statistics_from_dataframe(590        dataframe=dataframe, stats_options=stats_options_allowlist, n_jobs=1)591    self.assertLen(result.datasets, 1)592    test_util.assert_dataset_feature_stats_proto_equal(593        self, result.datasets[0], expected_result.datasets[0])594  def test_stats_gen_with_dataframe_invalid_njobs_zero(self):595    records, _, _ = self._get_csv_test(delimiter=',', with_header=True)596    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),597                                                 'input_data.csv')598    dataframe = pd.read_csv(input_data_path)599    with self.assertRaisesRegexp(600        ValueError, 'Invalid n_jobs parameter.*'):601      _ = stats_gen_lib.generate_statistics_from_dataframe(602          dataframe=dataframe,603          stats_options=self._default_stats_options, n_jobs=0)604  def test_stats_gen_with_dataframe_invalid_njobs_negative(self):605    records, _, _ = self._get_csv_test(delimiter=',', with_header=True)606    input_data_path = self._write_records_to_csv(records, self._get_temp_dir(),607                                                 'input_data.csv')608    dataframe = pd.read_csv(input_data_path)609    with self.assertRaisesRegexp(610        ValueError, 'Invalid n_jobs parameter.*'):611      _ = stats_gen_lib.generate_statistics_from_dataframe(612          dataframe=dataframe,613          stats_options=self._default_stats_options, n_jobs=-2)614  def test_get_csv_header(self):615    temp_directory = self._get_temp_dir()616    delimiter = ','617    records = ['feature1,feature2', '1.0,aa']618    expected_header = ['feature1', 'feature2']619    self._write_records_to_csv(records, temp_directory, 'input_data_1.csv')620    self._write_records_to_csv(records, temp_directory, 'input_data_2.csv')621    data_location = os.path.join(temp_directory, 'input_data_*.csv')622    header = stats_gen_lib.get_csv_header(data_location, delimiter)623    self.assertEqual(header, expected_header)624  def test_get_csv_header_no_file(self):625    data_location = os.path.join(self._get_temp_dir(), 'fileA.csv')626    delimiter = ','627    with self.assertRaisesRegexp(ValueError, 'No file found.*'):628      _ = stats_gen_lib.get_csv_header(data_location, delimiter)629  def test_get_csv_header_empty_file(self):630    empty_file = os.path.join(self._get_temp_dir(), 'empty.csv')631    open(empty_file, 'w+').close()632    delimiter = ','633    with self.assertRaisesRegexp(ValueError, 'Found empty file.*'):634      _ = stats_gen_lib.get_csv_header(empty_file, delimiter)635  def test_get_csv_header_different_headers(self):636    temp_directory = self._get_temp_dir()637    delimiter = ','638    records_1 = ['feature1,feature2', '1.0,aa']639    records_2 = ['feature1,feature2_different', '2.0,bb']640    self._write_records_to_csv(records_1, temp_directory, 'input_data_1.csv')641    self._write_records_to_csv(records_2, temp_directory, 'input_data_2.csv')642    data_location = os.path.join(temp_directory, 'input_data_*.csv')643    with self.assertRaisesRegexp(ValueError, 'Files have different headers.'):644      _ = stats_gen_lib.get_csv_header(data_location, delimiter)645  def test_get_csv_header_gzip(self):646    temp_directory = self._get_temp_dir()647    delimiter = ','648    records = ['feature1,feature2', '1.0,aa']649    expected_header = ['feature1', 'feature2']650    self._write_records_to_csv(651        records, temp_directory, 'input_data_1.csv.gz', compression_type='gzip')652    self._write_records_to_csv(653        records, temp_directory, 'input_data_2.csv.gz', compression_type='gzip')654    data_location = os.path.join(temp_directory, 'input_data_*.csv.gz')655    header = stats_gen_lib.get_csv_header(data_location, delimiter)656    self.assertEqual(header, expected_header)657  def test_get_csv_header_new_line(self):658    temp_directory = self._get_temp_dir()659    delimiter = ','660    records = ['"\n","feature2"', '1.0,aa']661    expected_header = ['\n', 'feature2']662    self._write_records_to_csv(663        records, temp_directory, 'input_data_1.csv.gz', compression_type='gzip')664    self._write_records_to_csv(665        records, temp_directory, 'input_data_2.csv.gz', compression_type='gzip')666    data_location = os.path.join(temp_directory, 'input_data_*.csv.gz')667    header = stats_gen_lib.get_csv_header(data_location, delimiter)668    self.assertEqual(header, expected_header)669if __name__ == '__main__':...

encrypted_file_cache.py

Source:encrypted_file_cache.py

...18        self._auto_delete = auto_delete19    def __del__(self):20        try:21            if self._auto_delete:22                PathUtilsClass.DeleteDirectory(self._get_temp_dir())23        except AttributeError as e:24            pass25    def _setup(self, partial_url):26        try:27            self._id, self._name = partial_url.split('/')28            self._partial_url = partial_url29            self._server_sha1 = None30        except ValueError:31            raise EncryptedFileCacheError("Invalid format partial_url. Needs to be '<guid>/filename.'")32    @staticmethod33    def _write_file(path, content, mode='wb'):34        try:35            with open(path, mode) as f:36                f.write(content)37                logging.debug('Wrote file: {}'.format(path))38                return path39        except (OSError, IOError) as e:40            raise EncryptedFileCacheError('Unable to write to file: {}.'.format(e))41    @staticmethod42    def _read_file(path, mode='rb'):43        try:44            with open(path, mode) as f:45                return f.read()46        except (OSError, IOError) as e:47            raise EncryptedFileCacheError('Unable to read file: {}'.format(e))48    @staticmethod49    def _is_acceptable_code(code, acceptable_codes=None):50        if code == requests.codes.ok:51            return True52        try:53            if code in acceptable_codes:54                return True55        except:56            return False57    def _requests_get(self, url, acceptable_status_codes=None, allow_redirects=False, no_headers=False):58        logging.debug('Downloading file:{}'.format(url))59        headers = {'Authorization': self._config.HttpHeaders['Authorization']}60        if no_headers:61            headers = {}62        response = requests.get(url, headers=headers, verify=False, allow_redirects=allow_redirects)63        if self._is_acceptable_code(response.status_code, acceptable_status_codes):64            return response65        else:66            raise ConnectionError('Unable to download file from server ({}).'.format(response.status_code))67    def _download(self):68        url = self._config.ServerUrl + '/downloads/files/' + self._partial_url69        response = self._requests_get(url, acceptable_status_codes=[200, 302])70        if response.status_code == 302:71            redirect_url = response.headers.get('location')72            response = self._requests_get(redirect_url, allow_redirects=True, no_headers=True)73        return response.content74    def _download_write_encrypt(self):75        if not os.path.isfile(self._get_encrypted_path()) or self._get_local_sha1() != self._get_server_sha1():76            logging.debug('Local and server sha1 are different.')77            content = self._download()78            self._write_file(self._get_encrypted_path(), self._aes.encrypt(content))79            self._write_file(self._get_sha1_path(), self._get_server_sha1(), 'w')80        else:81            logging.debug('Local and server sha1 are the same.')82    def _join_and_create_dir(self, path, paths):83        try:84            new_path = os.path.join(path, paths)85            return PathUtilsClass.MakeDirectory(new_path)86        except OSError as e:87            raise EncryptedFileCacheError('Unable to access/create directory: {}'.format(e))88    def _get_file_cache_dir(self):89        return self._join_and_create_dir(PathUtilsClass.GetFileCacheDirectory(), self._id)90    def _get_sha1_path(self):91        return os.path.join(self._get_file_cache_dir(), SHA1_FILE_NAME)92    def _get_encrypted_path(self):93        return os.path.join(self._get_file_cache_dir(), self._name)94    def _get_local_sha1(self):95        if os.path.isfile(self._get_sha1_path()):96            return  self._read_file(self._get_sha1_path(), 'r')97        else:98            logging.debug('No local sha1 file present.')99            return None100    def _get_temp_dir(self):101        return self._join_and_create_dir(PathUtilsClass.GetFilesDirectory(), self._id)102    def _get_server_sha1(self):103        if not self._server_sha1:104            url = self._config.ServerUrl + '/v1/files/' + self._id105            response = self._requests_get(url)106            data = json.loads(response.content)107            self._server_sha1 = data['sha1']108            logging.debug('Server sha1: {}'.format(self._server_sha1))109        return self._server_sha1110    def _decrypt_file(self):111        encrypted_content = self._read_file(self._get_encrypted_path())112        return self._aes.decrypt(encrypted_content)113    def _write_decrypted_file(self, content):114        decrypted_path = os.path.join(self._get_temp_dir(), self._name)115        return self._write_file(decrypted_path, content)116    def _unzip_file(self, path, ignore_unzip_errors):117        try:118            zip_ref = zipfile.ZipFile(path, 'r')119            zip_ref.extractall(self._get_temp_dir())120            zip_ref.close()121            logging.debug('Unzipped {} to {}'.format(path, self._get_temp_dir()))122        except Exception as e:123            if not ignore_unzip_errors:124                raise EncryptedFileCacheError('Unable to unzip file: {}'.format(e))125            else:126                logging.info('Unable to unzip file: {}'.format(path))127    def _try_download(self):128        try:129            self._download_write_encrypt()130        except ConnectionError as e:131            logging.debug('Got connection error connecting to server.')132            if os.path.isfile(self._get_encrypted_path()):133                logging.debug('Using cached version of file.')134            else:135                raise EncryptedFileCacheError('Unable to download file, and no existing cache: {}'.format(e))136    def get(self, partial_url, unzip=True, ignore_unzip_errors=False):137        self._setup(partial_url)138        self._try_download()139        decrypted_content = self._decrypt_file()140        decrypted_path = self._write_decrypted_file(decrypted_content)141        if unzip:142            self._unzip_file(decrypted_path, ignore_unzip_errors)...

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.