Best Python code snippet using slash
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),...
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!!