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)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!