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

How to use test_alias method in pandera

Best Python code snippet using pandera_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...

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.