KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use charge method in hypothesis
Best Python code snippet using hypothesis
charge_actions.py
Source:charge_actions.py 
1from core.admin import admin_main2from core.charge import charge_main,charge_types,charge_rule3from core.lib.general import *4from core.lib.time_lib import *5from core.lib.day_of_week import *6from core.ibs_exceptions import *7from core.errors import errorText8from core.db import db_main,ibs_db9from core.db.ibs_query import IBSQuery10from core.ras import ras_main11from core.user import user_main12from core.group import group_main13from core.bandwidth_limit import bw_main14from core.charge.voip_tariff import tariff_main15class ChargeActions:16 CHARGE_TYPES=["Internet","VoIP"]17 def addCharge(self,name,comment,charge_type,admin_id,visible_to_all):18 """19 add a charge to database, and command the charge loader to load it20 charge_type(str): Internet or VoIP21 """22 self.__addChargeCheckInput(name,comment,charge_type,admin_id,visible_to_all)23 charge_id=self.__getNewChargeID()24 self.__insertNewCharge(charge_id,name,comment,charge_type,admin_id,visible_to_all)25 charge_main.getLoader().loadCharge(charge_id)26 27 28 def __addChargeCheckInput(self,name,comment,charge_type,admin_id,visible_to_all):29 """30 check addCharge inputs validity31 raise exception on bad input32 """33 admin_main.getLoader().checkAdminID(admin_id)34 checkDBBool(visible_to_all,"Visible to all")35 self.__checkChargeType(charge_type)36 if charge_main.getLoader().chargeNameExists(name):37 raise GeneralException(errorText("CHARGES","CHARGE_NAME_EXISTS") % name)38 39 if not isValidName(name):40 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_NAME") % name)41 def __checkChargeType(self,charge_type):42 if charge_type not in self.CHARGE_TYPES:43 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_TYPE")%charge_type)44 def __getNewChargeID(self):45 """46 return a new id for new charge47 """48 return db_main.getHandle().seqNextVal("charges_id_seq")49 50 def __insertNewCharge(self,charge_id,name,comment,charge_type,admin_id,visible_to_all):51 """52 insert the new charge to db53 """54 db_main.getHandle().insert("charges",{"charge_id":charge_id,55 "name":dbText(name),56 "comment":dbText(comment),57 "charge_type":dbText(charge_type),58 "admin_id":admin_id,59 "visible_to_all":dbText(visible_to_all)60 })61###########################################################62 def updateCharge(self,charge_id,name,comment,visible_to_all):63 """64 change charge rule properties65 """66 self.__updateChargeCheckInput(charge_id,name,comment,visible_to_all)67 self.__updateCharge(charge_id,name,comment,visible_to_all)68 charge_main.getLoader().unloadCharge(charge_id)69 charge_main.getLoader().loadCharge(charge_id)70 def __updateChargeCheckInput(self,charge_id,name,comment,visible_to_all):71 """72 check inputs of changeChargeInfo73 raise an exception on bad input74 """75 charge_obj=charge_main.getLoader().getChargeByID(charge_id)76 checkDBBool(visible_to_all,"Visible To All")77 if charge_obj.getChargeName() != name and charge_main.getLoader().chargeNameExists(name):78 raise GeneralException(errorText("CHARGES","CHARGE_NAME_EXISTS") % name)79 if not isValidName(name):80 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_NAME") % name)81 def __updateCharge(self,charge_id,name,comment,visible_to_all):82 """83 update a charge rule information in DB84 """85 db_main.getHandle().update("charges",{"name":dbText(name),86 "comment":dbText(comment),87 "visible_to_all":dbText(visible_to_all)88 },"charge_id=%s"%charge_id)89###########################################################90 def addInternetChargeRule(self,charge_name,start_time,end_time,day_of_weeks,cpm,cpk,\91 assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports):92 """93 add a charge rule to charge with id "charge_id" and reload the charge94 it will add charge_rule and it's ports to db too95 96 start_time(str): time representation eg "0:0:0"97 end_time(str): time representation eg "0:0:0"98 day_of_weeks(list of str): list of full day names eg ["Monday", "Wednesday"]99 ports(list of str): list of port names or ["_ALL_"] for all ports100 """101 start_time,end_time,day_of_weeks=self.__chargeRuleTimesCheck(start_time,end_time,day_of_weeks)102 self.__internetChargeRuleCheckInput(charge_name,cpm,cpk,assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports)103 charge_obj=charge_main.getLoader().getChargeByName(charge_name)104 day_of_weeks_container=self.__createDayOfWeekContainer(day_of_weeks)105 tx_leaf_id,rx_leaf_id=self.__convertBwLeaves(tx_leaf_name,rx_leaf_name)106 rule_obj=self.__createInternetChargeRuleObject(charge_obj,start_time,end_time,day_of_weeks_container,\107 cpm,cpk,assumed_kps,bandwidth_limit_kbytes,tx_leaf_id,rx_leaf_id,ras_id,ports)108 self.__checkRuleConflict(charge_obj,rule_obj)109 self.__addInternetChargeRuleToDB(rule_obj)110 charge_main.getLoader().loadCharge(charge_obj.getChargeID())111 def __internetChargeRuleCheckInput(self,charge_name,cpm,cpk,assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports):112 self.__chargeRuleCheckInput(charge_name,"Internet",ras_id,ports)113 try:114 assumed_kps=int(assumed_kps)115 except:116 raise GeneralException(errorText("CHARGES","ASSUMED_KPS_NOT_INTEGER"))117 try:118 bandwidth_limit_kbytes=int(bandwidth_limit_kbytes)119 except:120 raise GeneralException(errorText("CHARGES","BANDWIDTH_LIMIT_NOT_INTEGER"))121 try:122 cpm=float(cpm)123 except:124 raise GeneralException(errorText("CHARGES","CPM_NOT_NUMERIC"))125 try:126 cpk=float(cpk)127 except:128 raise GeneralException(errorText("CHARGES","CPK_NOT_NUMERIC"))129 130 if assumed_kps<=0:131 raise GeneralException(errorText("CHARGES","ASSUMED_KPS_NOT_POSITIVE"))132 if bandwidth_limit_kbytes<0:133 raise GeneralException(errorText("CHARGES","BANDWIDTH_LIMIT_NOT_POSITIVE"))134 135 if cpm<0:136 raise GeneralException(errorText("CHARGES","CPM_NOT_POSITIVE"))137 138 if cpk<0:139 raise GeneralException(errorText("CHARGES","CPK_NOT_POSITIVE"))140 if len(tx_leaf_name)!=0 and len(rx_leaf_name)!=0:141 bw_main.getLoader().getLeafByName(tx_leaf_name)142 bw_main.getLoader().getLeafByName(rx_leaf_name)143 elif (len(tx_leaf_name)==0 and len(rx_leaf_name)!=0) or (len(tx_leaf_name)!=0 and len(rx_leaf_name)==0):144 raise GeneralException(errorText("CHARGES","BW_LEAF_NAMES_SHOULD_BOTH_SET"))145 146 def __addInternetChargeRuleToDB(self,rule_obj):147 """148 add rule_obj properties to DB149 """150 charge_rule_id=self.__getNewChargeRuleID()151 rule_obj.setRuleID(charge_rule_id)152 query=self.__addInternetChargeRuleAndPortsQuery(rule_obj)153 db_main.getHandle().transactionQuery(query)154 def __getNewChargeRuleID(self):155 """156 return an id for new charge_rule157 """158 return db_main.getHandle().seqNextVal("charge_rules_id_seq")159 def __addInternetChargeRuleAndPortsQuery(self,rule_obj):160 """161 return query for inserting "rule_obj" with id "charge_rule_id"162 """163 return self.__addInternetChargeRuleQuery(rule_obj) + \164 self.__addChargeRulePortsQuery(rule_obj.getPorts(),rule_obj.getRuleID()) + \165 self.__addChargeRuleDowsQuery(rule_obj.getDows(),rule_obj.getRuleID())166 167 def __addInternetChargeRuleQuery(self,rule_obj):168 """169 return query for inserting rule_obj properties into charge_rules table170 """171 ras_id=self.__convertRasID(rule_obj.getRasID())172 173 return ibs_db.createInsertQuery("internet_charge_rules",{"charge_id":rule_obj.charge_obj.getChargeID(),174 "charge_rule_id":rule_obj.getRuleID(),175 "start_time":dbText(rule_obj.start_time),176 "end_time":dbText(rule_obj.end_time),177 "cpm":float(rule_obj.cpm),178 "cpk":float(rule_obj.cpk),179 "assumed_kps":integer(rule_obj.assumed_kps),180 "bandwidth_limit_kbytes":integer(rule_obj.bandwidth_limit),181 "bw_transmit_leaf_id":dbNull(rule_obj.bw_tx_leaf_id),182 "bw_receive_leaf_id":dbNull(rule_obj.bw_rx_leaf_id),183 "ras_id":ras_id184 })185 def __addChargeRulePortsQuery(self,ports_list,charge_rule_id):186 """187 return query for inserting "ports" to charge_rule "charge_rule_id"188 """189 if None in ports_list:190 ports_list=[dbNull(None)]191 query=""192 for port in ports_list:193 query+=ibs_db.createInsertQuery("charge_rule_ports",{"charge_rule_id":charge_rule_id,194 "ras_port":dbText(port)195 })196 return query197 def __addChargeRuleDowsQuery(self,dows_container,charge_rule_id):198 """199 return query for inserting "dows_container" to charge_rule "charge_rule_id"200 """201 query=""202 for dow in dows_container:203 query+=ibs_db.createInsertQuery("charge_rule_day_of_weeks",{"charge_rule_id":charge_rule_id,204 "day_of_week":dow.getIntValue()205 })206 return query207################################################################3208 def __checkChargeRuleInCharge(self,charge_rule_id,charge_name):209 """210 check if charge rule with id "charge_rule_id" is in charge with name "charge_name"211 raise a GeneralException if it isn't212 """213 if charge_rule_id not in charge_main.getLoader().getChargeByName(charge_name).getRules():214 raise GeneralException(errorText("CHARGES","CHARGE_RULE_NOT_IN_CHARGE")%(charge_rule_id,charge_name))215 216 def __createDayOfWeekContainer(self,day_of_weeks):217 """218 day_of_weeks: list of day of weeks strings219 """220 dows_container=DayOfWeekIntContainer()221 for dow in day_of_weeks:222 dows_container.append(dow.getDowInt())223 return dows_container224 def __createInternetChargeRuleObject(self,charge_obj,start_time,end_time,day_of_week_container,cpm,cpk,\225 assumed_kps,bandwidth_limit_kbytes,tx_leaf_id,rx_leaf_id,ras_id,ports,charge_rule_id=None):226 """227 create an half complete rule object from arguments and return it228 this object is useful for checking conflict229 """230 rule_info={}231 rule_info["charge_rule_id"]=charge_rule_id232 rule_info["cpm"]=cpm233 rule_info["cpk"]=cpk234 rule_info["start_time"]=start_time.getFormattedTime()235 rule_info["end_time"]=end_time.getFormattedTime()236 rule_info["bandwidth_limit_kbytes"]=bandwidth_limit_kbytes237 rule_info["assumed_kps"]=assumed_kps238 rule_info["ras_id"]=ras_id239 rule_info["bw_transmit_leaf_id"]=tx_leaf_id240 rule_info["bw_receive_leaf_id"]=rx_leaf_id241 return charge_types.getChargeRuleObjForType("Internet",rule_info,charge_obj,day_of_week_container,ports)242 def __checkRuleConflict(self,charge_obj,rule_obj,ignore_rule_ids=[]):243 """244 check if rule_obj conflicts with any other rule in charge charge_obj245 raise a generalException if there's a conflict246 """247 charge_obj.checkConflict(rule_obj,ignore_rule_ids)248 def __chargeRuleTimesCheck(self,start_time,end_time,day_of_weeks):249 """250 Check rule times and convert them the way ibs needed them to insert to IBS251 day_of_weeks(list): list of day of week strings252 """253 try:254 start_time=Time(start_time)255 except GeneralException,e:256 raise GeneralException(errorText("CHARGES","INVALID_RULE_START_TIME")%e)257 try:258 end_time=Time(end_time)259 except GeneralException,e:260 raise GeneralException(errorText("CHARGES","INVALID_RULE_END_TIME")%e)261 if start_time >= end_time:262 raise GeneralException(errorText("CHARGES","RULE_END_LESS_THAN_START"))263 264 try:265 dows=map(DayOfWeekString,day_of_weeks)266 except GeneralException,e:267 raise GeneralException(errorText("CHARGES","INVALID_DAY_OF_WEEK")%e)268 return (start_time,end_time,dows)269 270 def __chargeRuleCheckInput(self,charge_name,charge_type,ras_id,ports):271 """272 check ChargeRule inputs and raise an exception on bad input273 """274 charge_obj=charge_main.getLoader().getChargeByName(charge_name)275 if charge_obj.getType()!=charge_type:276 raise GeneralException(errorText("CHARGES","ANOTHER_CHARGE_TYPE_REQUIRED")%charge_type)277 if ras_id!=charge_rule.ChargeRule.ALL:278 ras_main.getLoader().checkRasID(ras_id)279 if len(ports)==0:280 raise GeneralException(errorText("CHARGES","NO_PORT_SELECTED"))281 282 def __convertBwLeaves(self,tx_leaf_name,rx_leaf_name):283 """284 convert bw leaf names to their id, or to None if they are not specified285 """286 if tx_leaf_name=='':287 return (None,None)288 else:289 return (bw_main.getLoader().getLeafByName(tx_leaf_name).getLeafID(),290 bw_main.getLoader().getLeafByName(rx_leaf_name).getLeafID())291 def __convertRasID(self,ras_id):292 """293 convert ras_id to be ready for db insertion, ras_id wildcard is shown by null294 """ 295 if ras_id == charge_rule.ChargeRule.ALL:296 ras_id="NULL"297 return ras_id298#######################################################3299 def updateInternetChargeRule(self,charge_name,charge_rule_id,start_time,end_time,day_of_weeks,cpm,cpk,\300 assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports):301 """302 add a charge rule to charge with id "charge_id" and reload the charge303 it will add charge_rule and it's ports to db too304 """305 start_time,end_time,day_of_weeks=self.__chargeRuleTimesCheck(start_time,end_time,day_of_weeks)306 charge_rule_id=self.__updateInternetChargeRuleCheckInput(charge_name,charge_rule_id,start_time,end_time,day_of_weeks,cpm,cpk,\307 assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports)308 309 charge_obj=charge_main.getLoader().getChargeByName(charge_name)310 day_of_weeks_container=self.__createDayOfWeekContainer(day_of_weeks)311 tx_leaf_id,rx_leaf_id=self.__convertBwLeaves(tx_leaf_name,rx_leaf_name)312 rule_obj=self.__createInternetChargeRuleObject(charge_obj,start_time,end_time,day_of_weeks_container,\313 cpm,cpk,assumed_kps,bandwidth_limit_kbytes,tx_leaf_id,rx_leaf_id,ras_id,\314 ports,charge_rule_id)315 self.__checkRuleConflict(charge_obj,rule_obj,[charge_rule_id])316 self.__updateInternetChargeRuleDB(rule_obj)317 charge_main.getLoader().loadCharge(charge_obj.getChargeID())318 def __updateInternetChargeRuleCheckInput(self,charge_name,charge_rule_id,start_time,end_time,day_of_weeks,cpm,cpk,\319 assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports):320 self.__internetChargeRuleCheckInput(charge_name,cpm,cpk,assumed_kps,bandwidth_limit_kbytes,tx_leaf_name,rx_leaf_name,ras_id,ports)321 return self.__updateChargeRuleCheckInput(charge_rule_id,charge_name)322 def __updateChargeRuleCheckInput(self,charge_rule_id,charge_name):323 try:324 charge_rule_id=int(charge_rule_id)325 except:326 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_RULE_ID")%charge_rule_id)327 self.__checkChargeRuleInCharge(charge_rule_id,charge_name)328 329 return charge_rule_id330 331 def __updateInternetChargeRuleDB(self,rule_obj):332 """333 add rule_obj properties to DB334 """335 query=self.__updateInternetChargeRuleAndPortsQuery(rule_obj)336 db_main.getHandle().transactionQuery(query)337 def __updateInternetChargeRuleAndPortsQuery(self,rule_obj):338 """339 return query for inserting "rule_obj" with id "charge_rule_id"340 """341 return self.__updateInternetChargeRuleQuery(rule_obj) + \342 self.__updateChargeRulePortsAndDowsQuery(rule_obj)343 344 def __updateChargeRulePortsAndDowsQuery(self,rule_obj):345 return self.__delChargeRulePortsQuery(rule_obj.getRuleID()) + \346 self.__delChargeRuleDowsQuery(rule_obj.getRuleID()) + \347 self.__addChargeRulePortsQuery(rule_obj.getPorts(),rule_obj.getRuleID()) + \348 self.__addChargeRuleDowsQuery(rule_obj.getDows(),rule_obj.getRuleID())349 350 def __updateInternetChargeRuleQuery(self,rule_obj):351 """352 return query for inserting rule_obj properties into charge_rules table353 """354 ras_id=self.__convertRasID(rule_obj.getRasID())355 return ibs_db.createUpdateQuery("internet_charge_rules",{"start_time":dbText(rule_obj.start_time),356 "end_time":dbText(rule_obj.end_time),357 "cpm":float(rule_obj.cpm),358 "cpk":float(rule_obj.cpk),359 "assumed_kps":integer(rule_obj.assumed_kps),360 "bandwidth_limit_kbytes":integer(rule_obj.bandwidth_limit),361 "ras_id":ras_id,362 "bw_transmit_leaf_id":dbNull(rule_obj.bw_tx_leaf_id),363 "bw_receive_leaf_id":dbNull(rule_obj.bw_rx_leaf_id),364 },"charge_rule_id=%s"%rule_obj.getRuleID())365 def __delChargeRulePortsQuery(self,rule_id):366 return ibs_db.createDeleteQuery("charge_rule_ports","charge_rule_id=%s"%rule_id)367 368 369 def __delChargeRuleDowsQuery(self,rule_id):370 return ibs_db.createDeleteQuery("charge_rule_day_of_weeks","charge_rule_id=%s"%rule_id)371 372######################373 def delChargeRule(self,charge_rule_id,charge_name):374 """375 delete charge_rule with id "charge_rule_id" from charge with name "charge_name"376 it will delete all of charge_rule ports too377 """378 self.__delChargeRuleCheckInput(charge_rule_id,charge_name)379 charge_obj=charge_main.getLoader().getChargeByName(charge_name)380 self.__delChargeRuleFromDB(charge_rule_id,charge_obj)381 charge_main.getLoader().loadCharge(charge_obj.getChargeID())382 def __delChargeRuleCheckInput(self,charge_rule_id,charge_name):383 """384 check delChargeRule Inputs385 raise a generalExcetpion on bad input386 """387 self.__checkChargeRuleInCharge(charge_rule_id,charge_name)388 def __delChargeRuleFromDB(self,charge_rule_id,charge_obj):389 """390 delete charge_rule with id "charge_rule_id" from db391 """392 query=self.__delChargeRuleTotallyQuery(charge_rule_id,charge_obj)393 db_main.getHandle().transactionQuery(query)394 def __delChargeRuleTotallyQuery(self,charge_rule_id,charge_obj):395 """396 return query for compeletly removing charge rule from database397 """398 query=self.__delChargeRulePortsQuery(charge_rule_id)+ \399 self.__delChargeRuleDowsQuery(charge_rule_id) 400 if charge_obj.getType()=="Internet":401 query+=self.__delInternetChargeRuleQuery(charge_rule_id)402 elif charge_obj.getType()=="VoIP":403 query+=self.__delVoIPChargeRuleQuery(charge_rule_id)404 return query405 def __delInternetChargeRuleQuery(self,charge_rule_id):406 """407 return query needed to delete rule only from charge_rules table408 it won't delete ports409 """410 return ibs_db.createDeleteQuery("internet_charge_rules","charge_rule_id=%s"%charge_rule_id)411 def __delVoIPChargeRuleQuery(self,charge_rule_id):412 """413 return query needed to delete rule only from charge_rules table414 it won't delete ports415 """416 return ibs_db.createDeleteQuery("voip_charge_rules","charge_rule_id=%s"%charge_rule_id)417###############################418 def delCharge(self,charge_name):419 """420 delete a charge from both db and list of active charges421 """422 self.__delChargeCheckInput(charge_name)423 charge_obj=charge_main.getLoader().getChargeByName(charge_name)424 self.__checkChargeUsage(charge_obj)425 self.__delChargeFromDB(charge_obj)426 charge_main.getLoader().unloadCharge(charge_obj.getChargeID())427 428 def __checkChargeUsage(self,charge_obj):429 """430 check if charge used in any user/group, if so we can't delete it and we raise an exception431 """432 if charge_obj.isInternetCharge():433 attr_name="normal_charge"434 else:435 attr_name="voip_charge"436 self.__checkChargeUsageInUsers(charge_obj,attr_name)437 self.__checkChargeUsageInGroups(charge_obj,attr_name)438 439 def __checkChargeUsageInUsers(self,charge_obj,attr_name):440 user_ids=user_main.getActionManager().getUserIDsWithAttr(attr_name,charge_obj.getChargeID())441 if len(user_ids)>0:442 raise GeneralException(errorText("CHARGES","CHARGE_USED_IN_USER")%(charge_obj.getChargeName(),443 ",".join(map(str,user_ids))))444 def __checkChargeUsageInGroups(self,charge_obj,attr_name):445 group_ids=group_main.getActionManager().getGroupIDsWithAttr(attr_name,charge_obj.getChargeID())446 if len(group_ids)>0:447 raise GeneralException(errorText("CHARGES","CHARGE_USED_IN_GROUP")%(charge_obj.getChargeName(),448 ",".join(map(lambda _id: group_main.getLoader().getGroupByID(_id).getGroupName(),group_ids))))449 450 def __delChargeCheckInput(self,charge_name):451 """452 check del Charge inputs453 raise exception on bad input454 """455 charge_main.getLoader().checkChargeName(charge_name)456 def __delChargeFromDB(self,charge_obj):457 """458 --completely-- delete charge with id "charge_id" from db, also delete459 it's rules and ports460 """461 ibs_query=IBSQuery()462 for charge_rule_id in charge_obj.getRules():463 ibs_query+=self.__delChargeRuleTotallyQuery(charge_rule_id,charge_obj)464 465 ibs_query+=self.__delChargeQuery(charge_obj.getChargeID())466 ibs_query.runQuery()467 468 469 def __delChargeQuery(self,charge_id):470 """471 return query to delete the charge itself from charges table472 """473 return ibs_db.createDeleteQuery("charges","charge_id=%s"%charge_id)474 ##########################################475 def addVoIPChargeRule(self,charge_name,start_time,end_time,day_of_weeks,tariff_name,ras_id,ports):476 start_time,end_time,day_of_weeks=self.__chargeRuleTimesCheck(start_time,end_time,day_of_weeks)477 self.__voipChargeRuleCheckInput(charge_name,tariff_name,ras_id,ports)478 charge_obj=charge_main.getLoader().getChargeByName(charge_name)479 day_of_weeks_container=self.__createDayOfWeekContainer(day_of_weeks)480 tariff_id=tariff_main.getLoader().getTariffByName(tariff_name).getTariffID()481 rule_obj=self.__createVoIPChargeRuleObject(charge_obj,start_time,end_time,day_of_weeks_container,\482 tariff_id,ras_id,ports)483 self.__checkRuleConflict(charge_obj,rule_obj)484 self.__addVoIPChargeRuleToDB(rule_obj)485 charge_main.getLoader().loadCharge(charge_obj.getChargeID())486 def __voipChargeRuleCheckInput(self,charge_name,tariff_name,ras_id,ports):487 self.__chargeRuleCheckInput(charge_name,"VoIP",ras_id,ports)488 tariff_main.getLoader().getTariffByName(tariff_name)489 490 def __createVoIPChargeRuleObject(self,charge_obj,start_time,end_time,day_of_week_container,tariff_id,ras_id,ports,charge_rule_id=None):491 """492 create an half complete rule object from arguments and return it493 this object is useful for checking conflict494 """495 rule_info={}496 rule_info["charge_rule_id"]=charge_rule_id497 rule_info["start_time"]=start_time.getFormattedTime()498 rule_info["end_time"]=end_time.getFormattedTime()499 rule_info["ras_id"]=ras_id500 rule_info["tariff_id"]=tariff_id501 return charge_types.getChargeRuleObjForType("VoIP",rule_info,charge_obj,day_of_week_container,ports)502 503 def __addVoIPChargeRuleToDB(self,rule_obj):504 """505 add rule_obj properties to DB506 """507 charge_rule_id=self.__getNewChargeRuleID()508 rule_obj.setRuleID(charge_rule_id)509 query=self.__addVoIPChargeRuleAndPortsQuery(rule_obj)510 db_main.getHandle().transactionQuery(query)511 def __addVoIPChargeRuleAndPortsQuery(self,rule_obj):512 """513 return query for inserting "rule_obj" with id "charge_rule_id"514 """515 return self.__addVoIPChargeRuleQuery(rule_obj) + \516 self.__addChargeRulePortsQuery(rule_obj.getPorts(),rule_obj.getRuleID()) + \517 self.__addChargeRuleDowsQuery(rule_obj.getDows(),rule_obj.getRuleID())518 519 def __addVoIPChargeRuleQuery(self,rule_obj):520 """521 return query for inserting rule_obj properties into charge_rules table522 """523 ras_id=self.__convertRasID(rule_obj.getRasID())524 525 return ibs_db.createInsertQuery("voip_charge_rules",{"charge_id":rule_obj.charge_obj.getChargeID(),526 "charge_rule_id":rule_obj.getRuleID(),527 "start_time":dbText(rule_obj.start_time),528 "end_time":dbText(rule_obj.end_time),529 "tariff_id":rule_obj.tariff_id,530 "ras_id":ras_id531 })532 ##########################################533 def updateVoIPChargeRule(self,charge_name,charge_rule_id,start_time,end_time,day_of_weeks,tariff_name,ras_id,ports):534 """535 add a charge rule to charge with id "charge_id" and reload the charge536 it will add charge_rule and it's ports to db too537 """538 start_time,end_time,day_of_weeks=self.__chargeRuleTimesCheck(start_time,end_time,day_of_weeks)539 charge_rule_id=self.__updateVoIPChargeRuleCheckInput(charge_name,charge_rule_id,start_time,end_time,day_of_weeks,tariff_name,ras_id,ports)540 charge_obj=charge_main.getLoader().getChargeByName(charge_name)541 day_of_weeks_container=self.__createDayOfWeekContainer(day_of_weeks)542 tariff_id=tariff_main.getLoader().getTariffByName(tariff_name).getTariffID()543 rule_obj=self.__createVoIPChargeRuleObject(charge_obj,start_time,end_time,day_of_weeks_container,\544 tariff_id,ras_id,ports,charge_rule_id)545 self.__checkRuleConflict(charge_obj,rule_obj,[charge_rule_id])546 self.__updateVoIPChargeRuleDB(rule_obj)547 charge_main.getLoader().loadCharge(charge_obj.getChargeID())548 def __updateVoIPChargeRuleCheckInput(self,charge_name,charge_rule_id,start_time,end_time,day_of_weeks,tariff_name,ras_id,ports):549 self.__voipChargeRuleCheckInput(charge_name,tariff_name,ras_id,ports)550 return self.__updateChargeRuleCheckInput(charge_rule_id,charge_name)551 def __updateVoIPChargeRuleDB(self,rule_obj):552 """553 add rule_obj properties to DB554 """555 query=self.__updateVoIPChargeRuleAndPortsQuery(rule_obj)556 db_main.getHandle().transactionQuery(query)557 def __updateVoIPChargeRuleAndPortsQuery(self,rule_obj):558 """559 return query for inserting "rule_obj" with id "charge_rule_id"560 """561 return self.__updateVoIPChargeRuleQuery(rule_obj) + \562 self.__updateChargeRulePortsAndDowsQuery(rule_obj)563 564 def __updateVoIPChargeRuleQuery(self,rule_obj):565 """566 return query for inserting rule_obj properties into charge_rules table567 """568 ras_id=self.__convertRasID(rule_obj.getRasID())569 return ibs_db.createUpdateQuery("voip_charge_rules",{"start_time":dbText(rule_obj.start_time),570 "end_time":dbText(rule_obj.end_time),571 "ras_id":ras_id,572 "tariff_id":rule_obj.tariff_id,573 },"charge_rule_id=%s"%rule_obj.getRuleID())574 ############################################# 575 def getChargesWithBwLeaf(self, leaf_id):576 """577 return a list of charge_names that leaf_id is used in charge rules578 """579 charge_names = []580 def checkLeafInChargeObj(charge_obj):581 if charge_obj.isInternetCharge():582 for rule_obj in charge_obj.getRules().itervalues():583 if rule_obj.bw_tx_leaf_id==leaf_id or rule_obj.bw_rx_leaf_id==leaf_id:584 charge_names.append(charge_obj.getChargeName())585 break586 587 charge_main.getLoader().runOnAllCharges(checkLeafInChargeObj) 588 return charge_names...
charge_loader.py
Source:charge_loader.py
1from core.charge.charge_rule import *2from core.charge.charge_types import *3from core.db import ibs_db,db_main4from core.ras import ras5from core.ibs_exceptions import *6from core.errors import errorText7from core.lib.time_lib import *8from core.lib.day_of_week import *9import threading10class ChargeLoader:11 def __init__(self):12 self.charges_id={} #{charge_id=>charge_obj}13 self.charges_name={} #{charge_name=>charge_obj}14 self.rule_loader=ChargeRuleLoader(self)15 16 def __getitem__(self,key):17 return self.getChargeByID(key)18 19 def __iter__(self):20 return self.charges_id.iterkeys()21 22 def getChargeByID(self,charge_id):23 try:24 return self.charges_id[charge_id]25 except KeyError:26 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_ID")%charge_id)27 def getChargeByName(self,charge_name):28 try:29 return self.charges_name[charge_name]30 except KeyError:31 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_NAME")%charge_name)32 33 def loadCharge(self,charge_id):34 """35 create a new charge object and all corresponding rules and put it in self.charges36 """37 charge_obj=self.__createCharge(charge_id)38 self.charges_id[charge_id]=charge_obj39 self.charges_name[charge_obj.getChargeName()]=charge_obj40 def loadAllCharges(self):41 """42 load all charges from db and put em in self.charges43 """44 charge_ids=self.__getAllChargeIDs()45 for charge_id in charge_ids:46 self.loadCharge(charge_id)47 48 def unloadCharge(self,charge_id):49 """50 delete charge with id "charge_id" from object charges dic51 """52 charge_obj=self.getChargeByID(charge_id)53 del(self.charges_id[charge_obj.getChargeID()])54 del(self.charges_name[charge_obj.getChargeName()])55 def checkChargeID(self,charge_id):56 """57 check if charge with id "charge_id" exists58 raise a general exception if not59 """60 if charge_id not in self.charges_id:61 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_ID") % charge_id)62 def checkChargeName(self,charge_name):63 """64 check if charge with name "charge_name" exists65 raise a general exception if not66 """67 if not self.chargeNameExists(charge_name):68 raise GeneralException(errorText("CHARGES","INVALID_CHARGE_NAME")%charge_name)69 70 def chargeNameExists(self,charge_name):71 """72 check if charge with name "charge_name" exists73 return 1 if it exists, and there's a charge with name "charge_name"74 return 0 if not75 """76 return self.charges_name.has_key(charge_name)77 def runOnAllCharges(self,function):78 """79 function(function instance): function to be called with charge_obj as argument80 function should accept one argument that is a charge_obj. Function will be called for81 all loaded charge_objs82 """83 map(function,self.charges_id.values())84 def getAllChargeNames(self):85 return self.charges_name.keys()86 def __createCharge(self,charge_id):87 """88 create and return a new charge object from charge_id information . it'll load all corresponding89 rules into the object90 """91 charge_info=self.__getChargeInfo(charge_id)92 charge_obj=self.__createChargeObject(charge_info) 93 rules=self.rule_loader.loadChargeRules(charge_obj)94 charge_obj.setRules(rules)95 return charge_obj96 97 def __createChargeObject(self,charge_info):98 """99 create and return a charge object from charge_info dic100 """101 klass=getChargeClassForType(charge_info["charge_type"])102 return klass(charge_info["charge_id"],charge_info["name"],charge_info["comment"],103 charge_info["admin_id"],charge_info["visible_to_all"],charge_info["charge_type"])104 105 def __getChargeInfo(self,charge_id):106 """107 return a dic of charge properties or raise an exception on error108 """109 try:110 charge_info=db_main.getHandle().get("charges","charge_id=%s"%charge_id)[0]111 except:112 logException("ChargeLoader.getChargeInfo: error in getting charge information")113 raise114 115 return charge_info116 117 def __getAllChargeIDs(self):118 """119 return a list of charge id's from database120 """121 charge_ids=db_main.getHandle().get("charges","true",0,-1,"",["charge_id"])122 return [m["charge_id"] for m in charge_ids]123class ChargeRuleLoader:124 def __init__(self,charge_loader):125 self.charge_loader=charge_loader126 def loadChargeRules(self,charge_obj):127 """128 return a dic of rules of charge_id in format {charge_rule_id=>charge_rule_obj}129 """130 rules_dic={}131 rules=self.__getChargeRuleIDs(charge_obj)132 for rule_info in rules:133 ports=self.__getRulePorts(rule_info["charge_rule_id"])134 day_of_weeks=self.__getDayOfWeeks(rule_info["charge_rule_id"])135 day_of_week_container=apply(DayOfWeekIntContainer,day_of_weeks)136 rule_obj=self.__createChargeRuleObject(charge_obj,rule_info,day_of_week_container,ports)137 rules_dic[rule_obj.getRuleID()]=rule_obj138 return rules_dic139 def __getChargeRuleIDs(self,charge_obj):140 """141 return a list of rules and their properties belongs to charge_id142 """143 rules_table=getRulesTable(charge_obj.getType())144 return db_main.getHandle().get(rules_table,"charge_id=%s"%charge_obj.getChargeID())145 def __createChargeRuleObject(self,charge_obj,rule_info,day_of_weeks,ports):146 """147 create a charge rule object from rule_info dic and ports list148 """149 return getChargeRuleObjForType(charge_obj.getType(),rule_info,charge_obj,day_of_weeks,ports)150 def __getRulePorts(self,charge_rule_id):151 """152 return a list of ports belongs to charge_rule_id153 """154 ports=db_main.getHandle().get("charge_rule_ports","charge_rule_id=%s"%charge_rule_id)155 return [m["ras_port"] for m in ports]156 157 def __getDayOfWeeks(self,charge_rule_id):158 """159 return a list of DayOfWeekInt instances 160 """161 return map(DayOfWeekInt,self.__getDayOfWeeksDB(charge_rule_id))162 def __getDayOfWeeksDB(self,charge_rule_id):163 """164 return a list of day of week integers belongs to charge_rule_id165 """166 dows=db_main.getHandle().get("charge_rule_day_of_weeks","charge_rule_id=%s"%charge_rule_id)167 return [m["day_of_week"] for m in dows]...
IsoelectricPoint.py
Source:IsoelectricPoint.py
1# Copyright Yair Benita Y.Benita@pharm.uu.nl2# Biopython (http://biopython.org) license applies3"""Calculate isoelectric points of polypeptides using methods of Bjellqvist.4pK values and the methos are taken from:5* Bjellqvist, B.,Hughes, G.J., Pasquali, Ch., Paquet, N., Ravier, F., Sanchez,6J.-Ch., Frutiger, S. & Hochstrasser, D.F.7The focusing positions of polypeptides in immobilized pH gradients can be predicted8from their amino acid sequences. Electrophoresis 1993, 14, 1023-1031. 9* Bjellqvist, B., Basse, B., Olsen, E. and Celis, J.E.10Reference points for comparisons of two-dimensional maps of proteins from11different human cell types defined in a pH scale where isoelectric points correlate12with polypeptide compositions. Electrophoresis 1994, 15, 529-539.13I designed the algorithm according to a note by David L. Tabb, available at:14http://fields.scripps.edu/DTASelect/20010710-pI-Algorithm.pdf15"""16positive_pKs = { 'Nterm': 7.5, 'K': 10.0, 'R': 12.0, 'H':5.98 }17negative_pKs = { 'Cterm': 3.55, 'D': 4.05, 'E': 4.45, 'C':9.0, 'Y':10.0 }18pKcterminal= {'D':4.55, 'E':4.75}19pKnterminal = {'A':7.59, 'M':7.0, 'S':6.93, 'P':8.36, 'T':6.82, 'V':7.44, 'E':7.7}20charged_aas = ('K', 'R', 'H', 'D', 'E', 'C', 'Y')21# access this module through ProtParam.ProteinAnalysis class.22# first make a ProteinAnalysis object and then call its isoelectric_point method.23class IsoelectricPoint:24 def __init__(self, ProteinSequence, AminoAcidsContent):25 self.sequence = ProteinSequence 26 self.charged_aas_content = self._select_charged(AminoAcidsContent)27 # This function creates a dictionary with the contents of each charged aa, 28 # plus Cterm and Nterm.29 def _select_charged(self, AminoAcidsContent):30 charged = {} 31 for aa in charged_aas:32 charged[aa] = float(AminoAcidsContent[aa])33 charged['Nterm'] = 1.034 charged['Cterm'] = 1.035 return charged36 #This function calculates the total charge of the protein at a given pH.37 def _chargeR(self, pH, pos_pKs, neg_pKs):38 PositiveCharge = 0.039 for aa, pK in pos_pKs.iteritems(): 40 CR = 10**(pK-pH)41 partial_charge = CR/(CR+1.0)42 PositiveCharge += self.charged_aas_content[aa] * partial_charge 43 NegativeCharge = 0.044 for aa, pK in neg_pKs.iteritems(): 45 CR = 10**(pH-pK)46 partial_charge = CR/(CR+1.0)47 NegativeCharge += self.charged_aas_content[aa] * partial_charge 48 return PositiveCharge - NegativeCharge 49 # This is the action function, it tries different pH until the charge of the protein is 0 (or close).50 def pi(self): 51 pos_pKs = dict(positive_pKs)52 neg_pKs = dict(negative_pKs)53 nterm = self.sequence[0]54 cterm = self.sequence[-1] 55 if nterm in pKnterminal.keys():56 pos_pKs['Nterm'] = pKnterminal[nterm]57 if cterm in pKcterminal.keys():58 neg_pKs['Cterm'] = pKcterminal[cterm]59 # Bracket between pH1 and pH260 pH = 7.061 Charge = self._chargeR(pH, pos_pKs, neg_pKs)62 if Charge > 0.0:63 pH1 = pH64 Charge1 = Charge65 while Charge1 > 0.0:66 pH = pH1 + 1.067 Charge = self._chargeR(pH, pos_pKs, neg_pKs)68 if Charge > 0.0:69 pH1 = pH70 Charge1 = Charge71 else:72 pH2 = pH73 Charge2 = Charge74 break75 else:76 pH2 = pH77 Charge2 = Charge78 while Charge2 < 0.0:79 pH = pH2 - 1.080 Charge = self._chargeR(pH, pos_pKs, neg_pKs)81 if Charge < 0.0:82 pH2 = pH83 Charge2 = Charge84 else:85 pH1 = pH86 Charge1 = Charge87 break88 # Bisection89 while pH2 - pH1 > 0.0001 and Charge!=0.0:90 pH = (pH1 + pH2) / 2.091 Charge = self._chargeR(pH, pos_pKs, neg_pKs)92 if Charge > 0.0:93 pH1 = pH94 Charge1 = Charge95 else:96 pH2 = pH97 Charge2 = Charge...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run hypothesis automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
