KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _hold_event method in Lemoncheesecake
Best Python code snippet using lemoncheesecake
Timer.py
Source:Timer.py 
1# Copyright L.P.Klyne 2013 2# Licenced under 3 clause BSD licence 3#4#5# Generic On-Off timer6#7# Andy Harris --- 24th September 20108#9#10# <eventInterface module='EventHandlers.LightTimer' name='LightTimer' >11# <presence type='http://id.webbrick.co.uk/events/webbrick/DO' source='webbrick/36/DO/6' key="state" invert="true"/>12# <duration type='http://id.webbrick.co.uk/events/config/get' source='family_bathroom_on_time' /> 13# <enable type='http://id.webbrick.co.uk/events/config/get' source='family_bathroom_on' /> 14# <hold type='http://id.webbrick.co.uk/events/webbrick/DO' source='webbrick/36/DO/4' key="state" invert="true" /> 15# 16# <eventtype type="http://id.webbrick.co.uk/events/timer" >17# <eventsource source="family/bathroom/timer">18# <event>19# <params>20# <testEq name="dayphase" value='*:dark' />21# <testEq name="occupancy" value='home,away' />22# <testEq name="hold" value="0" />23# </params>24# <newEvent type='internal' source='familybath/lights/tryon' >25# <copy_other_data/>26# </newEvent>27# </event>28# </eventsource>29# </eventtype>30# </eventInterface>31import logging32from EventLib.Event import Event33from EventLib.Status import StatusVal34from EventLib.SyncDeferred import makeDeferred35from EventHandlers.BaseHandler import BaseHandler36from EventHandlers.Utils import *37# make logging global to module.38_log = None39CONFIG_TYPE = "http://id.webbrick.co.uk/events/config/get"40class Timer( BaseHandler ):41 """42 Timer class that subscribes to events and generates timer events43 """44 45 def __init__ (self, localRouter):46 super(Timer,self).__init__(localRouter)47 global _log48 _log = self._log # make global49 self._debug = True # 50 self._counter = 0 # Will be used for counting51 self._timer_event = None52 self._light_state_event = None53 self._timer_evt_type = 'http://id.webbrick.co.uk/events/timer'54 self._timer_evt_source = self._log.name55 self._duration = 12056 self._duration_event = None57 self._enable = True58 self._enable_event = None59 self._hold = 060 self._hold_event = None61 self._presence = -1 # -1: Unknown 0: No Presence 1: Presence Detected62 self._p_presence = -2 # -1: Unknown 0: No Presence 1: Presence Detected -2: Initial63 self._light_state = -1 # -1: Unknown 0: Off 1: On -2: Initial64 self._p_light_state = -2 # -1: Unknown 0: Off 1: On65 self._threshold = 50.0 # default to mid point66 67 self._occupancy = "Home" # a default value68 self._isDark = 0 # a default value69 self._dayphasetext = "Unknown" # a default value70 def start(self):71 self._log.debug( 'start' )72 self._localRouter.subscribe( self._subscribeTime, self, 'http://id.webbrick.co.uk/events/config/get' )73 self._localRouter.subscribe( self._subscribeTime, self, 'http://id.webbrick.co.uk/events/time/dayphaseext' )74 self._localRouter.subscribe( self._subscribeTime, self, 'http://id.webbrick.co.uk/events/time/isDark' )75 self._localRouter.subscribe( self._subscribeTime, self, 'http://id.webbrick.co.uk/events/time/second', 'time/second' )76 #77 # Subscribe to the presence, enable and hold event type78 #79 self._localRouter.subscribe( self._subscribeTime, self, self._presence_event['type'] )80 if self._enable_event :81 self._localRouter.subscribe( self._subscribeTime, self, self._enable_event['type'] )82 if self._duration_event :83 self._localRouter.subscribe( self._subscribeTime, self, self._duration_event['type'] )84 if self._hold_event :85 self._localRouter.subscribe( self._subscribeTime, self, self._hold_event['type'] )86 if self._light_state_event :87 self._localRouter.subscribe( self._subscribeTime, self, self._light_state_event['type'] )88 self.subscribeAll()89 def stop(self):90 self._log.debug( 'stop' )91 if self._debug:92 self._log.debug ("--------- Variables Were ----------------")93 self._log.debug ("MyType %s" % str(self._timer_evt_type)) 94 self._log.debug ("MySource %s" % str(self._timer_evt_source)) 95 self._log.debug ("Enabled %s" % str(self._enable)) 96 self._log.debug ("Duration %s" % str(self._duration)) 97 self._log.debug ("Occupancy %s" % str(self._occupancy)) 98 self._log.debug ("DayPhase %s" % str(self._dayphasetext)) 99 self._log.debug ("Dark %s" % str(self._isDark)) 100 self._log.debug ("Presence %s" % str(self._presence)) 101 self._log.debug ("Light_State %s" % str(self._light_state)) 102 self._log.debug ("-----------------------------------------")103 self._localRouter.unsubscribe( self, 'http://id.webbrick.co.uk/events/config/get' )104 self._localRouter.unsubscribe( self, 'http://id.webbrick.co.uk/events/time/dayphaseext' )105 self._localRouter.unsubscribe( self, 'http://id.webbrick.co.uk/events/time/isDark' )106 self._localRouter.unsubscribe( self, self._presence_event['type'] )107 self._localRouter.unsubscribe( self, 'http://id.webbrick.co.uk/events/time/second', 'time/second' )108 if self._enable_event :109 self._localRouter.unsubscribe( self, self._enable_event['type'] )110 if self._duration_event :111 self._localRouter.unsubscribe( self, self._duration_event['type'] )112 if self._hold_event :113 self._localRouter.unsubscribe( self, self._hold_event['type'] )114 if self._light_state_event :115 self._localRouter.unsubscribe( self, self._light_state_event['type'] )116 self.unSubscribeAll()117 def configure( self, cfgDict ):118 from string import upper119 super(Timer,self).configure(cfgDict)120 self._log.debug(cfgDict)121 if cfgDict.has_key('presence'):122 self._presence_event = cfgDict['presence']123 if self._presence_event.has_key('invert'):124 if upper(self._presence_event['invert']) not in ("TRUE","1"):125 del self._presence_event['invert'] # remove the key for a non true or 1 value126 else:127 _log.error( 'No presence event defined for Timer')128 if cfgDict.has_key('light_state'):129 self._light_state_event = cfgDict['light_state']130 if self._light_state_event.has_key('invert'):131 if upper(self._light_state_event['invert']) not in ("TRUE","1"):132 del self._light_state_event['invert'] # remove the key for a non true or 1 value133 if self._light_state_event.has_key('threshold'):134 self._threshold = float(self._light_state_event['threshold'])135 if cfgDict.has_key('enable'):136 self._enable_event = cfgDict['enable']137 138 if cfgDict.has_key('duration'):139 self._duration_event = cfgDict['duration']140 if cfgDict.has_key('hold'):141 self._hold_event = cfgDict['hold']142 if self._hold_event.has_key('invert'):143 if upper(self._hold_event['invert']) not in ("TRUE","1"):144 del self._hold_event['invert'] # remove the key for a non true or 1 value145 if cfgDict.has_key('eventtype'):146 if cfgDict['eventtype'].has_key('type'):147 self._timer_evt_type = cfgDict['eventtype']['type']148 149 if cfgDict['eventtype'].has_key('eventsource'):150 if cfgDict['eventtype']['eventsource'].has_key('source'):151 self._timer_evt_source = cfgDict['eventtype']['eventsource']['source']152 153 if self._debug:154 self._log.debug ("--------- Config Debug ----------------")155 self._log.debug ("presence event %s" % str(self._presence_event)) 156 self._log.debug ("duration event %s" % str(self._duration_event)) 157 self._log.debug ("enable event %s" % str(self._enable_event))158 self._log.debug ("hold event %s" % str(self._hold_event))159 self._log.debug ("light_state event %s" % str(self._light_state_event))160 self._log.debug ("---------------------------------------")161 def doActions( self, actions, inEvent ):162 if actions:163 for action in actions:164 # logged in BaseHandler.sendEvent165 self._log.debug( 'Generate event %s' % ( action ) )166 self.sendEvent( makeNewEvent( action, inEvent, None ) )167 168 def configureActions( self, cfgDict ):169 self._log.debug("configureActions %s" % (cfgDict) )170 result = None171 if cfgDict.has_key("newEvent"):172 if isinstance( cfgDict["newEvent"], list ):173 result = cfgDict["newEvent"]174 else:175 result = list()176 result.append(cfgDict["newEvent"])177 self._log.debug("configureActions %s" % (result) )178 return result179 def indexTimer(self):180 if self._timer_event :181 #182 # An event exists therefore index the counter183 #184 self._counter += 1185 if (self._counter >= self._duration):186 self.sendTimerEvent()187 def setTimerEvent(self):188 self._log.debug ("Setting Timer Event ....")189 self._timer_event = Event(self._timer_evt_type,self._timer_evt_source, {'hold': self._hold, 'occupancy': self._occupancy, 'isDark': self._isDark, 'dayphase': self._dayphasetext, 'presence': self._presence, 'light_state': self._light_state} ) 190 self._counter = 0 # reset191 def sendTimerEvent(self):192 self._log.debug ("Sending Timer Event ....")193 self._timer_event = Event(self._timer_evt_type,self._timer_evt_source, {'hold': self._hold, 'occupancy': self._occupancy, 'isDark': self._isDark, 'dayphase': self._dayphasetext, 'presence': self._presence, 'light_state': self._light_state} ) 194 self.sendEvent( self._timer_event)195 self._timer_event = None # reset196 def doHandleConfig( self, inEvent ):197 from string import upper198 src = inEvent.getSource()199 #self._log.debug ("Found Event: %s" % str(inEvent))200 #self._log.debug ("Handle Config Event SRC %s" % str(src))201 #202 # Now see if this matches anything we need203 #204 if self._enable_event:205 if self._enable_event['type'] == CONFIG_TYPE:206 if self._enable_event['source'] == src:207 en = inEvent.getPayload()['val']208 if en == "1":209 self._enable = True210 if upper(en) == "TRUE":211 self._enable = True212 else:213 self._enable = False214 if self._duration_event['type'] == CONFIG_TYPE:215 if self._duration_event['source'] == src:216 self._duration = int(inEvent.getPayload()['val'])217 218 if src == 'occupants/home':219 self._occupancy = int(inEvent.getPayload()['val']) 220 def doHandleHold( self, inEvent ):221 #222 # What sense is Hold in223 #224 lstate = int(inEvent.getPayload()['state'])225 if self._hold_event.has_key('invert'):226 self._log.debug ("Handle invert for hold: %s" % inEvent.getPayload()['state'])227 if lstate == 1:228 self._hold = 0229 else:230 self._hold = 1231 else:232 if lstate == 1:233 self._hold = 1234 else:235 self._hold = 0236 self._log.debug ("hold set to: %s" % self._hold)237 238 def doHandleLightState( self, inEvent ):239 self._log.debug ("Found Event for Light_State: %s" % str(inEvent))240 #241 # light_state may come in from a DO,AO or AI242 # these need to be treated slightly differently243 #244 # PREPROCESS to derive lstate245 if inEvent.getType() == 'http://id.webbrick.co.uk/events/webbrick/DO' :246 # digital Output247 lstate = inEvent.getPayload()['state']248 elif (inEvent.getType() == 'http://id.webbrick.co.uk/events/webbrick/AI') or (inEvent.getType() == 'http://id.webbrick.co.uk/events/webbrick/AI'):249 # Analogue In or Out 250 lval = inEvent.getPayload()['val']251 if float(lval) >= self._threshold :252 lstate = 1253 else:254 lstate = 0255 else:256 # not a state we recognise257 self._log.error ("Wrong Event Type for light_state: %s" % str(inEvent))258 259 if self._light_state_event.has_key('invert'):260 if lstate == 1:261 doit = False262 self._light_state = 0263 else:264 doit = True265 self._light_state = 1266 else:267 if lstate == 1:268 doit = True269 self._light_state = 1270 else:271 doit = False272 self._light_state = 0273 if doit: 274 if not self._hold:275 if self._enable:276 if not (self._p_light_state == self._light_state) :277 self._log.debug ("Setting Timer due to light_state" )278 self.setTimerEvent()279 self._p_light_state = self._light_state280 def doHandlePresence( self, inEvent ):281 lstate = int(inEvent.getPayload()['state'])282 self._log.debug ("Found Event for Presence: %s previous %s" % (lstate, self._p_presence) )283 if self._presence_event.has_key('invert'):284 self._log.debug ("Processing Presence for invert")285 if lstate == 1:286 doit = False287 self._presence = 0288 else:289 doit = True290 self._presence = 1291 else:292 if lstate == 1:293 doit = True294 self._presence = 1295 else:296 doit = False297 self._presence = 0298 self._log.debug ("Process Presence: %s Previous %s Hold %s Enable %s" % (self._presence, self._p_presence, self._hold, self._enable) )299 if doit: 300 if not self._hold:301 if self._enable:302 if not (self._presence == self._p_presence):303 self._log.debug ("Setting Timer due to Presence" )304 self.setTimerEvent()305 self._p_presence = self._presence306 def doHandleDayPhase( self, inEvent ):307 src = inEvent.getSource()308 #self._log.debug ("Found Event: %s" % str(inEvent))309 #self._log.debug ("Handle DayPhase Event SRC %s" % str(src))310 if src == 'time/dayphaseext':311 self._dayphasetext = inEvent.getPayload()['dayphasetext']312 def doHandleDark( self, inEvent ):313 src = inEvent.getSource()314 #self._log.debug ("Found Event: %s" % str(inEvent))315 #self._log.debug ("Handle Dark Event SRC %s" % str(src))316 if src == 'time/isDark':317 self._isDark = inEvent.getPayload()['state']318 def doHandleEvent( self, handler, inEvent ):319 320 if inEvent.getType() == 'http://id.webbrick.co.uk/events/config/get':321 self.doHandleConfig( inEvent )322 return makeDeferred(StatusVal.OK) 323 324 elif inEvent.getType() == 'http://id.webbrick.co.uk/events/time/dayphaseext':325 self.doHandleDayPhase( inEvent ) 326 return makeDeferred(StatusVal.OK) 327 328 elif inEvent.getType() == 'http://id.webbrick.co.uk/events/time/isDark':329 self.doHandleDark( inEvent )330 return makeDeferred(StatusVal.OK) 331 332 elif inEvent.getType() == "http://id.webbrick.co.uk/events/time/second":333 self.indexTimer()334 return makeDeferred(StatusVal.OK)335 elif inEvent.getType() == self._presence_event['type'] and inEvent.getSource() == self._presence_event['source']:336 self.doHandlePresence( inEvent )337 return makeDeferred(StatusVal.OK) 338 339 elif self._hold_event and inEvent.getType() == self._hold_event['type'] and inEvent.getSource() == self._hold_event['source']:340 self.doHandleHold( inEvent )341 return makeDeferred(StatusVal.OK) 342 elif self._light_state_event and inEvent.getType() == self._light_state_event['type'] and inEvent.getSource() == self._light_state_event['source']:343 self.doHandleLightState( inEvent )344 return makeDeferred(StatusVal.OK) 345 else: 346 return super(Timer,self).doHandleEvent( handler, inEvent)...
session.py
Source:session.py
...60 return self._local.cursor61 @cursor.setter62 def cursor(self, cursor):63 self._local.cursor = cursor64 def _hold_event(self, event):65 self.cursor.pending_events.append(event)66 def _flush_pending_events(self):67 for event in self.cursor.pending_events:68 self.event_manager.fire(event)69 del self.cursor.pending_events[:]70 def _discard_pending_event_if_any(self, event_class):71 if self.cursor.pending_events and isinstance(self.cursor.pending_events[-1], event_class):72 self.cursor.pending_events.pop()73 return True74 else:75 return False76 def _discard_or_fire_event(self, event_class, event):77 discarded = self._discard_pending_event_if_any(event_class)78 if not discarded:79 self.event_manager.fire(event)80 def _mark_location_as_failed(self, location):81 self._failures.add(location)82 def is_successful(self, location=None):83 if location:84 return location not in self._failures85 else:86 return len(self._failures) == 087 def start_step(self, description):88 self._end_step_if_any()89 self.cursor.step = description90 self._hold_event(91 events.StepStartEvent(self.cursor.location, description, _get_thread_id())92 )93 set_step = start_step94 def end_step(self):95 assert self.cursor.step, "There is no started step"96 self._discard_or_fire_event(97 events.StepStartEvent, events.StepEndEvent(self.cursor.location, self.cursor.step, _get_thread_id())98 )99 self.cursor.step = None100 def _end_step_if_any(self):101 if self.cursor.step:102 self.end_step()103 def _log(self, level, content):104 self._flush_pending_events()105 if level == Log.LEVEL_ERROR:106 self._mark_location_as_failed(self.cursor.location)107 self.event_manager.fire(108 events.LogEvent(self.cursor.location, self.cursor.step, _get_thread_id(), level, content)109 )110 def log_debug(self, content):111 return self._log(Log.LEVEL_DEBUG, content)112 def log_info(self, content):113 return self._log(Log.LEVEL_INFO, content)114 def log_warning(self, content):115 return self._log(Log.LEVEL_WARN, content)116 def log_error(self, content):117 return self._log(Log.LEVEL_ERROR, content)118 def log_check(self, description, is_successful, details):119 self._flush_pending_events()120 if is_successful is False:121 self._mark_location_as_failed(self.cursor.location)122 self.event_manager.fire(events.CheckEvent(123 self.cursor.location, self.cursor.step, _get_thread_id(), description, is_successful, details124 ))125 def log_url(self, url, description):126 self._flush_pending_events()127 self.event_manager.fire(128 events.LogUrlEvent(self.cursor.location, self.cursor.step, _get_thread_id(), url, description)129 )130 @contextmanager131 def prepare_attachment(self, filename, description, as_image=False):132 with self._attachment_lock:133 attachment_filename = "%04d_%s" % (self._attachment_count + 1, filename)134 self._attachment_count += 1135 if not os.path.exists(self._attachments_dir):136 os.mkdir(self._attachments_dir)137 yield os.path.join(self._attachments_dir, attachment_filename)138 self._flush_pending_events()139 self.event_manager.fire(events.LogAttachmentEvent(140 self.cursor.location, self.cursor.step, _get_thread_id(),141 "%s/%s" % (_ATTACHMENTS_DIR, attachment_filename), description, as_image142 ))143 def start_test_session(self):144 self.event_manager.fire(events.TestSessionStartEvent(self.report))145 def end_test_session(self):146 self.event_manager.fire(events.TestSessionEndEvent(self.report))147 def start_test_session_setup(self):148 self.cursor = _Cursor(ReportLocation.in_test_session_setup())149 self._hold_event(events.TestSessionSetupStartEvent())150 def end_test_session_setup(self):151 self._end_step_if_any()152 self._discard_or_fire_event(events.TestSessionSetupStartEvent, events.TestSessionSetupEndEvent())153 def start_test_session_teardown(self):154 self.cursor = _Cursor(ReportLocation.in_test_session_teardown())155 self._hold_event(events.TestSessionTeardownStartEvent())156 def end_test_session_teardown(self):157 self._end_step_if_any()158 self._discard_or_fire_event(events.TestSessionTeardownStartEvent, events.TestSessionTeardownEndEvent())159 def start_suite(self, suite):160 self.event_manager.fire(events.SuiteStartEvent(suite))161 def end_suite(self, suite):162 self.event_manager.fire(events.SuiteEndEvent(suite))163 def start_suite_setup(self, suite):164 self.cursor = _Cursor(ReportLocation.in_suite_setup(suite))165 self._hold_event(events.SuiteSetupStartEvent(suite))166 def end_suite_setup(self, suite):167 self._end_step_if_any()168 self._discard_or_fire_event(events.SuiteSetupStartEvent, events.SuiteSetupEndEvent(suite))169 def start_suite_teardown(self, suite):170 self.cursor = _Cursor(ReportLocation.in_suite_teardown(suite))171 self._hold_event(events.SuiteTeardownStartEvent(suite))172 def end_suite_teardown(self, suite):173 self._end_step_if_any()174 self._discard_or_fire_event(events.SuiteTeardownStartEvent, events.SuiteTeardownEndEvent(suite))175 def start_test(self, test):176 self.event_manager.fire(events.TestStartEvent(test))177 self.cursor = _Cursor(ReportLocation.in_test(test))178 def end_test(self, test):179 self._end_step_if_any()180 self.event_manager.fire(events.TestEndEvent(test))181 def skip_test(self, test, reason):182 self.event_manager.fire(events.TestSkippedEvent(test, reason))183 self._mark_location_as_failed(ReportLocation.in_test(test))184 def disable_test(self, test, reason):185 self.event_manager.fire(events.TestDisabledEvent(test, reason))...
buttons.py
Source:buttons.py
...29 self.bind(30 on_press=self._set_hold_event,31 on_release=self._unset_hold_event,32 )33 def _set_hold_event(self, *args):34 self._hold_event = Clock.schedule_once(35 lambda event: self.reset(vibrate=True),36 0.537 )38 def _unset_hold_event(self, *args):39 Clock.unschedule(self._hold_event)40 def _init_time(self, complete=True):41 # Warmup isn't typically used between rounds.42 if complete:43 self.warmup_time = self.config.warmup_duration44 self.round_number = 145 self.on_round_number(None, 1)46 self.recovery_time = self.config.recovery_duration47 # Round time comes last so its duration is displayed on the clock.48 self.round_time = self.config.round_duration49 def _start_ticking(self):50 self._update_ticker = Clock.schedule_interval(51 lambda ev: self.update(), 152 )...
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 Lemoncheesecake 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?
