Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use is_not_none method in assertpy

Best Python code snippet using assertpy_python

test_csv_scenarios_updated.py

Source:test_csv_scenarios_updated.py

...59    for scn_dict in scenarios_dicts:60        # Profile Score Calculation ..................................................61        recent_p = ds.get_user_profile(user_id)62        revised_p = create_revised_profile(user_id=user_id, recent_p=recent_p)63        if is_not_none(scn_dict['KYC']):64            revised_p.has_kyc = scn_dict['KYC']65        if is_not_none(scn_dict['Military']):66            revised_p.military_service_status = ProfileMilitaryServiceStatusEnum.__getitem__(scn_dict['Military'])67        if is_not_none(scn_dict['SimCard']):68            revised_p.sim_card_ownership = scn_dict['SimCard']69        if is_not_none(scn_dict['Address']):70            revised_p.address_verification = scn_dict['Address']71        if is_not_none(scn_dict['Membership']):72            revised_p.membership_date = date.today() - timedelta(days=int(scn_dict['Membership']))73        if is_not_none(scn_dict['Recommendation']):74            revised_p.recommended_to_others_count = scn_dict['Recommendation']75        if is_not_none(scn_dict['WeightedAveStars']):76            revised_p.star_count_average = scn_dict['WeightedAveStars']77        cs.calculate_user_profile_score(recent_p=recent_p, revised_p=revised_p)78        print()79        # DoneTrade Score Calculation ..................................................80        # dt = DoneTrade(user_id=user_id)81        # if is_not_none(scn_dict['Last3MSD']):82        #     dt.timely_trades_count_of_last_3_months = scn_dict['Last3MSD']83        # if is_not_none(scn_dict['Last1YSD']):84        #     dt.timely_trades_count_between_last_3_to_12_months = scn_dict['Last1YSD']85        # if is_not_none(scn_dict['B30DayDelayLast3M']):86        #     dt.past_due_trades_count_of_last_3_months = scn_dict['B30DayDelayLast3M']87        # if is_not_none(scn_dict['B30DayDelayLast3-12M']):88        #     dt.past_due_trades_count_between_last_3_to_12_months = scn_dict['B30DayDelayLast3-12M']89        # if is_not_none(scn_dict['A30DayDelayLast3M']):90        #     dt.arrear_trades_count_of_last_3_months = scn_dict['A30DayDelayLast3M']91        # if is_not_none(scn_dict['A30DayDelay3-12M']):92        #     dt.arrear_trades_count_between_last_3_to_12_months = scn_dict['A30DayDelay3-12M']93        # if is_not_none(scn_dict['AverageDelayRatio']):94        #     dt.total_delay_days = scn_dict['AverageDelayRatio']95        # if is_not_none(scn_dict['SDealAmountRatio']):96        #     # todo: 100000000 is fix Denominator that is all_other_users_done_trades_amount, it should be change later97        #     dt.trades_total_balance = round(float(scn_dict['SDealAmountRatio']) * ALL_USERS_AVERAGE_DEAL_AMOUNT)98        # recent_dt = ds.get_user_done_trade(user_id)99        # cs.calculate_user_done_trades_score(revised_p=revised_p, recent_dt=recent_dt, revised_dt=dt)100        # ds.insert_or_update_done_trade(dt, update_flag=recent_dt.user_id is not None)101        # print()102        # UndoneTrade Score Calculation ..................................................103        # udt = UndoneTrade(user_id=user_id)104        # if is_not_none(scn_dict['NumNotDueDeal']):105        #     udt.undue_trades_count = scn_dict['NumNotDueDeal']106        # if is_not_none(scn_dict['UnfinishedB30DayDelay']):107        #     udt.past_due_trades_count = scn_dict['UnfinishedB30DayDelay']108        # if is_not_none(scn_dict['UnfinishedA30DayDelay']):109        #     udt.arrear_trades_count = scn_dict['UnfinishedA30DayDelay']110        # dt.trades_total_balance = get_zero_if_none(dt.trades_total_balance)111        # if is_not_none(scn_dict['NotDueDealAmountRatio']):112        #     udt.undue_trades_total_balance_of_last_year = round(float(scn_dict['NotDueDealAmountRatio']) * dt.trades_total_balance)113        # if is_not_none(scn_dict['UnfinishedB30Din1YRatio']):114        #     udt.past_due_trades_total_balance_of_last_year = round(float(scn_dict['UnfinishedB30Din1YRatio']) * dt.trades_total_balance)115        # if is_not_none(scn_dict['UnfinishedA30Din1YRatio']):116        #     udt.arrear_trades_total_balance_of_last_year = round(float(scn_dict['UnfinishedA30Din1YRatio']) * dt.trades_total_balance)117        # recent_udt = ds.get_user_undone_trade(user_id)118        # cs.calculate_user_undone_trades_score(revised_p=revised_p, recent_udt=recent_udt, revised_udt=udt, dt=dt)119        # ds.insert_or_update_undone_trade(udt, update_flag=recent_udt.user_id is not None)120        # print()121        # Loan Score Calculation ..................................................122        ln = Loan(user_id=user_id)123        if is_not_none(scn_dict['Loans']):124            ln.loans_total_count = scn_dict['Loans']125        ln.loans_total_balance = ALL_USERS_AVERAGE_PRINCIPAL_INTEREST_AMOUNT126        if is_not_none(scn_dict['PastDueLoans']):127            ln.past_due_loans_total_count = int(scn_dict['PastDueLoans'])128        if is_not_none(scn_dict['DelayedLoans']):129            ln.arrear_loans_total_count = int(scn_dict['DelayedLoans'])130        if is_not_none(scn_dict['DoubfulCollectionLoans']):131            ln.suspicious_loans_total_count = int(scn_dict['DoubfulCollectionLoans'])132        if is_not_none(scn_dict['MonthlyInstallments']):133            ln.monthly_installments_total_balance = float(scn_dict['MonthlyInstallments'])134        if is_not_none(scn_dict['CurrentLoanAmountRatio']):135            ln.overdue_loans_total_balance = round(float(scn_dict['CurrentLoanAmountRatio']) * ln.loans_total_balance)136        if is_not_none(scn_dict['PastDueLoanAmountRatio']):137            ln.past_due_loans_total_balance = round(float(scn_dict['PastDueLoanAmountRatio']) * ln.loans_total_balance)138        if is_not_none(scn_dict['DelayedLoanAmountRatio']):139            ln.arrear_loans_total_balance = round(float(scn_dict['DelayedLoanAmountRatio']) * ln.loans_total_balance)140        if is_not_none(scn_dict['DoubtfulCollectionAmountRatio']):141            ln.suspicious_loans_total_balance = round(float(scn_dict['DoubtfulCollectionAmountRatio']) * ln.loans_total_balance)142        recent_ln = ds.get_user_loan(user_id)143        loan_score = cs.calculate_user_loans_score(revised_p=revised_p, recent_ln=recent_ln, revised_ln=ln)144        ds.insert_or_update_loan(ln, update_flag=recent_ln.user_id is not None)145        print()146        # Cheque Score Calculation ..................................................147        ch = Cheque(user_id=user_id)148        if is_not_none(scn_dict['DishonouredChequesL3M']):149            ch.unfixed_returned_cheques_count_of_last_3_months = scn_dict['DishonouredChequesL3M']150        if is_not_none(scn_dict['DishonouredChequesL3-12M']):151            ch.unfixed_returned_cheques_count_between_last_3_to_12_months = scn_dict['DishonouredChequesL3-12M']152        if is_not_none(scn_dict['DishonouredChequesA12M']):153            ch.unfixed_returned_cheques_count_of_more_12_months = scn_dict['DishonouredChequesA12M']154        if is_not_none(scn_dict['AllDishonouredCheques']):155            ch.unfixed_returned_cheques_count_of_last_5_years = scn_dict['AllDishonouredCheques']156        if is_not_none(scn_dict['DCAmountRatio']):157            ch.unfixed_returned_cheques_total_balance = round(float(scn_dict['DCAmountRatio']) * ALL_USERS_AVERAGE_UNFIXED_RETURNED_CHEQUES_AMOUNT)158        recent_ch = ds.get_user_cheque(user_id)159        cheque_score = cs.calculate_user_cheques_score(revised_p=revised_p, recent_ch=recent_ch, revised_ch=ch)160        ds.insert_or_update_cheque(ch, update_flag=recent_ch.user_id is not None)161        # round profile scores162        # revised_p = cs.calculate_profile_rounded_score(revised_p)163        # save profile data164        ds.insert_or_update_profile(revised_p, update_flag=recent_p.user_id is not None)165def create_50_test_user_profiles_for_test_kafka_contract_messages():166    csv_file_path = '/home/mohammad-reza/vsq-docs-live/scoring/SCENARIOS/0-vscore-scenario.csv'167    # csv_file_path = '/home/mohammad-reza/vsq-docs-live/scoring/SCENARIOS/vscore-scenario-test1.csv'168    sen_dict = read_scenarios_dicts_from_csv(csv_file_path)169    user_id = 100170    calculate_score(sen_dict, user_id)...

tf_template.pyt

Source:tf_template.pyt

...9{% for conn in connectors %}10    {{conn.name}} = {{print_connector(conn)}}11{% endfor %}12{% for topic in tf.topic %}13{% if is_not_none(topic.body) %}14    @nrp.MapRobotPublisher("{{topic.name}}", Topic('{{topic.topic}}', {{topic.type}}))15{% else %}16    @nrp.MapRobotSubscriber("{{topic.name}}", Topic('{{topic.topic}}', {{topic.type}}))17{% endif %}{% endfor %}18{% for dev in tf.device %}19{% if is_not_none(dev.body) %}20    @nrp.MapSpikeSource("{{dev.name}}", {{print_neurons(dev.neurons, "nrp.brain.")}}, nrp.{{get_device_name(dev.type)}}{{print_device_config(dev)}})21{% else %}22    @nrp.MapSpikeSink("{{dev.name}}", {{print_neurons(dev.neurons, "nrp.brain.")}}, nrp.{{get_device_name(dev.type)}}{{print_device_config(dev)}})23{% endif %}{% endfor %}24{% for group in tf.deviceGroup %}25{% if is_not_none(group.body) %}26    @nrp.MapSpikeSource("{{group.name}}", {{print_neuron_group(group.neurons)}}, nrp.{{get_device_name(group.type)}}{{print_device_config(group)}})27{% else %}28    @nrp.MapSpikeSink("{{group.name}}", {{print_neuron_group(group.neurons)}}, nrp.{{get_device_name(group.type)}}{{print_device_config(group)}})29{% endif %}{% endfor %}30    @nrp.Neuron2Robot({% if is_not_none(tf.returnValue) %}Topic('{{tf.returnValue.topic}}', {{tf.returnValue.type}}){% endif %})31    def {{tf.name}}(t{% for t in tf.topic %}, {{t.name}}{%endfor%}{% for dev in tf.device %}, {{dev.name}}{%endfor%}{% for group in tf.deviceGroup %}, {{group.name}}{%endfor%}):32{% for local in tf.local %}33        {{local.name}} = {{print_expression(local.body)}}34{% endfor %}35{% for dev in tf.device %}{% if is_not_none(dev.body) %}36        {{dev.name}}.{{get_default_property(dev.type)}} = {{print_expression(dev.body)}}37{% endif %}{% endfor %}38{% for group in tf.deviceGroup %}{% if is_not_none(group.body) %}39        {{group.name}}.{{get_default_property(group.type)}} = {{print_expression(group.body)}}40{% endif %}{% endfor %}41{% for top in tf.topic %}{% if is_not_none(top.body) %}42        {{top.name}}.send_message({{print_expression(top.body)}})43{% endif %}{% endfor %}44{% if is_not_none(tf.returnValue) %}45        return {{print_expression(tf.returnValue.body)}}46{% endif %}47{% elif __builtins__.type(tf) == bibi_api_gen.Neuron2Monitor %}48    @nrp.NeuronMonitor({{print_neurons(tf.device[0].neurons, "nrp.brain.")}}, nrp.{{get_device_name(tf.device[0].type)}})49    def {{tf.name}}(t):50        return True51{% elif __builtins__.type(tf) == bibi_api_gen.Robot2Neuron %}52{% for dyn in dynamics %}53    {{dyn.name}} = {{print_synapse_dynamics(dyn)}}54{% endfor %}55{% for conn in connectors %}56    {{conn.name}} = {{print_connector(conn)}}57{% endfor %}58{% for topic in tf.topic %}59{% if is_not_none(topic.body) %}60    @nrp.MapRobotPublisher("{{topic.name}}", Topic('{{topic.topic}}', {{topic.type}}))61{% else %}62    @nrp.MapRobotSubscriber("{{topic.name}}", Topic('{{topic.topic}}', {{topic.type}}))63{% endif %}{% endfor %}64{% for dev in tf.device %}65{% if is_not_none(dev.body) %}66    @nrp.MapSpikeSource("{{dev.name}}", {{print_neurons(dev.neurons, "nrp.brain.")}}, nrp.{{get_device_name(dev.type)}}{{print_device_config(dev)}})67{% else %}68    @nrp.MapSpikeSink("{{dev.name}}", {{print_neurons(dev.neurons, "nrp.brain.")}}, nrp.{{get_device_name(dev.type)}}{{print_device_config(dev)}})69{% endif %}{% endfor %}70{% for group in tf.deviceGroup %}{% if is_not_none(group.body) %}71    @nrp.MapSpikeSource("{{group.name}}", {{print_neuron_group(group.neurons)}}, nrp.{{get_device_name(group.type)}}{{print_device_config(group)}})72{% else %}73    @nrp.MapSpikeSink("{{group.name}}", {{print_neuron_group(group.neurons)}}, nrp.{{get_device_name(group.type)}}{{print_device_config(group)}})74{% endif %}{% endfor %}75    @nrp.Robot2Neuron()76    def {{tf.name}}(t{% for topic in tf.topic %}, {{topic.name}}{%endfor%}{% for dev in tf.device %}, {{dev.name}}{%endfor%}{% for group in tf.deviceGroup %}, {{group.name}}{%endfor%}):77{% for local in tf.local %}78        {{local.name}} = {{print_expression(local.body)}}79{% endfor %}80{% for dev in tf.device %}{% if is_not_none(dev.body) %}81        {{dev.name}}.{{get_default_property(dev.type)}} = {{print_expression(dev.body)}}82{% endif %}{% endfor %}83{% for group in tf.deviceGroup %}{% if is_not_none(group.body) %}84        {{group.name}}.{{get_default_property(group.type)}} = {{print_expression(group.body)}}85{% endif %}{% endfor %}86{% for top in tf.topic %}{% if is_not_none(top.body) %}87        {{top.name}}.send_message({{print_expression(top.body)}})88{% endif %}{% endfor %}...

jtl_validator.py

Source:jtl_validator.py

1import time2from csv import DictReader3from pathlib import Path4from types import FunctionType5from typing import List, Dict6from util.jtl_convertor.validation_exception import ValidationException7from util.jtl_convertor.validation_funcs import is_not_none, is_number, is_not_blank8CONNECT = 'Connect'9HOSTNAME = 'Hostname'10LATENCY = 'Latency'11ALL_THREADS = 'allThreads'12GRP_THREADS = 'grpThreads'13BYTES = 'bytes'14SUCCESS = 'success'15THREAD_NAME = 'threadName'16RESPONSE_MESSAGE = 'responseMessage'17RESPONSE_CODE = 'responseCode'18LABEL = 'label'19ELAPSED = 'elapsed'20TIME_STAMP = 'timeStamp'21METHOD = 'method'22SUPPORTED_JTL_HEADER: List[str] = [TIME_STAMP, ELAPSED, LABEL, SUCCESS]23VALIDATION_FUNCS_BY_COLUMN: Dict[str, List[FunctionType]] = {24    TIME_STAMP: [is_not_none, is_number],25    ELAPSED: [is_not_none, is_number],26    LABEL: [is_not_blank],27    RESPONSE_CODE: [],28    RESPONSE_MESSAGE: [],29    THREAD_NAME: [],30    SUCCESS: [],31    BYTES: [is_not_none, is_number],32    GRP_THREADS: [is_not_none, is_number],33    ALL_THREADS: [is_not_none, is_number],34    LATENCY: [],35    HOSTNAME: [],36    CONNECT: [],37    METHOD: [],38}39def get_validation_func(column: str) -> List[FunctionType]:40    validation_funcs = VALIDATION_FUNCS_BY_COLUMN.get(column)41    if validation_funcs is None:42        raise Exception(f"There is no validation function for column: [{column}]")43    return validation_funcs44def __validate_value(column: str, value: str) -> None:45    validation_funcs = get_validation_func(column)46    try:47        for validation_func in validation_funcs:48            validation_func(value)49    except ValidationException as e:50        raise ValidationException(f"Column: [{column}]. Validation message: {str(e)}")51def __validate_row(jtl_row: Dict) -> None:52    for column, value in jtl_row.items():53        __validate_value(column, str(value))54def __validate_header(headers: List) -> None:55    for header in SUPPORTED_JTL_HEADER:56        if header not in headers:57            __raise_validation_error(f"Headers is not correct. Required headers is {SUPPORTED_JTL_HEADER}. "58                                     f"{header} is missed")59def __raise_validation_error(error_msg: str) -> None:60    raise ValidationException(error_msg)61def __validate_rows(reader) -> None:62    for file_row_num, jtl_row in enumerate(reader, 2):63        try:64            __validate_row(jtl_row)65        except ValidationException as e:66            __raise_validation_error(f"File row number: {file_row_num}. {str(e)}")67def validate(file_path: Path) -> None:68    print(f'Started validating jtl file: {file_path}')69    start_time = time.time()70    try:71        with file_path.open(mode='r') as f:72            reader: DictReader = DictReader(f)73            __validate_header(reader.fieldnames)74            __validate_rows(reader)75    except (ValidationException, FileNotFoundError) as e:76        raise SystemExit(f"ERROR: Validation failed. File path: [{file_path}]. Validation details: {str(e)}")...

exec.py

Source:exec.py

...13                        help='starting date. format : 2000-01-01')14    # for memory DB15    parser.add_argument('--dbfile', nargs='?', type=str, help='DB file')16    return parser17def is_not_none(target):18    return target is not None19def main():20    parser = define_parser()21    args = parser.parse_args()22    parameter_check_with_condition(23        parser,24        ("key", args.key, is_not_none),25        ("start_date", args.start_date, is_not_none),26        ("end_date", args.end_date, is_not_none)27    )28    dbtype, dbfilename = (DBType.sqlite3, args.dbfile) if args.dbfile is not None else \29        (DBType.memory, None)30    db_session_open(31        db_session_string=db_session_string_definition(dbtype,file_path=dbfilename),...

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.