Best Python code snippet using pytest-bdd_python
test_csv_parser.py
Source:test_csv_parser.py
1#2# Copyright (c) 2021 Airbyte, Inc., all rights reserved.3#4import json5import os6from pathlib import Path7from typing import Any, List, Mapping8import pytest9from source_s3.source_files_abstract.formats.csv_parser import CsvParser10from .abstract_test_parser import AbstractTestParser11SAMPLE_DIRECTORY = Path(__file__).resolve().parent.joinpath("sample_files/")12test_files = [13 {14 # basic 'normal' test15 "test_alias": "basic_test",16 "AbstractFileParser": CsvParser(17 format={"filetype": "csv"},18 master_schema={19 "id": "integer",20 "name": "string",21 "valid": "boolean",22 "code": "integer",23 "degrees": "number",24 "birthday": "string",25 "last_seen": "string",26 },27 ),28 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_1.csv"),29 "num_records": 8,30 "inferred_schema": {31 "id": "integer",32 "name": "string",33 "valid": "boolean",34 "code": "integer",35 "degrees": "number",36 "birthday": "string",37 "last_seen": "string",38 },39 "line_checks": {},40 "fails": [],41 },42 {43 # tests custom CSV parameters (odd delimiter, quote_char, escape_char & newlines in values in the file)44 "test_alias": "custom csv parameters",45 "AbstractFileParser": CsvParser(46 format={"filetype": "csv", "delimiter": "^", "quote_char": "|", "escape_char": "!", "newlines_in_values": True},47 master_schema={48 "id": "integer",49 "name": "string",50 "valid": "boolean",51 "code": "integer",52 "degrees": "number",53 "birthday": "string",54 "last_seen": "string",55 },56 ),57 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_2_params.csv"),58 "num_records": 8,59 "inferred_schema": {60 "id": "integer",61 "name": "string",62 "valid": "boolean",63 "code": "integer",64 "degrees": "number",65 "birthday": "string",66 "last_seen": "string",67 },68 "line_checks": {},69 "fails": [],70 },71 {72 # tests encoding: Big573 "test_alias": "encoding: Big5",74 "AbstractFileParser": CsvParser(75 format={"filetype": "csv", "encoding": "big5"}, master_schema={"id": "integer", "name": "string", "valid": "boolean"}76 ),77 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_3_enc_Big5.csv"),78 "num_records": 8,79 "inferred_schema": {"id": "integer", "name": "string", "valid": "boolean"},80 "line_checks": {81 3: {82 "id": 3,83 "name": "è®å½¢éåï¼å½è£çæ©å¨äºº",84 "valid": False,85 }86 },87 "fails": [],88 },89 {90 # tests encoding: Arabic (Windows 1256)91 "test_alias": "encoding: Arabic (Windows 1256)",92 "AbstractFileParser": CsvParser(93 format={"filetype": "csv", "encoding": "windows-1256"},94 master_schema={"id": "integer", "notes": "string", "valid": "boolean"},95 ),96 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_4_enc_Arabic.csv"),97 "num_records": 2,98 "inferred_schema": {"id": "integer", "notes": "string", "valid": "boolean"},99 "line_checks": {100 1: {101 "id": 1,102 "notes": "اÙباÙت اÙجÙÙ Ù٠اÙØ£ÙضÙ",103 "valid": False,104 }105 },106 "fails": [],107 },108 {109 # tests compression: gzip110 "test_alias": "compression: gzip",111 "AbstractFileParser": CsvParser(112 format={"filetype": "csv"},113 master_schema={114 "id": "integer",115 "name": "string",116 "valid": "boolean",117 "code": "integer",118 "degrees": "number",119 "birthday": "string",120 "last_seen": "string",121 },122 ),123 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_5.csv.gz"),124 "num_records": 8,125 "inferred_schema": {126 "id": "integer",127 "name": "string",128 "valid": "boolean",129 "code": "integer",130 "degrees": "number",131 "birthday": "string",132 "last_seen": "string",133 },134 "line_checks": {135 7: {136 "id": 7,137 "name": "xZhh1Kyl",138 "valid": False,139 "code": 10,140 "degrees": -9.2,141 "birthday": "2021-07-14",142 "last_seen": "2021-07-14 15:30:09.225145",143 }144 },145 "fails": [],146 },147 {148 # tests compression: bz2149 "test_alias": "compression: bz2",150 "AbstractFileParser": CsvParser(151 format={"filetype": "csv"},152 master_schema={153 "id": "integer",154 "name": "string",155 "valid": "boolean",156 "code": "integer",157 "degrees": "number",158 "birthday": "string",159 "last_seen": "string",160 },161 ),162 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_7_bz2.csv.bz2"),163 "num_records": 8,164 "inferred_schema": {165 "id": "integer",166 "name": "string",167 "valid": "boolean",168 "code": "integer",169 "degrees": "number",170 "birthday": "string",171 "last_seen": "string",172 },173 "line_checks": {174 7: {175 "id": 7,176 "name": "xZhh1Kyl",177 "valid": False,178 "code": 10,179 "degrees": -9.2,180 "birthday": "2021-07-14",181 "last_seen": "2021-07-14 15:30:09.225145",182 }183 },184 "fails": [],185 },186 {187 # tests extra columns in master schema188 "test_alias": "extra columns in master schema",189 "AbstractFileParser": CsvParser(190 format={"filetype": "csv"},191 master_schema={192 "EXTRA_COLUMN_1": "boolean",193 "EXTRA_COLUMN_2": "number",194 "id": "integer",195 "name": "string",196 "valid": "boolean",197 "code": "integer",198 "degrees": "number",199 "birthday": "string",200 "last_seen": "string",201 },202 ),203 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_1.csv"),204 "num_records": 8,205 "inferred_schema": {206 "id": "integer",207 "name": "string",208 "valid": "boolean",209 "code": "integer",210 "degrees": "number",211 "birthday": "string",212 "last_seen": "string",213 },214 "line_checks": {},215 "fails": [],216 },217 {218 # tests missing columns in master schema219 # TODO: maybe this should fail read_records, but it does pick up all the columns from file despite missing from master schema220 "test_alias": "missing columns in master schema",221 "AbstractFileParser": CsvParser(format={"filetype": "csv"}, master_schema={"id": "integer", "name": "string"}),222 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_1.csv"),223 "num_records": 8,224 "inferred_schema": {225 "id": "integer",226 "name": "string",227 "valid": "boolean",228 "code": "integer",229 "degrees": "number",230 "birthday": "string",231 "last_seen": "string",232 },233 "line_checks": {},234 "fails": [],235 },236 {237 # tests empty file, SHOULD FAIL INFER & STREAM RECORDS238 "test_alias": "empty csv file",239 "AbstractFileParser": CsvParser(format={"filetype": "csv"}, master_schema={}),240 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_6_empty.csv"),241 "num_records": 0,242 "inferred_schema": {},243 "line_checks": {},244 "fails": ["test_get_inferred_schema", "test_stream_records"],245 },246 {247 # no header test248 "test_alias": "no header csv file",249 "AbstractFileParser": CsvParser(250 format={251 "filetype": "csv",252 "advanced_options": json.dumps({"column_names": ["id", "name", "valid", "code", "degrees", "birthday", "last_seen"]}),253 },254 master_schema={},255 ),256 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/test_file_8_no_header.csv"),257 "num_records": 8,258 "inferred_schema": {259 "id": "integer",260 "name": "string",261 "valid": "boolean",262 "code": "integer",263 "degrees": "number",264 "birthday": "string",265 "last_seen": "string",266 },267 "line_checks": {},268 "fails": [],269 },270 {271 # tests if infer_datatype parameter set to false disables data type inference on schema272 "test_alias": "infer_datatype set to false without custom schema",273 "AbstractFileParser": CsvParser(format={"filetype": "csv", "infer_datatypes": False, "newlines_in_values": True}),274 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/infer_schema_test.csv"),275 "num_records": 18,276 "inferred_schema": {277 "pk": "string",278 "full_name": "string",279 "street_address": "string",280 "customer_code": "string",281 "email": "string",282 "dob": "string",283 },284 "line_checks": {},285 "fails": [],286 },287 {288 # tests if infer_datatype parameter set to false disables data type inference on schema289 "test_alias": "infer_datatype set to false with custom delimiter and quote",290 "AbstractFileParser": CsvParser(291 format={"filetype": "csv", "infer_datatypes": False, "quote_char": "|", "delimiter": ";", "newlines_in_values": True}292 ),293 "filepath": os.path.join(SAMPLE_DIRECTORY, "csv/infer_schema_test_quote_delim.csv"),294 "num_records": 18,295 "inferred_schema": {296 "pk": "string",297 "full_name": "string",298 "street_address": "string",299 "customer_code": "string",300 "email": "string",301 "dob": "string",302 },303 "line_checks": {},304 "fails": [],305 },306]307@pytest.mark.parametrize("test_file", argvalues=test_files, ids=[file["test_alias"] for file in test_files])308class TestCsvParser(AbstractTestParser):309 @property310 def test_files(self) -> List[Mapping[str, Any]]:...
test_keystores.py
Source:test_keystores.py
1import uuid2from click.testing import CliRunner3from apigee.keystores.commands import create, delete, delete_alias, get_alias, get_all_certs, list, get, list_aliases, create_alias, get_cert4test_keystore = f"test-keystore-{uuid.uuid4()}"5test_alias = {6 "alias": f"test-alias-{uuid.uuid4()}",7 "keySize": 2048, 8 "sigAlg": "SHA256withRSA", 9 "certValidityInDays": 2, 10 "subject": "{\"commonName\":\"test\"}", 11 "alternativeNames": "[\"test2\"]" 12}13runner = CliRunner()14def test_create_keystore():15 res = runner.invoke(create, ["-n", test_keystore, "-e", "eval"])16 assert res.output is not None and "error" not in res.output and test_keystore in res.output17def test_list_keystores():18 res = runner.invoke(list, ["-e", "eval"])19 assert res.output is not None and "error" not in res.output and test_keystore in res.output20def test_get_keystores():21 res = runner.invoke(get, ["-n", test_keystore, "-e", "eval"])22 assert res.output is not None and "error" not in res.output and test_keystore in res.output23def test_create_alias():24 res = runner.invoke(25 create_alias,26 [27 "-n", test_keystore, "-e", "eval", "-a", test_alias["alias"], "-k", test_alias["keySize"], 28 "--sig-alg", test_alias["sigAlg"], "-v", test_alias["certValidityInDays"], "-s", test_alias["subject"], 29 "--alternative-names", test_alias["alternativeNames"]30 ]31 )32 assert res.output is not None and "error" not in res.output 33 assert test_alias["alias"] in res.output and str(test_alias["keySize"]) in res.output34def test_get_alias():35 res = runner.invoke(get_alias, ["-n", test_keystore, "--alias-name", test_alias["alias"], "-e", "eval"])36 assert res.output is not None and "error" not in res.output 37 assert test_alias["alias"] in res.output and str(test_alias["keySize"]) in res.output38def test_list_aliases():39 res = runner.invoke(list_aliases, ["-n", test_keystore, "-e", "eval"])40 assert res.output is not None and "error" not in res.output and test_alias["alias"] in res.output41 42def test_get_cert():43 res = runner.invoke(get_cert, ["-n", test_keystore, "-e", "eval", "-a", test_alias["alias"]])44 assert res.output is not None and "error" not in res.output 45 assert test_alias["alias"] in res.output and str(test_alias["keySize"]) in res.output46def test_delete_alias():47 res = runner.invoke(delete_alias, ["-n", test_keystore, "-e", "eval", "--alias-name", test_alias["alias"]])48 assert res.output is not None and "error" not in res.output49 assert test_alias["alias"] in res.output50 res = runner.invoke(delete_alias, ["-n", test_keystore, "-e", "eval", "--alias-name", test_alias["alias"]])51 assert res.output is not None and "error" in res.output and "404" in res.output52def test_delete_keystore():53 res = runner.invoke(delete, ["-n", test_keystore, "-e", "eval"])54 assert res.output is not None and "error" not in res.output and test_keystore in res.output55 res = runner.invoke(delete, ["-n", test_keystore, "-e", "eval"])...
test_builder.py
Source:test_builder.py
2from fireant import *3test_field = Field('my_field', None, label='My Field', prefix='a', suffix='b', precision=2)4class OperationBuilderTests(TestCase):5 def test_alias_derived_from_field(self):6 def test_alias(op, args, expected_alias):7 test_op = op(*args)8 self.assertEqual(expected_alias, test_op.alias)9 with self.subTest(CumSum.__name__):10 test_alias(CumSum, [test_field], 'cumsum(my_field)')11 with self.subTest(CumMean.__name__):12 test_alias(CumMean, [test_field], 'cummean(my_field)')13 with self.subTest(CumProd.__name__):14 test_alias(CumProd, [test_field], 'cumprod(my_field)')15 with self.subTest(RollingMean.__name__):16 test_alias(RollingMean, [test_field, 3], 'rollingmean(my_field,3)')17 test_alias(RollingMean, [test_field, 7], 'rollingmean(my_field,7)')18 with self.subTest(Share.__name__):19 test_dimension = Field('my_other_field', None)20 test_alias(Share, [test_field, test_dimension], 'share(my_field,my_other_field)')21 def test_label_derived_from_field(self):22 def test_label(op, args, expected_label):23 test_op = op(*args)24 self.assertEqual(expected_label, test_op.label)25 with self.subTest(CumSum.__name__):26 test_label(CumSum, [test_field], 'CumSum(My Field)')27 with self.subTest(CumMean.__name__):28 test_label(CumMean, [test_field], 'CumMean(My Field)')29 with self.subTest(CumProd.__name__):30 test_label(CumProd, [test_field], 'CumProd(My Field)')31 with self.subTest(RollingMean.__name__):32 test_label(RollingMean, [test_field, 3], 'RollingMean(My Field,3)')33 test_label(RollingMean, [test_field, 7], 'RollingMean(My Field,7)')34 with self.subTest(Share.__name__):...
test_alias_model.py
Source:test_alias_model.py
1import pytest2from clockzy.lib.models.alias import Alias3from clockzy.lib.db.database_interface import item_exists4from clockzy.lib.test_framework.database import intratime_user_parameters, alias_parameters5from clockzy.lib.db.db_schema import ALIAS_TABLE6from clockzy.lib.handlers.codes import SUCCESS, ITEM_ALREADY_EXISTS7# Pytest tierdown fixture is executed from righ to left position8@pytest.mark.parametrize('user_parameters, alias_parameters', [(intratime_user_parameters, alias_parameters)])9def test_save_alias(add_pre_user, delete_post_user, delete_post_alias):10 test_alias = Alias(**alias_parameters)11 # Add the alias and check that 1 row has been affected (no exception when running)12 assert test_alias.save() == SUCCESS13 alias_parameters['id'] = test_alias.id14 # If we try to add the same alias, check that it can not be inserted15 assert test_alias.save() == ITEM_ALREADY_EXISTS16 # Query and check that the alias exist17 assert item_exists({'id': test_alias.id}, ALIAS_TABLE)18@pytest.mark.parametrize('user_parameters, alias_parameters', [(intratime_user_parameters, alias_parameters)])19def test_update_alias(add_pre_user, add_pre_alias, delete_post_user, delete_post_alias):20 test_alias = Alias(**alias_parameters)21 test_alias.id = add_pre_alias22 alias_parameters['id'] = test_alias.id23 # Update the alias info and check that 1 row has been affected (no exception when running)24 test_alias.alias = 'alias_updated'25 test_alias.update()26 # Query and check that the alias exist27 assert item_exists({'id': test_alias.id, 'alias': 'alias_updated'}, ALIAS_TABLE)28@pytest.mark.parametrize('user_parameters, alias_parameters', [(intratime_user_parameters, alias_parameters)])29def test_delete_alias(add_pre_user, add_pre_alias, delete_post_user):30 test_alias = Alias(**alias_parameters)31 test_alias.id = add_pre_alias32 alias_parameters['id'] = test_alias.id33 # Delete the alias info and check that 1 row has been affected (no exception when running)34 assert test_alias.delete() == SUCCESS35 # Query and check that the alias does not exist...
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!!