Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use set_event method in localstack

Best Python code snippet using localstack_python

sample_model.py

Source:sample_model.py

...119            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev3b_name: self.parameter_model.timer_interval(120                sm_name, ev_name, current_time, parameter121            ),122        )123        trg1a.set_event(ev1.event(ev1a_name))124        trg1b.set_event(ev1.event(ev1b_name))125        trg1c.set_event(ev1.event(ev1c_name))126        trg2a.set_event(ev2.event(ev2a_name))127        trg2b.set_event(ev2.event(ev2b_name))128        trg3a.set_event(ev3a)129        trg3b.set_event(ev3b)130        ev_reset_name = "{}_change".format(self.val_name)131        ev_reset = ParameterEvent(ev_reset_name, self.val_name)132        sm.set_reset_event(ev_reset)133        return sm134class VACModel:135    """136    Model of vaccination.137    VACModel uses following health parameters.138    - vac_type139    - vac_get140    VACModel uses following functions defined in parameter_model.141    - function which returns a choice selected randomly for a RandomEvent named "type_vac_branch".142      Choices are "vaccine" and "placebo".143    - function which returns a choice selected randomly for a RandomEvent named "get_vac_branch".144    - function which returns the wait time for a TimerEvent named "vac<N>_req".145    - function which returns the wait time for a TimerEvent named "vac<N>".146    VACModel provides following event.147    - ves["vac<N>"]148    VACModel uses following events.149    - ves["vac<N>_check"]150    - ves["vac<N>_cancel"]151    - ves["vac<N>_reset"]152    Constructor options153    -------------------154    name : str155        name of state machine.156    vac_num : int157        maximum number of vaccine doeses.158    parameter_model: HealthParameterModel159    ves : VirtualEventSet160    Examples161    --------162    >>> vac_model = VACModel("sm_vac", 2, hpm, ves)163    >>> sm_vac = vac_model.make_state_machine()164    >>> state_machines.append(sm_vac)165    """166    def __init__(self, name, vac_num, parameter_model, ves):167        self.name = name168        self.vac_num = vac_num169        self.parameter_model = parameter_model170        self.ves = ves171    def set_vac_type(self, vac_type, current_time, parameter):172        parameter["vac_type"] = vac_type173    def set_vac_get(self, vac_i, current_time, parameter):174        parameter["vac_get"] = vac_i175    def make_state_machine(self):176        sm = StateMachine(self.name)177        # RandomEvent178        ev1_name = "type_vac_branch"179        ev1 = RandomEvent(ev1_name, None, sm("start"))180        ev1.set_name_and_probability_func(181            lambda current_time, parameter, sm_name=self.name, ev_name=ev1_name: self.parameter_model.branch_prob(182                sm_name, ev_name, current_time, parameter183            )184        )185        for vac_type in ["vaccine", "placebo"]:186            state_name1 = "start"187            state_name2 = vac_type188            trg1 = Trigger(state_name2)189            sm.add_transition(state_name1, state_name2, trg1)190            ev1child_name = "{}".format(state_name2)191            ev1child = ev1.event(ev1child_name)192            trg1.set_event(ev1child)193            sm(state_name2).set_update_parameter_func(194                lambda current_time, parameter, vac_type=vac_type: self.set_vac_type(195                    vac_type, current_time, parameter196                )197            )198        state_name3 = "select_get_vac"199        trg3 = Trigger(state_name3)200        ev3 = DummyEvent("{}".format(state_name3))201        for vac_type in ["vaccine", "placebo"]:202            state_name2 = vac_type203            sm.add_transition(state_name2, state_name3, trg3)204            trg3.set_event(ev3)205        # RandomEvent206        ev2_name = "get_vac_branch"207        ev2 = RandomEvent(ev2_name, None, sm("select_get_vac"))208        ev2.set_name_and_probability_func(209            lambda current_time, parameter, sm_name=self.name, ev_name=ev2_name: self.parameter_model.branch_prob(210                sm_name, ev_name, current_time, parameter211            )212        )213        for vac_i in range(self.vac_num + 1):214            state_name1 = "select_get_vac"215            state_name2 = "get_vac_{}".format(vac_i)216            trg2 = Trigger(state_name2)217            sm.add_transition(state_name1, state_name2, trg2)218            ev2child_name = "{}".format(state_name2)219            ev2child = ev2.event(ev2child_name)220            trg2.set_event(ev2child)221            sm(state_name2).set_update_parameter_func(222                lambda current_time, parameter, vac_i=vac_i: self.set_vac_get(223                    vac_i, current_time, parameter224                )225            )226        state_name3 = "init_end"227        trg3 = Trigger(state_name3)228        ev3 = DummyEvent("{}".format(state_name3))229        for vac_i in range(self.vac_num + 1):230            state_name2 = "get_vac_{}".format(vac_i)231            sm.add_transition(state_name2, state_name3, trg3)232            trg3.set_event(ev3)233        # 1: init_end/vacN-1_after -(ev_vacN_req)----> vacN_req234        # 2: vacN_req              -(ev_vacN)--------> vacN_after235        # 3:                       -(ev_vacN_cancel)-> vacN_cancel236        # 4: vacN_cancel           -(ev_vacN_reset)--> init_end/vacN-1_after237        for vac_i in range(self.vac_num):238            if vac_i == 0:239                state_name0 = "init_end"240            else:241                state_name0 = "vac{}_after".format(vac_i)242            state_name1a = "vac{}_ready".format(vac_i + 1)243            state_name1b = "vac{}_out".format(vac_i)244            state_name2 = "vac{}_check".format(vac_i + 1)245            state_name3 = "vac{}_req".format(vac_i + 1)246            state_name4a = "vac{}_after".format(vac_i + 1)247            state_name4b = "vac{}_cancel".format(vac_i + 1)248            trg1a = Trigger(state_name1a)249            trg1b = Trigger(state_name1b)250            trg2 = Trigger(state_name2)251            trg3 = Trigger(state_name3)252            trg4a = Trigger(state_name4a)253            trg4b = Trigger(state_name4b)254            trg4c = Trigger("vac{}_reset".format(vac_i + 1))255            sm.add_transition(state_name0, state_name1a, trg1a)256            sm.add_transition(state_name0, state_name1b, trg1b)257            sm.add_transition(state_name1a, state_name2, trg2)258            sm.add_transition(state_name2, state_name3, trg3)259            sm.add_transition(state_name3, state_name4a, trg4a)260            sm.add_transition(state_name3, state_name4b, trg4b)261            sm.add_transition(state_name4b, state_name1a, trg4c)262            # ev_vacN_ready263            ev1a = ParameterEvent(264                "{}".format(state_name1a),265                lambda parameter, vac_i=vac_i: parameter["vac_get"] > vac_i,266            )267            trg1a.set_event(ev1a)268            # ev_vacN_out269            ev1b = ParameterEvent(270                "{}".format(state_name1b),271                lambda parameter, vac_i=vac_i: parameter["vac_get"] <= vac_i,272            )273            trg1b.set_event(ev1b)274            # ev_vacN_check275            trg2.set_event(self.ves["vac{}_check".format(vac_i + 1)])276            # ev_vacN_req277            ev3_name = "{}".format(state_name3)278            ev3 = TimerEvent(279                ev3_name,280                start_state_or_event=sm(state_name1a),281                interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.timer_interval(282                    sm_name, ev_name, current_time, parameter283                ),284            )285            trg3.set_event(ev3)286            # ev_vacN287            ev3a_name = "vac{}".format(vac_i + 1)288            self.ves[ev3a_name] = StateEvent(289                ev3a_name,290                sm(state_name4a),291            )292            ev4a = TimerEvent(293                "{}_timer".format(ev3a_name),294                start_state_or_event=sm(state_name3),295                interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev3a_name: self.parameter_model.timer_interval(296                    sm_name, ev_name, current_time, parameter297                ),298            )299            trg4a.set_event(ev4a)300            trg4b.set_event(self.ves["vac{}_cancel".format(vac_i + 1)])301            trg4c.set_event(self.ves["vac{}_reset".format(vac_i + 1)])302        return sm303class PCRModel:304    """305    Model of PCR test.306    PCRModel uses following functions defined in parameter_model.307    - function which returns a choice selected randomly for a RandomEvent named "PCR_reception_branch".308      Choices are "PCR_accept" and "PCR_reject".309    - function which returns a choice selected randomly for a RandomEvent named "PCR_result_branch".310      Choices are "PCR_positive" and "PCR_negative".311    - function which returns the wait time for a TimerEvent named "PCR_check".312    - function which returns the wait time for a TimerEvent named "PCR_result".313    PCRModel provides following event.314    - ves["on_PCR_positive"]315    PCRModel uses following event.316    - ves["PCR_request"]317    Constructor options318    -------------------319    name : str320        name of state machine.321    parameter_model: HealthParameterModel322    ves : VirtualEventSet323    Examples324    --------325    >>> pcr_model = PCRModel("sm_pcr", hpm, ves)326    >>> sm_pcr = pcr_model.make_state_machine()327    >>> state_machines.append(sm_pcr)328    """329    def __init__(self, name, parameter_model, ves):330        self.name = name331        self.parameter_model = parameter_model332        self.ves = ves333    def make_state_machine(self):334        sm = StateMachine(self.name)335        trg1 = Trigger("PCR_request")336        trg2 = Trigger("PCR_request_received")337        trg3a = Trigger("PCR_accept")338        trg3b = Trigger("PCR_reject")339        trg4 = Trigger("PCR_check")340        trg5 = Trigger("PCR_result")341        trg6a = Trigger("PCR_positive")342        trg6b = Trigger("PCR_negative")343        sm.add_transition("start", "PCR_request", trg1)344        sm.add_transition("PCR_request", "PCR_reception_branch", trg2)345        sm.add_transition("PCR_reception_branch", "PCR_accept", trg3a)346        sm.add_transition("PCR_reception_branch", "start", trg3b)347        sm.add_transition("PCR_accept", "PCR_check", trg4)348        sm.add_transition("PCR_check", "PCR_result", trg5)349        sm.add_transition("PCR_result", "PCR_positive", trg6a)350        sm.add_transition("PCR_result", "start", trg6b)351        trg1.set_event(self.ves["PCR_request"])352        # DummyEvent353        ev2 = DummyEvent("PCR_request_received")354        trg2.set_event(ev2)355        # RandomEvent356        ev3_name = "PCR_reception_branch"357        ev3 = RandomEvent(358            ev3_name,359            None,360            sm("PCR_reception_branch"),361        )362        ev3a = ev3.event("PCR_accept")363        ev3b = ev3.event("PCR_reject")364        ev3.set_name_and_probability_func(365            lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.branch_prob(366                sm_name, ev_name, current_time, parameter367            )368        )369        trg3a.set_event(ev3a)370        trg3b.set_event(ev3b)371        # TimerEvent372        ev4_name = "PCR_check"373        ev4 = TimerEvent(374            ev4_name,375            start_state_or_event=sm("PCR_request"),376            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4_name: self.parameter_model.timer_interval(377                sm_name, ev_name, current_time, parameter378            ),379        )380        trg4.set_event(ev4)381        # TimerEvent382        ev5_name = "PCR_result"383        ev5 = TimerEvent(384            ev5_name,385            start_state_or_event=sm("PCR_check"),386            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev5_name: self.parameter_model.timer_interval(387                sm_name, ev_name, current_time, parameter388            ),389        )390        trg5.set_event(ev5)391        # RandomEvent392        ev6_name = "PCR_result_branch"393        ev6 = RandomEvent(394            ev6_name,395            None,396            sm("PCR_check"),397        )398        ev6a = ev6.event("PCR_positive")399        ev6b = ev6.event("PCR_negative")400        ev6.set_name_and_probability_func(401            lambda current_time, parameter, sm_name=self.name, ev_name=ev6_name: self.parameter_model.branch_prob(402                sm_name, ev_name, current_time, parameter403            )404        )405        trg6a.set_event(ev6a)406        trg6b.set_event(ev6b)407        ev_on_PCR_positive = StateEvent(408            "on_PCR_positive",409            sm("PCR_positive"),410            EventType.LEVEL,411        )412        self.ves["on_PCR_positive"] = ev_on_PCR_positive413        return sm414class VACADVModel:415    """416    Model of vaccine adverse effect.417    VACADVModel uses following functions defined in parameter_model.418    - function which returns a choice selected randomly for a RandomEvent named "<prefix>_vac_branch".419    - function which returns the wait time for a TimerEvent named "<prefix>_none".420    - function which returns the wait time for a TimerEvent named "<prefix>_fever".421    - function which returns the wait time for a TimerEvent named "<prefix>_fever_before_worse".422    - function which returns the wait time for a TimerEvent named "<prefix>_worse".423    VACADVModel uses following event.424    - ves["vac<N>"]425    Constructor options426    -------------------427    name : str428        name of state machine.429    prefix : str430        prefix of parameter.431    vac_n : int432        n-th vaccine dose.433    parameter_model: HealthParameterModel434    ves : VirtualEventSet435    Examples436    --------437    >>> vac1adv_model = VACADVModel("sm_vac1adv", "vac1adv", 1, hpm, ves)438    >>> sm_vac1adv = vac1adv_model.make_state_machine()439    >>> state_machines.append(sm_vac1adv)440    """441    def __init__(self, name, prefix, vac_n, parameter_model, ves):442        self.name = name443        self.prefix = prefix444        self.vac_n = vac_n445        self.parameter_model = parameter_model446        self.ves = ves447    def make_state_machine(self):448        sm = StateMachine(self.name)449        state_name1 = "start"450        state_name2 = "{}_vac".format(self.prefix)451        state_name3a = "{}_none_path".format(self.prefix)452        state_name3b = "{}_fever_path".format(self.prefix)453        state_name3c = "{}_worse_path".format(self.prefix)454        state_name4a = "{}_none".format(self.prefix)455        state_name4b = "{}_fever".format(self.prefix)456        state_name4c = "{}_fever_before_worse".format(self.prefix)457        state_name5c = "{}_worse".format(self.prefix)458        trg2 = Trigger(state_name2)459        trg3a = Trigger(state_name3a)460        trg3b = Trigger(state_name3b)461        trg3c = Trigger(state_name3c)462        trg4a = Trigger(state_name4a)463        trg4b = Trigger(state_name4b)464        trg4c = Trigger(state_name4c)465        trg5c = Trigger(state_name5c)466        sm.add_transition(state_name1, state_name2, trg2)467        sm.add_transition(state_name2, state_name3a, trg3a)468        sm.add_transition(state_name2, state_name3b, trg3b)469        sm.add_transition(state_name2, state_name3c, trg3c)470        sm.add_transition(state_name3a, state_name4a, trg4a)471        sm.add_transition(state_name3b, state_name4b, trg4b)472        sm.add_transition(state_name3c, state_name4c, trg4c)473        sm.add_transition(state_name4c, state_name5c, trg5c)474        #475        trg2.set_event(self.ves["vac{}".format(self.vac_n)])476        # Random Event477        ev3_name = "{}_vac_branch".format(self.prefix)478        ev3 = RandomEvent(479            ev3_name,480            lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.branch_prob(481                sm_name, ev_name, current_time, parameter482            ),483            sm(state_name2),484        )485        ev3a = ev3.event("{}_none_path".format(self.prefix))486        ev3b = ev3.event("{}_fever_path".format(self.prefix))487        ev3c = ev3.event("{}_worse_path".format(self.prefix))488        trg3a.set_event(ev3a)489        trg3b.set_event(ev3b)490        trg3c.set_event(ev3c)491        # TimerEvent492        ev4a_name = "{}_none".format(self.prefix)493        ev4a = TimerEvent(494            ev4a_name,495            start_state_or_event=sm(state_name3a),496            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4a_name: self.parameter_model.timer_interval(497                sm_name, ev_name, current_time, parameter498            ),499        )500        trg4a.set_event(ev4a)501        # TimerEvent502        ev4b_name = "{}_fever".format(self.prefix)503        ev4b = TimerEvent(504            ev4b_name,505            start_state_or_event=sm(state_name3b),506            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4b_name: self.parameter_model.timer_interval(507                sm_name, ev_name, current_time, parameter508            ),509        )510        trg4b.set_event(ev4b)511        # TimerEvent512        ev4c_name = "{}_fever_before_worse".format(self.prefix)513        ev4c = TimerEvent(514            ev4c_name,515            start_state_or_event=sm(state_name3c),516            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4c_name: self.parameter_model.timer_interval(517                sm_name, ev_name, current_time, parameter518            ),519        )520        trg4c.set_event(ev4c)521        # TimerEvent522        ev5c_name = "{}_worse".format(self.prefix)523        ev5c = TimerEvent(524            ev5c_name,525            start_state_or_event=sm(state_name4c),526            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev5c_name: self.parameter_model.timer_interval(527                sm_name, ev_name, current_time, parameter528            ),529        )530        trg5c.set_event(ev5c)531        return sm532class VACEffModel:533    """534    Model of vaccine effect.535    VACEffModel uses following functions defined in parameter_model.536    - function which returns a choice selected randomly for a RandomEvent named "<prefix>_vac_branch".537    - function which returns the wait time for a TimerEvent named "<prefix>_no_effect".538    - function which returns the wait time for a TimerEvent named "<prefix>_effect".539    VACEffModel uses following event.540    - ves["vac<N>"]541    Constructor options542    -------------------543    name : str544        name of state machine.545    prefix : str546        prefix of parameter.547    vac_n : int548        n-th vaccine dose.549    parameter_model: HealthParameterModel550    ves : VirtualEventSet551    Examples552    --------553    >>> vac1eff_model = VACEffModel("sm_vac1eff", "vac1eff", 1, hpm, ves)554    >>> sm_vac1eff = vac1eff_model.make_state_machine()555    >>> state_machines.append(sm_vac1eff)556    """557    def __init__(self, name, prefix, vac_n, parameter_model, ves):558        self.name = name559        self.prefix = prefix560        self.vac_n = vac_n561        self.parameter_model = parameter_model562        self.ves = ves563    def make_state_machine(self):564        sm = StateMachine(self.name)565        state_name1 = "start"566        state_name2 = "{}_vac".format(self.prefix)567        state_name3a = "{}_no_effect_path".format(self.prefix)568        state_name4a = "{}_no_effect".format(self.prefix)569        state_name3b = "{}_effect_path".format(self.prefix)570        state_name4b = "{}_effect".format(self.prefix)571        trg2 = Trigger(state_name2)572        trg3a = Trigger(state_name3a)573        trg3b = Trigger(state_name3b)574        trg4a = Trigger(state_name4a)575        trg4b = Trigger(state_name4b)576        sm.add_transition(state_name1, state_name2, trg2)577        sm.add_transition(state_name2, state_name3a, trg3a)578        sm.add_transition(state_name2, state_name3b, trg3b)579        sm.add_transition(state_name3a, state_name4a, trg4a)580        sm.add_transition(state_name3b, state_name4b, trg4b)581        #582        trg2.set_event(self.ves["vac{}".format(self.vac_n)])583        # Random Event584        ev3_name = "{}_vac_branch".format(self.prefix)585        ev3 = RandomEvent(586            ev3_name,587            lambda current_time, parameter, sm_name=self.name, ev_name=ev3_name: self.parameter_model.branch_prob(588                sm_name, ev_name, current_time, parameter589            ),590            sm(state_name2),591        )592        ev3a = ev3.event("{}_no_effect_path".format(self.prefix))593        ev3b = ev3.event("{}_effect_path".format(self.prefix))594        trg3a.set_event(ev3a)595        trg3b.set_event(ev3b)596        # TimerEvent597        ev4a_name = "{}_no_effect".format(self.prefix)598        ev4a = TimerEvent(599            ev4a_name,600            start_state_or_event=sm(state_name3a),601            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4a_name: self.parameter_model.timer_interval(602                sm_name, ev_name, current_time, parameter603            ),604        )605        trg4a.set_event(ev4a)606        # TimerEvent607        ev4b_name = "{}_effect".format(self.prefix)608        ev4b = TimerEvent(609            ev4b_name,610            start_state_or_event=sm(state_name3b),611            interval=lambda current_time, parameter, sm_name=self.name, ev_name=ev4b_name: self.parameter_model.timer_interval(612                sm_name, ev_name, current_time, parameter613            ),614        )615        trg4b.set_event(ev4b)616        return sm617class SickSumModel:618    """619    Summation of SickModels.620    SickModel uses following health parameters.621    - <sick_name>_val where <sick_name> is on sick_list.622    - <prefix>_status623    - <prefix>_value624    By default, <prefix>_value is the maximum value among <sick_name>_vals.625    Constructor options626    -------------------627    name : str628        name of state machine.629    prefix : str630        prefix of parameter.631    level_num : int632        maximum level of symptom.633    sick_list: list634        list of sick.635    ves : VirtualEventSet636    Examples637    --------638    >>> fever_level = ["no_fever", "subclinical", "fever"]639    >>> sick_list = ["cold", "c19", "vacadv"]640    >>> feversum_model = SickSumModel(641            "sm_feversum", "feversum", len(fever_level), sick_list, None642        )643    # example of set_calc_sick_value644    >>> fever_level = ["no_fever", "subclinical", "fever"]645    >>> sick_list = ["cold", "c19", "vacadv"]646    >>> feversum_model = SickSumModel(647            "sm_feversum", "feversum", len(fever_level), sick_list, None648        )649    >>> def calc_fever_value(parameter, sick_list):650            val = 0651            for sick_name in sick_list:652                sick_val = parameter.get("{}_val".format(sick_name))653                if sick_val is not None:654                    val += sick_val655            return val656    >>> feversum_model.set_calc_sick_value(calc_fever_value)657    >>> sm_feversum = feversum_model.make_state_machine()658    """659    def __init__(self, name, prefix, level_num, sick_list, ves):660        self.name = name661        self.prefix = prefix662        self.level_num = level_num663        self.sick_list = sick_list664        self.ves = ves665        self.calc_sick_value = self.calc_sick_max_value666    def calc_sick_max_value(self, parameter, sick_list):667        max_val = 0668        smpl_log.info("parameter={}, sick_list={}".format(parameter, sick_list))669        for sick_name in sick_list:670            sick_val = parameter.get("{}_val".format(sick_name))671            if sick_val is not None and sick_val > max_val:672                max_val = sick_val673        return max_val674    def set_calc_sick_value(self, calc_sick_value):675        self.calc_sick_value = calc_sick_value676    def make_state_machine(self):677        """678        Make a state machine.679        Returns:680            StateMachine : state machine of SickModel681        """682        sm = StateMachine(self.name)683        def set_parameter(prefix, state_name, val, current_time, parameter):684            parameter["{}_status".format(prefix)] = state_name685            parameter["{}_val".format(prefix)] = val686        for level in range(self.level_num):687            if level == 0:688                state_name1 = "start"689                state_name2 = "{}_{}".format(self.prefix, level)690                trg1a_name = "{}_up_to_{}".format(self.prefix, level)691                trg1a = Trigger(trg1a_name)692                sm.add_transition(state_name1, state_name2, trg1a)693                sm(state_name2).set_update_parameter_func(694                    lambda current_time, parameter, prefix=self.prefix, state_name=state_name2, val=level: set_parameter(695                        prefix, state_name, val, current_time, parameter696                    )697                )698                ev1a = ParameterEvent(699                    trg1a_name,700                    lambda parameter, level=-1: self.calc_sick_value(701                        parameter, self.sick_list702                    )703                    > level,704                )705                trg1a.set_event(ev1a)706            else:707                state_name1 = "{}_{}".format(self.prefix, level - 1)708                state_name2 = "{}_{}".format(self.prefix, level)709                trg1a_name = "{}_up_to_{}".format(self.prefix, level)710                trg1b_name = "{}_down_to_{}".format(self.prefix, level - 1)711                trg1a = Trigger(trg1a_name)712                trg1b = Trigger(trg1b_name)713                sm.add_transition(state_name1, state_name2, trg1a)714                sm.add_transition(state_name2, state_name1, trg1b)715                sm(state_name2).set_update_parameter_func(716                    lambda current_time, parameter, prefix=self.prefix, state_name=state_name2, val=level: set_parameter(717                        prefix, state_name, val, current_time, parameter718                    )719                )720                ev1a = ParameterEvent(721                    trg1a_name,722                    lambda parameter, level=level: self.calc_sick_value(723                        parameter, self.sick_list724                    )725                    >= level,726                )727                ev1b = ParameterEvent(728                    trg1b_name,729                    lambda parameter, level=level: self.calc_sick_value(730                        parameter, self.sick_list731                    )732                    < level,733                )734                trg1a.set_event(ev1a)735                trg1b.set_event(ev1b)...

Events.py

Source:Events.py

...3class Ref(weakref.ref):4    def __getattribute__(self, name):5        return self().__getattribute__(name)6class BaseEventManager(EventManager):7    def set_event(self, name, *aliases):8        self[name] = Event()9        for alias in aliases:10            self.aliases[alias] = Ref(self[name])11    def __init__(self):12        super(BaseEventManager, self).__init__()13        self.aliases = {}14        self.set_event("recv00", "recv_keepalive")15        self.set_event("sent00", "sent_keepalive")16        self.set_event("recv01", "recv_login_request")17        self.set_event("sent02", "sent_handshake")18        self.set_event("recv03", "recv_chat_message")19        self.set_event("sent03", "sent_chat_message")20        self.set_event("recv04", "recv_time_update")21        self.set_event("recv05", "recv_entity_equipment")22        self.set_event("recv06", "recv_spawn_position")23        self.set_event("sent07", "sent_use_entity")24        self.set_event("recv08", "recv_update_health")25        self.set_event("recv09", "recv_respawn")26        self.set_event("recv0A", "recv_player")27        self.set_event("sent0B", "sent_player_position")28        self.set_event("sent0C", "sent_player_look")29        self.set_event("recv0D", "recv_player_position_and_look")30        self.set_event("sent0D", "sent_player_position_and_look")31        self.set_event("sent0E", "sent_player_digging")32        self.set_event("sent0F", "sent_player_block_placement")33        self.set_event("recv10", "recv_held_item_change")34        self.set_event("sent10", "sent_held_item_change")35        self.set_event("recv11", "recv_use_bed")36        self.set_event("recv12", "recv_animation")37        self.set_event("sent12", "sent_animation")38        self.set_event("sent13", "sent_entity_action")39        self.set_event("recv14", "recv_spawn_named_entity")40        self.set_event("recv16", "recv_collect_item")41        self.set_event("recv17", "recv_spawn_object")42        self.set_event("recv18", "recv_spawn_mob")43        self.set_event("recv19", "recv_spawn_painting")44        self.set_event("recv1A", "recv_spawn_exp_orb")45        self.set_event("sent1B", "sent_steer_vehicle")46        self.set_event("recv1C", "recv_entity_velocity")47        self.set_event("recv1D", "recv_destroy_entity")48        self.set_event("recv1E", "recv_entity")49        self.set_event("recv1F", "recv_entity_relative_move")50        self.set_event("recv20", "recv_entity_look")51        self.set_event("recv21", "recv_entity_look_and_relative_move")52        self.set_event("recv22", "recv_entity_teleport")53        self.set_event("recv23", "recv_entity_head_look")54        self.set_event("recv26", "recv_entity_status")55        self.set_event("recv27", "recv_attach_entity")56        self.set_event("recv28", "recv_entity_metadata")57        self.set_event("recv29", "recv_entity_effect")58        self.set_event("recv2A", "recv_remove_entity_effect")59        self.set_event("recv2B", "recv_set_exp")60        self.set_event("recv2C", "recv_entity_properties")61        self.set_event("recv33", "recv_chunk_data")62        self.set_event("recv34", "recv_multi_block_change")63        self.set_event("recv35", "recv_block_change")64        self.set_event("recv36", "recv_block_action")65        self.set_event("recv37", "recv_block_break_animation")66        self.set_event("recv38", "recv_map_chunk_bulk")67        self.set_event("recv3C", "recv_explosion")68        self.set_event("recv3D", "recv_sound_or_particle_effect")69        self.set_event("recv3E", "recv_named_sound_effect")70        self.set_event("recv3F", "recv_particle")71        self.set_event("recv46", "recv_change_game_state")72        self.set_event("recv47", "recv_spawn_global_entity")73        self.set_event("recv64", "recv_open_window")74        self.set_event("recv65", "recv_close_window")75        self.set_event("sent65", "sent_close_window")76        self.set_event("sent66", "sent_click_window")77        self.set_event("recv67", "recv_set_slot")78        self.set_event("recv68", "recv_set_window_items")79        self.set_event("recv69", "recv_update_window_property")80        self.set_event("recv6A", "recv_confirm_transaction")81        self.set_event("sent6A", "sent_confirm_transaction")82        self.set_event("recv6B", "recv_creative_inventory_action")83        self.set_event("sent6B", "sent_creative_inventory_action")84        self.set_event("sent6C", "sent_enchant_item")85        self.set_event("recv82", "recv_update_sign")86        self.set_event("sent82", "sent_update_sign")87        self.set_event("recv83", "recv_item_data")88        self.set_event("recv84", "recv_update_tile_entity")89        self.set_event("recvC8", "recv_increment_statistic")90        self.set_event("recvC9", "recv_player_list_item")91        self.set_event("recvCA", "recv_player_abilities")92        self.set_event("sentCA", "sent_player_abilities")93        self.set_event("recvCB", "recv_tab_complete")94        self.set_event("sentCB", "sent_tab_complete")95        self.set_event("sentCC", "sent_client_settings")96        self.set_event("sentCD", "sent_client_statuses")97        self.set_event("recvCE", "recv_scoreboard_objective")98        self.set_event("recvCF", "recv_update_score")99        self.set_event("recvD0", "recv_display_scoreboard")100        self.set_event("recvD1", "recv_teams")101        self.set_event("recvFA", "recv_plugin_message")102        self.set_event("sentFA", "sent_plugin_message")103        self.set_event("recvFC", "recv_encryption_key_response")104        self.set_event("sentFC", "sent_encryption_key_response")105        self.set_event("recvFD", "recv_encryption_key_request")106        self.set_event("sentFE", "recv_client_list_ping")107        self.set_event("recvFF", "recv_client_disconnect")108        self.set_event("sentFF", "sent_client_disconnect")109    def __getitem__(self, key):110        try:111            return super(BaseEventManager, self).__getitem__(key)112        except KeyError:113            return self.aliases[key]114class EventManager(BaseEventManager):115    def __init__(self, connection, *args, **kwargs):116        self.connection = connection117        super(EventManager, self).__init__(*args, **kwargs)118        self.recv_packet = Event()119        self.recv_packet.name = "RECV"120        self.recv_packet.eventmanager = None121        self.sent_packet = Event()122        self.sent_packet.name = "SENT"...

test_event.py

Source:test_event.py

...22    apigw_event = json.loads(open(datadir.join('apigw_event_err.json'), 'rt').read())23    event = Event()24    with pytest.raises(Exception):25        event.parse_event(apigw_event)26def test_set_event(datadir):27    apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())28    event = Event()29    event.set_event(apigw_event)30    assert event.event['path'] == '/the/cool/path'31def test_set_http_method(datadir):32    apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())33    event = Event()34    event.set_event(apigw_event)35    event.set_http_method()36    assert event.method == 'POST'37def test_set_http_method_error_1(datadir):38    event = Event()39    with pytest.raises(AwAwsMissingRequirement, match='Event not set'):40        event.set_http_method()41def test_set_http_method_error_2(datadir):42    apigw_event = json.loads(open(datadir.join('apigw_event_3.json'), 'rt').read())43    event = Event()44    event.set_event(apigw_event)45    with pytest.raises(AwAwsInvalidHttpMethod, match='Not a valid http method: BLARGH'):46        event.set_http_method()47def test_set_body_encoded(datadir):48    apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())49    event = Event()50    event.set_event(apigw_event)51    event.set_body()52    print(event.body)53    assert event.body['data_stuff'] == "down is the new up"54def test_set_body_dencoded(datadir):55    apigw_event = json.loads(open(datadir.join('apigw_event_2.json'), 'rt').read())56    event = Event()57    event.set_event(apigw_event)58    event.set_body()59    assert event.body['data_stuff'] == "down is the new up"60def test_set_body_dencoded_error(datadir):61    apigw_event = json.loads(open(datadir.join('apigw_event_4.json'), 'rt').read())62    event = Event()63    event.set_event(apigw_event)64    with pytest.raises(Exception, match=r'Error decoding'):65        event.set_body()66def test_set_body_no_body(datadir):67    apigw_event = json.loads(open(datadir.join('apigw_event_3.json'), 'rt').read())68    event = Event()69    event.set_event(apigw_event)70    event.set_body()71    assert event.body is None72def test_set_body_no_event():73    event = Event()74    with pytest.raises(AwAwsMissingRequirement,75                       match=r'use event.set_event()'76                       ):77        event.set_body()78def test_set_qstring_params(datadir):79    apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())80    event = Event()81    event.set_event(apigw_event)82    event.set_qstring_params()83    assert event.params['foo'] == 'bar'84def test_set_qstring_params_error(datadir):85    event = Event()86    with pytest.raises(AwAwsMissingRequirement, match='Event not set'):87        event.set_qstring_params()88def test_set_qstring_params_no_qsring(datadir):89    apigw_event = json.loads(open(datadir.join('apigw_event_4.json'), 'rt').read())90    event = Event()91    event.set_event(apigw_event)92    check = event.set_qstring_params()93    assert check is None94def test_set_path_params(datadir):95    apigw_event = json.loads(open(datadir.join('apigw_event.json'), 'rt').read())96    event = Event()97    event.set_event(apigw_event)98    event.set_path_params()99    assert event.path == '/the/cool/path'100def test_set_path_params_error(datadir):101    event = Event()102    with pytest.raises(AwAwsMissingRequirement, match='Event not set'):103        event.set_path_params()104def test_set_path_missing(datadir):105    apigw_event = json.loads(open(datadir.join('apigw_event_no_path.json'), 'rt').read())106    event = Event()107    event.set_event(apigw_event)108    event.set_path_params()...

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.