Best Python code snippet using autotest_python
sanity_checks.py
Source:sanity_checks.py
...45# rules['UNKNOWN_DAE_FORMAT'] = "Le format du numéro douanier semble incorrect"46# rules['UNKNOWN_DOUBLE_COUNTING_CERTIFICATE'] = "Le certificat double compte est inconnu"47# rules['EXPIRED_DOUBLE_COUNTING_CERTIFICATE'] = "Le certificat double n'est plus valide"48# rules['POTENTIAL_DUPLICATE'] = "Doublon potentiel détecté. Un autre lot avec le même numéro douanier, biocarburant, matière première, volume et caractéristiques GES existe."49# def generic_error(error, **kwargs):50# d = {51# 'display_to_creator': True,52# 'display_to_admin': True,53# 'error': error,54# }55# d.update(kwargs)56# return GenericError(**d)57# def bulk_sanity_checks_old_deprecated(txs, prefetched_data, background=True):58# results = []59# errors = []60# if background == True:61# db.connections.close_all()62# # cleanup previous errors63# GenericError.objects.filter(tx__in=txs).delete()64# for tx in txs:65# try:66# lot_ok, tx_ok, is_sane, validation_errors = sanity_check(tx, prefetched_data)67# errors += validation_errors68# results.append((lot_ok, tx_ok, is_sane))69# except:70# traceback.print_exc()71# GenericError.objects.bulk_create(errors, batch_size=1000)72# return results73# def check_certificates(prefetched_data, tx, errors):74# # PRODUCTION SITE CERTIFICATES75# if tx.lot.production_site_is_in_carbure:76# if not tx.lot.carbure_production_site_reference:77# errors.append(generic_error(error='NO_PRODSITE_CERT', tx=tx, field='carbure_production_site_reference'))78# else:79# cert = tx.lot.carbure_production_site_reference.upper()80# if cert not in prefetched_data['certificates']:81# errors.append(generic_error(error='UNKNOWN_PRODSITE_CERT', tx=tx, field='carbure_production_site_reference'))82# else:83# # certificate is set and exists. is it valid?84# c = prefetched_data['certificates'][cert]85# if c.valid_until < tx.delivery_date:86# errors.append(generic_error(error='EXPIRED_PRODSITE_CERT', tx=tx, field='carbure_production_site_reference'))87# else:88# if not tx.lot.unknown_production_site_reference:89# errors.append(generic_error(error='NO_PRODSITE_CERT', tx=tx, field='unknown_production_site_reference'))90# else:91# cert = tx.lot.unknown_production_site_reference.upper()92# if cert not in prefetched_data['certificates']:93# errors.append(generic_error(error='UNKNOWN_PRODSITE_CERT', tx=tx, field='unknown_production_site_reference'))94# else:95# # certificate is set and exists. is it valid?96# c = prefetched_data['certificates'][cert]97# if c.valid_until < tx.delivery_date:98# errors.append(generic_error(error='EXPIRED_PRODSITE_CERT', tx=tx, field='unknown_production_site_reference')) 99# # SUPPLIER CERT100# if not tx.lot.producer_is_in_carbure:101# if not tx.lot.unknown_supplier_certificate:102# errors.append(generic_error(error='NO_SUPPLIER_CERT', tx=tx, field='unknown_supplier_certificate'))103# else:104# cert = tx.lot.unknown_supplier_certificate.upper()105# if cert not in prefetched_data['certificates']:106# errors.append(generic_error(error='UNKNOWN_SUPPLIER_CERT', tx=tx, field='unknown_supplier_certificate'))107# else:108# # certificate is set and exists. is it valid?109# c = prefetched_data['certificates'][cert]110# if c.valid_until < tx.delivery_date:111# errors.append(generic_error(error='EXPIRED_SUPPLIER_CERT', tx=tx, field='unknown_supplier_certificate')) 112# # VENDOR CERT - NOT REQUIRED WHEN TX UPLOADED BY OPERATOR113# if not tx.lot.added_by.entity_type == Entity.OPERATOR:114# if not tx.carbure_vendor_certificate:115# errors.append(generic_error(error='NO_VENDOR_CERT', tx=tx, field='carbure_vendor_certificate'))116# else:117# cert = tx.carbure_vendor_certificate.upper()118# if cert not in prefetched_data['certificates']:119# errors.append(generic_error(error='UNKNOWN_VENDOR_CERT', tx=tx, field='carbure_vendor_certificate'))120# else:121# # certificate is set and exists. is it valid?122# c = prefetched_data['certificates'][cert]123# if c.valid_until < tx.delivery_date:124# errors.append(generic_error(error='EXPIRED_VENDOR_CERT', tx=tx, field='carbure_vendor_certificate')) 125# # DOUBLE COUNTING CERTIFICATES126# if tx.lot.matiere_premiere and tx.lot.matiere_premiere.is_double_compte:127# # identify where the certificate is (attached to prod site or attached to lot)128# dc_cert = ''129# if tx.lot.carbure_production_site:130# if not tx.lot.carbure_production_site.dc_reference:131# errors.append(generic_error(error='MISSING_REF_DBL_COUNTING', tx=tx, field='dc_reference'))132# else:133# dc_cert = tx.lot.carbure_production_site.dc_reference.strip()134# else:135# if not tx.lot.unknown_production_site_dbl_counting:136# errors.append(generic_error(error='MISSING_REF_DBL_COUNTING', tx=tx, field='unknown_production_site_dbl_counting'))137# else:138# dc_cert = tx.lot.unknown_production_site_dbl_counting.strip()139# if dc_cert == '':140# # should not happen unless above cert is '' after strip()141# errors.append(generic_error(error='MISSING_REF_DBL_COUNTING', tx=tx, field='unknown_production_site_dbl_counting'))142# else:143# if dc_cert not in prefetched_data['double_counting_certificates']:144# errors.append(generic_error(error='UNKNOWN_DOUBLE_COUNTING_CERTIFICATE', tx=tx, field='dc_reference'))145# else:146# dcc = prefetched_data['double_counting_certificates'][dc_cert]147# if dcc.valid_until < tx.delivery_date:148# errors.append(generic_error(error='EXPIRED_DOUBLE_COUNTING_CERTIFICATE', tx=tx))149# return errors150# def sanity_check(tx, prefetched_data):151# lot = tx.lot152# is_sane = True153# errors = []154# # make sure all mandatory fields are set155# tx_valid = tx_is_valid(tx, prefetched_data)156# lot_valid = lot_is_valid(tx)157# if not lot_valid or not tx_valid:158# # without mandatory fields, we cannot start analyzing for sanity errors. return immediately159# is_sane = False160# return lot_valid, tx_valid, is_sane, errors161# if tx.is_mac and lot.biocarburant and lot.biocarburant.code not in ['ED95', 'B100', 'ETH', 'EMHV', 'EMHU']:162# errors.append(generic_error(error='MAC_BC_WRONG', tx=tx, is_blocking=True, fields=['biocarburant_code', 'mac']))163# # check volume164# if lot.volume < 2000 and not tx.is_mac:165# errors.append(generic_error(error='VOLUME_FAIBLE', tx=tx, field='volume'))166# # réduction de GES167# if tx.delivery_date >= july1st2021:168# # RED II169# if lot.ghg_reduction_red_ii >= 100:170# errors.append(generic_error(error='GHG_REDUC_SUP_100', tx=tx))171# elif lot.ghg_reduction_red_ii > 99:172# errors.append(generic_error(error='GHG_REDUC_SUP_99', tx=tx))173# elif lot.ghg_reduction_red_ii < 50:174# is_sane = False175# errors.append(generic_error(error='GHG_REDUC_INF_50', tx=tx, is_blocking=True))176# else:177# # all good178# pass179# else:180# if lot.ghg_reduction >= 100:181# errors.append(generic_error(error='GHG_REDUC_SUP_100', tx=tx))182# elif lot.ghg_reduction > 99:183# errors.append(generic_error(error='GHG_REDUC_SUP_99', tx=tx))184# elif lot.ghg_reduction < 50:185# is_sane = False186# errors.append(generic_error(error='GHG_REDUC_INF_50', tx=tx, is_blocking=True))187# else:188# # all good189# pass190# if lot.etd == 0:191# is_sane = False192# errors.append(generic_error(error='GHG_ETD_0', tx=tx, is_blocking=True, field='etd'))193# if lot.ep == 0:194# is_sane = False195# errors.append(generic_error(error='GHG_EP_0', tx=tx, is_blocking=True, field='ep'))196# if lot.el < 0:197# errors.append(generic_error(error='GHG_EL_NEG', tx=tx, field='el'))198 199# commissioning_date = lot.carbure_production_site.date_mise_en_service if lot.carbure_production_site else lot.unknown_production_site_com_date200# if commissioning_date and isinstance(commissioning_date, datetime.datetime) or isinstance(commissioning_date, datetime.date):201# #if tx.delivery_date and tx.delivery_date > july1st2021:202# # if commissioning_date > oct2015 and lot.ghg_reduction_red_ii < 60:203# # is_sane = False204# # errors.append(generic_error(error='GHG_REDUC_INF_60', tx=tx, is_blocking=True))205# # if commissioning_date >= jan2021 and lot.ghg_reduction_red_ii < 65:206# # is_sane = False207# # errors.append(generic_error(error='GHG_REDUC_INF_65', tx=tx, is_blocking=True))208# #else:209# if commissioning_date > oct2015 and lot.ghg_reduction < 60:210# is_sane = False211# errors.append(generic_error(error='GHG_REDUC_INF_60', tx=tx, is_blocking=True))212# if commissioning_date >= jan2021 and lot.ghg_reduction < 65:213# is_sane = False214# errors.append(generic_error(error='GHG_REDUC_INF_65', tx=tx, is_blocking=True))215# # provenance des matieres premieres216# if lot.matiere_premiere and lot.pays_origine:217# if lot.matiere_premiere.code == "RESIDUS_VINIQUES":218# errors.append(generic_error(error='DEPRECATED_MP', tx=tx, field='matiere_premiere'))219# if lot.matiere_premiere.category == 'CONV' and lot.eec == 0:220# errors.append(generic_error(error='GHG_EEC_0', tx=tx, extra="GES Culture 0 pour MP conventionnelle (%s)" % (lot.matiere_premiere.name), field='eec'))221# if lot.matiere_premiere.code == 'SOJA':222# if lot.pays_origine.code_pays not in ['US', 'AR', 'BR', 'UY', 'PY']:223# errors.append(generic_error(error='PROVENANCE_MP', tx=tx, extra="%s de %s" % (lot.matiere_premiere.name, lot.pays_origine.name), field='pays_origine_code'))224# elif lot.matiere_premiere.code == 'HUILE_PALME':225# if lot.pays_origine.code_pays not in ['ID', 'MY', 'HN']:226# errors.append(generic_error(error='PROVENANCE_MP', tx=tx, extra="%s de %s" % (lot.matiere_premiere.name, lot.pays_origine.name), field='pays_origine_code'))227# elif lot.matiere_premiere.code == 'COLZA':228# if lot.pays_origine.code_pays not in ['US', 'CA', 'AU', 'UA', 'CN', 'IN', 'DE', 'FR', 'PL', 'UK'] and not lot.pays_origine.is_in_europe:229# errors.append(generic_error(error='PROVENANCE_MP', tx=tx, extra="%s de %s" % (lot.matiere_premiere.name, lot.pays_origine.name), field='pays_origine_code'))230# elif lot.matiere_premiere.code == 'CANNE_A_SUCRE':231# if lot.pays_origine.code_pays not in ['BR', 'BO']:232# errors.append(generic_error(error='PROVENANCE_MP', tx=tx, extra="%s de %s" % (lot.matiere_premiere.name, lot.pays_origine.name), field='pays_origine_code'))233# elif lot.matiere_premiere.code == 'MAIS':234# if not lot.pays_origine.is_in_europe and lot.pays_origine.code_pays not in ['US', 'UA']:235# errors.append(generic_error(error='PROVENANCE_MP', tx=tx, extra="%s de %s" % (lot.matiere_premiere.name, lot.pays_origine.name), field='pays_origine_code'))236# elif lot.matiere_premiere.code == 'BETTERAVE':237# if not lot.pays_origine.is_in_europe:238# errors.append(generic_error(error='PROVENANCE_MP', tx=tx, extra="%s de %s" % (lot.matiere_premiere.name, lot.pays_origine.name), field='pays_origine_code'))239# else:240# pass241# if lot.biocarburant and lot.matiere_premiere:242# # consistence des matieres premieres avec biocarburant243# if lot.biocarburant.is_alcool and lot.matiere_premiere.compatible_alcool is False:244# is_sane = False245# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="%s issu de fermentation et %s n'est pas fermentescible" % (lot.biocarburant.name, lot.matiere_premiere.name), fields=['biocarburant_code', 'matiere_premiere_code']))246# if lot.biocarburant.is_graisse and lot.matiere_premiere.compatible_graisse is False:247# is_sane = False248# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="Matière première (%s) incompatible avec Esthers Méthyliques" % (lot.matiere_premiere.name), fields=['biocarburant_code', 'matiere_premiere_code']))249# # double comptage, cas specifiques250# if lot.matiere_premiere.is_double_compte:251# in_carbure_without_dc = lot.production_site_is_in_carbure and lot.carbure_production_site and not lot.carbure_production_site.dc_reference252# not_in_carbure_without_dc = not lot.production_site_is_in_carbure and not lot.unknown_production_site_dbl_counting253# if in_carbure_without_dc or not_in_carbure_without_dc:254# is_sane = False255# errors.append(generic_error(error='MISSING_REF_DBL_COUNTING', tx=tx, is_blocking=True, extra="%s de %s" % (lot.biocarburant.name, lot.matiere_premiere.name), field='production_site_dbl_counting'))256# if lot.biocarburant.is_graisse:257# if lot.biocarburant.code == 'EMHU' and lot.matiere_premiere.code != 'HUILE_ALIMENTAIRE_USAGEE':258# is_sane = False259# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="%s doit être à base d'huiles alimentaires usagées" % (lot.biocarburant.name), fields=['biocarburant_code', 'matiere_premiere_code']))260# if lot.biocarburant.code == 'EMHV' and lot.matiere_premiere.code not in ['COLZA', 'TOURNESOL', 'SOJA', 'HUILE_PALME']:261# is_sane = False262# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="%s doit être à base de végétaux (Colza, Tournesol, Soja, Huile de Palme)" % (lot.biocarburant.name), fields=['biocarburant_code', 'matiere_premiere_code']))263# if lot.biocarburant.code == 'EMHA' and lot.matiere_premiere.code not in ['HUILES_OU_GRAISSES_ANIMALES_CAT1_CAT2', 'HUILES_OU_GRAISSES_ANIMALES_CAT3']:264# is_sane = False265# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="%s doit être à base d'huiles ou graisses animales" % (lot.biocarburant.name), fields=['biocarburant_code', 'matiere_premiere_code']))266# if lot.matiere_premiere.code in ['HUILES_OU_GRAISSES_ANIMALES_CAT1_CAT2', 'HUILES_OU_GRAISSES_ANIMALES_CAT3'] and lot.biocarburant.code not in ['EMHA', 'HOE', 'HOG']:267# is_sane = False268# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="Des huiles ou graisses animales ne peuvent donner que des EMHA ou HOG/HOE", fields=['biocarburant_code', 'matiere_premiere_code']))269# if lot.matiere_premiere.code == 'HUILE_ALIMENTAIRE_USAGEE' and lot.biocarburant.code not in ['EMHU', 'HOE', 'HOG']:270# is_sane = False271# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="Des huiles alimentaires usagées ne peuvent donner que des EMHU ou HOG/HOE", fields=['biocarburant_code', 'matiere_premiere_code']))272# if lot.matiere_premiere.code in ['MAIS', 'BLE', 'BETTERAVE', 'CANNE_A_SUCRE', 'RESIDUS_VINIQUES', 'LIES_DE_VIN', 'MARC_DE_RAISIN'] and lot.biocarburant.code not in ['ETH', 'ETBE', 'ED95']:273# is_sane = False274# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="Maïs, Blé, Betterave, Canne à Sucre ou Résidus Viniques ne peuvent créer que de l'Ãthanol ou ETBE", fields=['biocarburant_code', 'matiere_premiere_code']))275# if not lot.matiere_premiere.is_huile_vegetale and lot.biocarburant.code in ['HVOE', 'HVOG']:276# is_sane = False277# errors.append(generic_error(error='MP_BC_INCOHERENT', tx=tx, is_blocking=True, extra="Un HVO doit provenir d'huiles végétales uniquement. Pour les autres huiles hydrotraitées, voir la nomenclature HOE/HOG", fields=['biocarburant_code', 'matiere_premiere_code']))278# # configuration279# if lot.matiere_premiere and lot.carbure_production_site:280# if lot.carbure_production_site.name in prefetched_data['production_sites']:281# mps = [psi.matiere_premiere for psi in prefetched_data['production_sites'][lot.carbure_production_site.name].productionsiteinput_set.all()]282# if lot.matiere_premiere not in mps:283# errors.append(generic_error(error='MP_NOT_CONFIGURED', tx=tx, display_to_recipient=False, field='matiere_premiere_code'))284# if lot.biocarburant and lot.carbure_production_site:285# if lot.carbure_production_site.name in prefetched_data['production_sites']:286# bcs = [pso.biocarburant for pso in prefetched_data['production_sites'][lot.carbure_production_site.name].productionsiteoutput_set.all()]287# if lot.biocarburant not in bcs:288# errors.append(generic_error(error='BC_NOT_CONFIGURED', tx=tx, display_to_recipient=False, field='biocarburant_code'))289# if tx.carbure_client:290# if tx.carbure_client.id not in prefetched_data['depotsbyentity']:291# # not a single delivery sites linked to entity292# errors.append(generic_error(error='DEPOT_NOT_CONFIGURED', tx=tx, display_to_recipient=True, display_to_creator=False, field='delivery_site'))293# else:294# # some delivery sites linked to entity295# if tx.carbure_delivery_site and tx.carbure_delivery_site.id not in prefetched_data['depotsbyentity'][tx.carbure_client.id]:296# # this specific delivery site is not linked297# errors.append(generic_error(error='DEPOT_NOT_CONFIGURED', tx=tx, display_to_recipient=True, display_to_creator=False, field='delivery_site'))298# # CERTIFICATES CHECK299# check_certificates(prefetched_data, tx, errors)300# if tx.lot.producer_is_in_carbure and tx.lot.added_by != tx.lot.carbure_producer and not tx.lot.parent_lot:301# is_sane = False302# errors.append(generic_error(error='NOT_ALLOWED', tx=tx, is_blocking=True))303 304# # transaction is not a MAC, is going to france, and delivery_site is unknown305# if not tx.is_mac and tx.unknown_delivery_site_country and tx.unknown_delivery_site_country.code_pays == 'FR' and not tx.carbure_delivery_site:306# is_sane = False307# errors.append(GenericError(tx=tx, field='unknown_delivery_site', error="UNKNOWN_DELIVERY_SITE", extra="Site de livraison non reconnu", value=tx.unknown_delivery_site, display_to_creator=True, is_blocking=True))308# # transaction is not a MAC, client is unknown309# if not tx.client_is_in_carbure and not tx.is_mac:310# # destination FRANCE -> blocking311# if (tx.carbure_delivery_site and tx.carbure_delivery_site.country.code_pays == 'FR') or (tx.unknown_delivery_site_country and tx.unknown_delivery_site_country.code_pays == 'FR'):312# is_sane = False313# errors.append(GenericError(tx=tx, field='client', error="UNKNOWN_CLIENT", extra="Livraison en France - Client inconnu", value=tx.unknown_client, display_to_creator=True, is_blocking=True))314# # otherwise, simple warning315# else:316# errors.append(generic_error(error='UNKNOWN_CLIENT', tx=tx, extra="Client inconnu", display_to_recipient=False, field="unknown_client"))317# if tx.delivery_date > future:318# is_sane = False319# errors.append(GenericError(tx=tx, field='delivery_date', error="DELIVERY_IN_THE_FUTURE", extra="La date de livraison est dans le futur", value=tx.delivery_date, display_to_creator=True, is_blocking=True))320# if not tx.is_mac:321# if not dae_pattern.match(tx.dae):322# errors.append(GenericError(tx=tx, field='dae', error="UNKNOWN_DAE_FORMAT", extra="Caractère non-standard trouvé dans le numéro douanier.", value=tx.dae, display_to_creator=True, is_blocking=False))323# return lot_valid, tx_valid, is_sane, errors324# def lot_is_valid(tx):325# lot = tx.lot326# is_valid = True327# errors = []328# if not lot.volume:329# errors.append(generic_error(error='MISSING_VOLUME', tx=tx, field='volume', extra='Veuillez renseigner le volume', is_blocking=True))330# is_valid = False331# if not lot.parent_lot:332# if not lot.biocarburant:333# errors.append(generic_error(error='MISSING_BIOFUEL', tx=tx, field='biocarburant_code', extra='Veuillez renseigner le type de biocarburant', is_blocking=True))334# is_valid = False335# if not lot.matiere_premiere:336# errors.append(generic_error(error='MISSING_FEEDSTOCK', tx=tx, field='matiere_premiere_code', extra='Veuillez renseigner la matière première', is_blocking=True))337# is_valid = False338# if lot.producer_is_in_carbure and lot.carbure_production_site is None:339# extra = 'Site de production %s inconnu pour %s' % (lot.unknown_production_site, lot.carbure_producer.name)340# errors.append(generic_error(error='UNKNOWN_PRODUCTION_SITE', tx=tx, field='carbure_production_site', extra=extra, is_blocking=True))341# is_valid = False342# if not lot.production_site_is_in_carbure:343# if not lot.unknown_production_site_com_date:344# extra = "Veuillez renseigner la date de mise en service de l'usine"345# errors.append(generic_error(error='MISSING_PRODUCTION_SITE_COMDATE', tx=tx, field='unknown_production_site_com_date', extra=extra, is_blocking=True))346# is_valid = False347# if len(errors):348# GenericError.objects.bulk_create(errors)349# return is_valid350# def tx_is_valid(tx, prefetched_data):351# is_valid = True352# today = datetime.date.today()353# errors = []354# # make sure all mandatory fields are set355# if not tx.dae:356# extra = 'DAE manquant'357# errors.append(generic_error(error='MISSING_DAE', tx=tx, field='dae', extra=extra, is_blocking=True))358# is_valid = False359# if not tx.is_mac:360# if not tx.delivery_site_is_in_carbure and not tx.unknown_delivery_site:361# extra = 'Site de livraison manquant'362# errors.append(generic_error(error='MISSING_UNKNOWN_DELIVERY_SITE', tx=tx, field='unknown_delivery_site', extra=extra, is_blocking=True))363# is_valid = False364# if tx.delivery_site_is_in_carbure and not tx.carbure_delivery_site:365# extra = 'Site de livraison manquant'366# errors.append(generic_error(error='MISSING_CARBURE_DELIVERY_SITE', tx=tx, field='carbure_delivery_site', extra=extra, is_blocking=True))367# is_valid = False368# if not tx.delivery_date or tx.delivery_date is None:369# extra = 'Date de livraison manquante'370# errors.append(generic_error(error='MISSING_DELIVERY_DATE', tx=tx, field='delivery_date', extra=extra, is_blocking=True))371# is_valid = False372# else:373# if (tx.delivery_date - today) > datetime.timedelta(days=3650) or (tx.delivery_date - today) < datetime.timedelta(days=-3650):374# extra = "Date incorrecte [%s]" % (tx.delivery_date.strftime('%d/%m/%Y'))375# errors.append(generic_error(error='WRONG_DELIVERY_DATE', tx=tx, field='delivery_date', extra=extra, is_blocking=True))376# is_valid = False377# if tx.client_is_in_carbure and not tx.carbure_client:378# extra = 'Veuillez renseigner un client'379# errors.append(generic_error(error='MISSING_CARBURE_CLIENT', tx=tx, field='carbure_client', extra=extra, is_blocking=True))380# is_valid = False381# if not tx.client_is_in_carbure and not tx.unknown_client:382# extra = 'Veuillez renseigner un client'383# errors.append(generic_error(error='MISSING_UNKNOWN_CLIENT', tx=tx, field='unknown_client', extra=extra, is_blocking=True))384# is_valid = False385# if not tx.delivery_site_is_in_carbure and not tx.unknown_delivery_site and not tx.is_mac:386# extra = 'Veuillez renseigner un site de livraison'387# errors.append(generic_error(error='MISSING_UNKNOWN_DELIVERY_SITE', tx=tx, field='unknown_delivery_site', extra=extra, is_blocking=True))388# is_valid = False389# if not tx.delivery_site_is_in_carbure and not tx.unknown_delivery_site_country and not tx.is_mac:390# extra = 'Veuillez renseigner un pays de livraison'391# errors.append(generic_error(error='MISSING_UNKNOWN_DELIVERY_SITE_COUNTRY', tx=tx, field='unknown_delivery_site_country', extra=extra, is_blocking=True))392# is_valid = False393# if tx.unknown_delivery_site_country and tx.unknown_delivery_site_country.is_in_europe and tx.lot.pays_origine is None:394# extra = "Veuillez renseigner le pays d'origine de la matière première - Marché européen"395# errors.append(generic_error(error='MISSING_FEEDSTOCK_COUNTRY_OF_ORIGIN', tx=tx, field='pays_origine_code', extra=extra, is_blocking=True))396# is_valid = False397# if tx.carbure_delivery_site and tx.carbure_delivery_site.country.is_in_europe and tx.lot.pays_origine is None:398# extra = "Veuillez renseigner le pays d'origine de la matière première - Marché européen"399# errors.append(generic_error(error='MISSING_FEEDSTOCK_COUNTRY_OF_ORIGIN', tx=tx, field='pays_origine_code', extra=extra, is_blocking=True))400# is_valid = False401# if tx.carbure_client and tx.carbure_client.entity_type == Entity.OPERATOR:402# # client is an operator403# # make sure we have a certificate404# if not tx.carbure_vendor_certificate and not tx.lot.unknown_supplier_certificate:405# # if I am an operator, I am the one who uploaded the file. request the certificate in excel file406# if tx.lot.added_by.entity_type == Entity.OPERATOR:407# extra = "Veuillez renseigner le certificat du fournisseur"408# errors.append(generic_error(error='MISSING_SUPPLIER_CERTIFICATE', tx=tx, field='unknown_supplier_certificate', extra=extra, is_blocking=True))409# is_valid = False410# else:411# # I am a producer or trader412# # I need to either set the certificate in the file or add them in my account413# extra = "Veuillez renseigner votre certificat de fournisseur ou ajouter un certificat sur votre compte"414# errors.append(generic_error(error='MISSING_SUPPLIER_CERTIFICATE', tx=tx, field='carbure_vendor_certificate', extra=extra, is_blocking=True))415# is_valid = False416# if len(errors):417# GenericError.objects.bulk_create(errors)...
forms.py
Source:forms.py
1# File encoding: utf-82from django.core.exceptions import ObjectDoesNotExist3from django import forms4from django.forms.extras.widgets import SelectDateWidget5import bugtracker.models as models6from django.contrib.auth.models import User7from django.contrib.admin.widgets import FilteredSelectMultiple8import re9fio_regexp = r'(?u)\w+(-\w+)?'10num_regexp = r'(^\d+$)'11kiril_regexp = re.compile(u'[Ð-Яа-Ñ][а-Ñ\s]+', re.UNICODE)12password_regexp = re.compile(u'^[Ð-Яа-Ñ0-9a-zA-Z]+$', re.UNICODE)13sloc_regexp = r'(^\d+$)'14generic_error = {"invalid": "ÐепÑавилÑно Ð²Ð²ÐµÐ´ÐµÐ½Ñ Ð´Ð°Ð½Ð½Ñе"}15class BugForm(forms.ModelForm):16 short_description = forms.CharField(label="ÐÑаÑкое опиÑание", max_length=100)17 severity = forms.CharField(label="ÐÑиÑиÑноÑÑÑ", widget=forms.RadioSelect(choices=models.Bug.SEVERITY_CHOICES))18 finding_description=forms.CharField(label="Ðак бÑл полÑÑен",widget=forms.Textarea,max_length=600)19 full_description=forms.CharField(label="ÐпиÑание",widget=forms.Textarea,max_length=600)20 #file_comment=forms.CharField(label="ÐоммеÑаÑиии к ÑайлÑ" ,widget=forms.Textarea, required=False,max_length=150)21 class Meta:22 model = models.Bug23 exclude = ['tester', 'status', 'status_comment', 'project']24class BugDetail(forms.ModelForm):25 status = forms.CharField(label="СÑаÑÑÑ", widget=forms.RadioSelect(choices=models.Bug.STATUS_CHOICES))26 status_comment = forms.CharField(label="ÐÑимеÑание", widget=forms.Textarea, required=False, max_length=300)27 class Meta:28 model = models.Bug29 fields=['status','status_comment']30class ProjectForm(forms.ModelForm):31 name = forms.CharField(label='Ðазвание', max_length=50)32 size = forms.RegexField(label='Ð Ð°Ð·Ð¼ÐµÑ Ð² SLOC', max_length=50, regex=sloc_regexp, error_messages=generic_error )33 program_language = forms.ModelMultipleChoiceField(label="ЯÐ", queryset=models.ProgramLang.objects.all(), widget=FilteredSelectMultiple(u'ЯÐ', False))34 document_languages = forms.ModelMultipleChoiceField(label="ЯзÑк докÑменÑаÑии", queryset=models.Language.objects.all(), widget=FilteredSelectMultiple(u'ЯзÑки', False))35 project_description = forms.CharField(label='ÐпиÑание пÑоекÑа', widget=forms.Textarea, required=False, max_length=300)36 def clean_name(self):37 name = self.cleaned_data.get('name')38 if models.Project.objects.filter(name=name).count() > 0:39 raise forms.ValidationError('ÐÑÐ¾ÐµÐºÑ Ñ Ñаким названием Ñже ÑÑÑеÑÑвÑеÑ')40 return name41 class Meta:42 model = models.Project43 fields = ['name', 'size', 'program_language', 'document_languages', 'project_description']44 exclude = ['customer', 'testers']45class UserForm(forms.ModelForm):46 email = forms.EmailField(label='ÐонÑакÑнÑй E-mail', max_length=50)47 password = forms.RegexField(label='ÐаÑолÑ', widget=forms.PasswordInput(render_value=False), regex=password_regexp, max_length=30, min_length=5, error_messages=generic_error)48 password_confirm = forms.RegexField(label='ÐодÑвеÑждение паÑолÑ', widget=forms.PasswordInput(render_value=False), regex=password_regexp, max_length=30, min_length=5, error_messages=generic_error)49 def clean_email(self):50 email = self.cleaned_data['email']51 if User.objects.filter(username=email).count() > 0:52 # raise forms.ValidationError('ÐолÑзоваÑÐµÐ»Ñ Ñ Ñаким адÑеÑом Ñже ÑÑÑеÑÑвÑеÑ')53 raise forms.ValidationError('ÐепÑавилÑно Ð²Ð²ÐµÐ´ÐµÐ½Ñ Ð´Ð°Ð½Ð½Ñе')54 return email55 56 def clean_password_confirm(self):57 cleaned_data = self.cleaned_data58 password = cleaned_data.get('password')59 password_confirm = cleaned_data.get('password_confirm')60 if password != password_confirm:61 # raise forms.ValidationError('ÐаÑÐ¾Ð»Ñ Ð¸ подÑвеÑждение не ÑовпадаÑÑ')62 raise forms.ValidationError('ÐепÑавилÑно Ð²Ð²ÐµÐ´ÐµÐ½Ñ Ð´Ð°Ð½Ð½Ñе')63 return password_confirm64class TesterForm(forms.Form):65 email = forms.EmailField(label='E-mail', max_length=50)66 password = forms.RegexField(label='ÐаÑолÑ', widget=forms.PasswordInput(render_value=False), regex=password_regexp, max_length=30, min_length=5, error_messages=generic_error)67 password_confirm = forms.RegexField(label='ÐодÑвеÑждение паÑолÑ', widget=forms.PasswordInput(render_value=False), regex=password_regexp, max_length=30, min_length=5, error_messages=generic_error)68 last_name = forms.RegexField(label="ФамилиÑ", max_length=80, regex=fio_regexp, error_messages=generic_error)69 first_name = forms.RegexField(label="ÐмÑ", max_length=30, regex=fio_regexp, error_messages=generic_error)70 second_name = forms.RegexField(label="ÐÑÑеÑÑво", max_length=30, required=False, regex=fio_regexp, error_messages=generic_error)71 description = forms.CharField(label="Ð Ñебе", widget=forms.Textarea, required=False, max_length=300)72 os = forms.ModelMultipleChoiceField(label="ÐпеÑаÑионнÑе ÑиÑÑемÑ", queryset=models.OSystem.objects.all(), widget=FilteredSelectMultiple(u'ÐС', False))73 #os = forms.ModelMultipleChoiceField(queryset=models.OSystem.objects.all())74 program_languages = forms.ModelMultipleChoiceField(label="ЯзÑки пÑогÑаммиÑованиÑ", queryset=models.ProgramLang.objects.all(), widget=FilteredSelectMultiple(u'ÑзÑки', False))75 testing_types = forms.ModelMultipleChoiceField(label="Ð¢Ð¸Ð¿Ñ ÑеÑÑиÑованиÑ", queryset=models.TestingType.objects.all(), widget=FilteredSelectMultiple(u'ÑипÑ', False))76 browsers = forms.ModelMultipleChoiceField(label="ÐÑаÑзеÑÑ", queryset=models.Browser.objects.all(), widget=FilteredSelectMultiple(u'бÑаÑзеÑÑ', False))77 78 def clean_email(self):79 email = self.cleaned_data['email']80 if User.objects.filter(username=email).count() > 0:81 # raise forms.ValidationError('ÐолÑзоваÑÐµÐ»Ñ Ñ Ñаким адÑеÑом Ñже ÑÑÑеÑÑвÑеÑ')82 raise forms.ValidationError('ÐепÑавилÑно Ð²Ð²ÐµÐ´ÐµÐ½Ñ Ð´Ð°Ð½Ð½Ñе')83 return email84 def clean(self):85 cleaned_data = self.cleaned_data86 password = cleaned_data.get('password')87 password_confirm = cleaned_data.get('password_confirm')88 if password != password_confirm:89 # raise forms.ValidationError('ÐаÑÐ¾Ð»Ñ Ð¸ подÑвеÑждение не ÑовпадаÑÑ')90 raise forms.ValidationError('ÐепÑавилÑно Ð²Ð²ÐµÐ´ÐµÐ½Ñ Ð´Ð°Ð½Ð½Ñе')91 return cleaned_data92class UrCustomerForm(UserForm):93 type = forms.CharField(widget=forms.HiddenInput, initial='y')94 repr_surname = forms.RegexField(label="Ð¤Ð°Ð¼Ð¸Ð»Ð¸Ñ Ð·Ð°ÐºÐ°Ð·Ñика", max_length=80, regex=fio_regexp, error_messages=generic_error)95 repr_first_name = forms.RegexField(label="ÐÐ¼Ñ Ð·Ð°ÐºÐ°Ð·Ñика", max_length=30, regex=fio_regexp, error_messages=generic_error)96 repr_second_name = forms.RegexField(label="ÐÑÑеÑÑво заказÑика", max_length=50, required=False, regex=fio_regexp, error_messages=generic_error)97 pay_type = forms.ModelMultipleChoiceField(label="СпоÑоб оплаÑÑ", queryset=models.PayingType.objects.all(), widget=forms.CheckboxSelectMultiple)98 inn = forms.RegexField(label="ÐÐÐ",max_length=10,regex=num_regexp, error_messages=generic_error)99 class Meta:100 model = models.UrCustomer101 exclude = ['customer']102 fields = ['type', 'email', 'password', 'password_confirm'] + [f.name for f in models.UrCustomer._meta.fields[2:]] + [f.name for f in models.UrCustomer._meta.many_to_many]103class PhysCustomerForm(UserForm):104 type = forms.CharField(widget=forms.HiddenInput, initial='f')105 surname = forms.RegexField(label="Ð¤Ð°Ð¼Ð¸Ð»Ð¸Ñ Ð·Ð°ÐºÐ°Ð·Ñика", max_length=80, regex=fio_regexp, error_messages=generic_error)106 first_name = forms.RegexField(label="ÐÐ¼Ñ Ð·Ð°ÐºÐ°Ð·Ñика", max_length=30, regex=fio_regexp, error_messages=generic_error)107 second_name = forms.RegexField(label="ÐÑÑеÑÑво заказÑика", max_length=50, required=False, regex=fio_regexp, error_messages=generic_error)108 pay_type = forms.ModelMultipleChoiceField(label="СпоÑоб оплаÑÑ", queryset=models.PayingType.objects.all(), widget=forms.CheckboxSelectMultiple)109 passport_when = forms.DateField(label="ÐаÑа вÑдаÑи", widget=SelectDateWidget(years=range(2010, 1900, -1)))110 passport_serial = forms.RegexField(label = "СеÑÐ¸Ñ Ð¿Ð°ÑпоÑÑа", max_length=4, regex=num_regexp, error_messages=generic_error)111 passport_number = forms.RegexField(label = "ÐÐ¾Ð¼ÐµÑ Ð¿Ð°ÑпоÑÑа", max_length=6, regex=num_regexp, error_messages=generic_error)112 passport_who = forms.RegexField(label = "Ðем вÑдан", max_length=100, regex=kiril_regexp, error_messages=generic_error )113 class Meta:114 model = models.PhysCustomer115 exclude = ['customer']116 fields = ['type', 'email', 'password', 'password_confirm'] + [f.name for f in models.PhysCustomer._meta.fields[2:]] + [f.name for f in models.PhysCustomer._meta.many_to_many]117class TesterDetailForm(forms.ModelForm):118 password = forms.RegexField(label='ÐаÑолÑ', widget=forms.PasswordInput(render_value=False), regex=password_regexp, max_length=30, min_length=5, required=False, error_messages=generic_error)119 password_confirm = forms.RegexField(label='ÐодÑвеÑждение паÑолÑ', widget=forms.PasswordInput(render_value=False), regex=password_regexp, max_length=30, min_length=5, required=False, error_messages=generic_error)120 osystems = forms.ModelMultipleChoiceField(label="ÐпеÑаÑионнÑе ÑиÑÑемÑ", queryset=models.OSystem.objects.all(), widget=forms.CheckboxSelectMultiple)121 program_languages = forms.ModelMultipleChoiceField(label="ЯзÑки пÑогÑаммиÑованиÑ", queryset=models.ProgramLang.objects.all(), widget=forms.CheckboxSelectMultiple)122 testing_types = forms.ModelMultipleChoiceField(label="Ð¢Ð¸Ð¿Ñ ÑеÑÑиÑованиÑ", queryset=models.TestingType.objects.all(), widget=forms.CheckboxSelectMultiple)123 browsers = forms.ModelMultipleChoiceField(label="ÐÑаÑзеÑÑ", queryset=models.Browser.objects.all(), widget=forms.CheckboxSelectMultiple)124 description = forms.CharField(label="Ð Ñебе", widget=forms.Textarea, required=False, max_length=300)125 class Meta:126 model = models.Tester127 fields = ['password', 'password_confirm', 'osystems', 'program_languages', 'testing_types', 'browsers', 'description']128 def save(self, *args, **kwargs):129 """Ðбновление ÑеÑÑеÑа Ñ ÑÑеÑом ÑÐ¼ÐµÐ½Ñ Ð¿Ð°ÑолÑ"""130 assert(self.is_valid())131 # ÐÑзов оÑигиналÑного save132 data = self.cleaned_data133 tester = super(TesterDetailForm, self).save(*args, **kwargs)134 # ÐÑовеÑка на ÑÐ¼ÐµÐ½Ñ Ð¿Ð°ÑолÑ135 if data['password']:136 # ÐаÑÐ¾Ð»Ñ Ð¸Ð·Ð¼ÐµÐ½Ð¸Ð»ÑÑ137 user = tester.user138 user.set_password(data['password'])139 user.save()140 return tester141 def clean(self):142 # ÐÑзов оÑигиналÑного clean()143 data = super(TesterDetailForm, self).clean()144 password = data.get('password')145 password_confirm = data.get('password_confirm')146 if password != password_confirm:147 # raise forms.ValidationError('ÐаÑÐ¾Ð»Ñ Ð¸ подÑвеÑждение не ÑовпадаÑÑ')148 raise forms.ValidationError('ÐепÑавилÑно Ð²Ð²ÐµÐ´ÐµÐ½Ñ Ð´Ð°Ð½Ð½Ñе')149 return data150# XXX:151# ÐÑедпÑоÑмоÑÑ ÑоÑм152from django.contrib.formtools.preview import FormPreview153from django.http import HttpResponseRedirect154class StupidFormPreview(FormPreview):155 def done(self, request, cleaned_data):...
safe_send.py
Source:safe_send.py
1from pyrogram.errors import RPCError, FloodWait2from pyrogram import Client3import time4import logging5def send_message(client: Client, sleep: bool = True, *args, **kwargs):6 """Sends a message in a way that never triggers exceptions and logs errors7 :param client: The pyrogram.Client instance to call the method for8 :type client: class: Client9 :param sleep: If True, the default, the function will call time.sleep()10 in case of a FloodWait exception and return the exception object11 after the sleep is done, otherwise the ``FloodWait`` exception is returned12 immediately13 """14 try:15 return client.send_message(*args, **kwargs)16 except FloodWait as fw:17 logging.warning(f"FloodWait! A wait of {fw.x} seconds is required")18 if sleep:19 time.sleep(fw.x)20 return fw21 except RPCError as generic_error:22 logging.error(f"An exception occurred: {generic_error}")23 return generic_error24def send_photo(client: Client, sleep: bool = True, *args, **kwargs):25 """Sends a photo in a way that never triggers exceptions and logs errors26 :param client: The pyrogram.Client instance to call the method for27 :type client: class: Client28 :param sleep: If True, the default, the function will call time.sleep()29 in case of a FloodWait exception and return the exception object30 after the sleep is done, otherwise the ``FloodWait`` exception is returned31 immediately32 """33 try:34 return client.send_photo(*args, **kwargs)35 except FloodWait as fw:36 logging.warning(f"FloodWait! A wait of {fw.x} seconds is required")37 if sleep:38 time.sleep(fw.x)39 return fw40 except RPCError as generic_error:41 logging.error(f"An exception occurred: {generic_error}")42 return generic_error43def send_audio(client: Client, sleep: bool = True, *args, **kwargs):44 """Sends an audio in a way that never triggers exceptions and logs errors45 :param client: The pyrogram.Client instance to call the method for46 :type client: class: Client47 :param sleep: If True, the default, the function will call time.sleep()48 in case of a FloodWait exception and return the exception object49 after the sleep is done, otherwise the ``FloodWait`` exception is returned50 immediately51 """52 try:53 return client.send_audio(*args, **kwargs)54 except FloodWait as fw:55 logging.warning(f"FloodWait! A wait of {fw.x} seconds is required")56 if sleep:57 time.sleep(fw.x)58 return fw59 except RPCError as generic_error:60 logging.error(f"An exception occurred: {generic_error}")61 return generic_error62def send_sticker(client: Client, sleep: bool = True, *args, **kwargs):63 """Sends a sticker in a way that never triggers exceptions and logs errors64 :param client: The pyrogram.Client instance to call the method for65 :type client: class: Client66 :param sleep: If True, the default, the function will call time.sleep()67 in case of a FloodWait exception and return the exception object68 after the sleep is done, otherwise the ``FloodWait`` exception is returned69 immediately70 """71 try:72 return client.send_sticker(*args, **kwargs)73 except FloodWait as fw:74 logging.warning(f"FloodWait! A wait of {fw.x} seconds is required")75 if sleep:76 time.sleep(fw.x)77 return fw78 except RPCError as generic_error:79 logging.error(f"An exception occurred: {generic_error}")80 return generic_error81def send_animation(client: Client, sleep: bool = True, *args, **kwargs):82 """Sends an animation in a way that never triggers exceptions and logs errors83 :param client: The pyrogram.Client instance to call the method for84 :type client: class: Client85 :param sleep: If True, the default, the function will call time.sleep()86 in case of a FloodWait exception and return the exception object87 after the sleep is done, otherwise the ``FloodWait`` exception is returned88 immediately89 """90 try:91 return client.send_animation(*args, **kwargs)92 except FloodWait as fw:93 logging.warning(f"FloodWait! A wait of {fw.x} seconds is required")94 if sleep:95 time.sleep(fw.x)96 return fw97 except RPCError as generic_error:98 logging.error(f"An exception occurred: {generic_error}")...
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!!