KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _get_state method in localstack
Best Python code snippet using localstack_python
test_process.py
Source:test_process.py 
...38 self.assertIsInstance(p1.id, Id)39 self.assertTrue(p1.receiver)40 self.assertFalse(p1.receiver.consumer)41 self.assertEquals(p1.receiver.consumer, None)42 self.assertEquals(p1._get_state(), "INIT")43 self.assertEquals(p1.spawn_args, {})44 self.assertTrue(p1.proc_init_time)45 self.assertTrue(p1.proc_name)46 self.assertTrue(p1.sys_name)47 self.assertTrue(p1.proc_group)48 self.assertTrue(p1.backend_id)49 self.assertTrue(p1.backend_receiver)50 self.assertEquals(len(p1.receivers), 2)51 self.assertEquals(p1.conversations, {})52 self.assertEquals(p1.child_procs, [])53 pid1 = yield p1.spawn()54 self.assertEquals(pid1, p1.id)55 self.assertEquals(p1._get_state(), "ACTIVE")56 procid = Id('local','container')57 args = {'proc-id':procid.full}58 p2 = Process(spawnargs=args)59 self.assertEquals(p2.id, procid)60 yield p2.initialize()61 self.assertEquals(p2._get_state(), "READY")62 yield p2.activate()63 self.assertEquals(p2._get_state(), "ACTIVE")64 args = {'arg1':'value1','arg2':{}}65 p3 = Process(None, args)66 self.assertEquals(p3.spawn_args, args)67 @defer.inlineCallbacks68 def test_process(self):69 # Also test the ReceiverProcess helper class70 log.debug('Spawning p1')71 p1 = ReceiverProcess(spawnargs={'proc-name':'p1'})72 pid1 = yield p1.spawn()73 log.debug('Spawning other processes')74 processes = [75 {'name':'echo','module':'ion.core.process.test.test_process','class':'EchoProcess'},76 ]77 sup = yield self._spawn_processes(processes, sup=p1)78 assert sup == p179 pid2 = p1.get_child_id('echo')80 proc2 = self._get_procinstance(pid2)81 yield p1.send(pid2, 'echo','content123')82 log.info('Sent echo message')83 msg = yield p1.await_message()84 log.info('Received echo message')85 self.assertEquals(msg.payload['op'], 'result')86 #self.assertEquals(msg.payload['content']['value'], 'content123')87 self.assertEquals(msg.payload['content'], 'content123')88 yield sup.terminate()89 self.assertEquals(sup._get_state(), "TERMINATED")90 self.assertEquals(proc2._get_state(), "TERMINATED")91 @defer.inlineCallbacks92 def test_child_processes(self):93 p1 = Process()94 pid1 = yield p1.spawn()95 child = ProcessDesc(name='echo', module='ion.core.process.test.test_process')96 pid2 = yield p1.spawn_child(child)97 (cont,hdrs,msg) = yield p1.rpc_send(pid2,'echo','content123')98 #self.assertEquals(cont['value'], 'content123')99 self.assertEqual(hdrs.get(p1.MSG_STATUS),'OK')100 self.assertEquals(cont, 'content123')101 yield p1.terminate()102 self.assertEquals(p1._get_state(), "TERMINATED")103 @defer.inlineCallbacks104 def test_spawn_child(self):105 child1 = ProcessDesc(name='echo', module='ion.core.process.test.test_process')106 self.assertEquals(child1._get_state(),'INIT')107 pid1 = yield self.test_sup.spawn_child(child1)108 self.assertEquals(child1._get_state(),'ACTIVE')109 proc = self._get_procinstance(pid1)110 self.assertEquals(str(proc.__class__),"<class 'ion.core.process.test.test_process.EchoProcess'>")111 self.assertEquals(pid1, proc.id)112 log.info('Process 1 spawned and initd correctly')113 (cont,hdrs,msg) = yield self.test_sup.rpc_send(pid1,'echo','content123')114 #self.assertEquals(cont['value'], 'content123')115 self.assertEquals(cont, 'content123')116 log.info('Process 1 responsive correctly')117 # The following tests the process attaching a second receiver118 msgName = pu.create_guid()119 extraRec = WorkerReceiver(label=proc.proc_name, name=msgName, handler=proc.receive)120 extraid = yield extraRec.attach()121 log.info('Created new receiver %s' % (msgName))122 (cont,hdrs,msg) = yield self.test_sup.rpc_send(msgName,'echo','content456')123 #self.assertEquals(cont['value'], 'content456')124 self.assertEquals(cont, 'content456')125 log.info('Process 1 responsive correctly on second receiver')126 @defer.inlineCallbacks127 def test_message_before_activate(self):128 p1 = ReceiverProcess(spawnargs={'proc-name':'p1'})129 pid1 = yield p1.spawn()130 proc1 = self._get_procinstance(pid1)131 child2 = ProcessDesc(name='echo', module='ion.core.process.test.test_process')132 pid2 = yield self.test_sup.spawn_child(child2, activate=False)133 self.assertEquals(child2._get_state(), 'READY')134 proc2 = self._get_procinstance(pid2)135 self.assertEquals(proc2._get_state(), 'READY')136 # The following tests that a message to a not yet activated process137 # is queued and not lost, but not delivered138 yield proc1.send(pid2,'echo','content123')139 self.assertEquals(proc1.inbox_count, 0)140 yield pu.asleep(1)141 self.assertEquals(proc1.inbox_count, 0)142 yield child2.activate()143 yield pu.asleep(1)144 self.assertEquals(child2._get_state(), 'ACTIVE')145 self.assertEquals(proc1.inbox_count, 1)146 (cont,hdrs,msg) = yield self.test_sup.rpc_send(pid2,'echo','content123')147 #self.assertEquals(cont['value'], 'content123')148 self.assertEquals(cont, 'content123')149 log.info('Process 1 responsive correctly after init')150 @defer.inlineCallbacks151 def test_error_in_op(self):152 child1 = ProcessDesc(name='echo', module='ion.core.process.test.test_process')153 pid1 = yield self.test_sup.spawn_child(child1)154 try:155 (cont,hdrs,msg) = yield self.test_sup.rpc_send(pid1,'echo_exception','content123')156 self.fail("ReceivedError expected")157 except ReceivedError, re:158 log.info('Process 1 responded to error correctly')159 @defer.inlineCallbacks160 def test_send_byte_string(self):161 """162 @brief Test that any arbitrary byte string can be sent through the163 ion CC stack. Use a 20 byte sha1 digest as test string.164 """165 p1 = ReceiverProcess()166 pid1 = yield p1.spawn()167 processes = [168 {'name':'echo','module':'ion.core.process.test.test_process','class':'EchoProcess'},169 ]170 sup = yield self._spawn_processes(processes, sup=p1)171 pid2 = p1.get_child_id('echo')172 173 byte_string = hashlib.sha1('test').digest()174 yield p1.send(pid2, 'echo', byte_string)175 log.info('Sent byte-string')176 msg = yield p1.await_message()177 log.info('Received byte-string')178 #self.assertEquals(msg.payload['content']['value'], byte_string)179 self.assertEquals(msg.payload['content'], byte_string)180 yield sup.shutdown()181 @defer.inlineCallbacks182 def test_shutdown(self):183 processes = [184 {'name':'echo1','module':'ion.core.process.test.test_process','class':'EchoProcess'},185 {'name':'echo2','module':'ion.core.process.test.test_process','class':'EchoProcess'},186 {'name':'echo3','module':'ion.core.process.test.test_process','class':'EchoProcess'},187 ]188 sup = yield self._spawn_processes(processes)189 yield self._shutdown_processes()190 @defer.inlineCallbacks191 def test_rpc_timeout(self):192 sup = self.test_sup193 try:194 yield sup.rpc_send('big_void', 'noop', 'arbitrary', timeout=1)195 self.fail("TimeoutError expected")196 except defer.TimeoutError, te:197 log.info('Timeout received')198 @defer.inlineCallbacks199 def test_register_lco(self):200 """201 Test the registration of life cycle objects owned by a process202 Do not spawn the process - want to manually move it through the FSM!203 """204 # Create a process which has an lco object in its init205 lco1 = life_cycle_process.LifeCycleObject()206 lcop = life_cycle_process.LCOProcess(lco1, spawnargs={'proc-name':'p1'}) 207 self.assertEquals(lcop._get_state(), state_object.BasicStates.S_INIT)208 self.assertEquals(lco1._get_state(), state_object.BasicStates.S_INIT)209 210 lco2 = life_cycle_process.LifeCycleObject()211 yield lcop.register_life_cycle_object(lco2)212 self.assertEquals(lco2._get_state(), state_object.BasicStates.S_INIT)213 214 # Initialize the process and its objects215 yield lcop.initialize()216 self.assertEquals(lcop._get_state(), state_object.BasicStates.S_READY)217 self.assertEquals(lco1._get_state(), state_object.BasicStates.S_READY)218 self.assertEquals(lco2._get_state(), state_object.BasicStates.S_READY)219 220 lco3 = life_cycle_process.LifeCycleObject()221 yield lcop.register_life_cycle_object(lco3)222 self.assertEquals(lco3._get_state(), state_object.BasicStates.S_READY)223 224 # Check that using add after init causes an error225 lcoa = life_cycle_process.LifeCycleObject()226 self.assertRaises(ProcessError,lcop.add_life_cycle_object,lcoa)227 228 # Activate the process and its objects229 yield lcop.activate()230 231 self.assertEquals(lcop._get_state(), state_object.BasicStates.S_ACTIVE)232 self.assertEquals(lco1._get_state(), state_object.BasicStates.S_ACTIVE)233 self.assertEquals(lco2._get_state(), state_object.BasicStates.S_ACTIVE)234 self.assertEquals(lco3._get_state(), state_object.BasicStates.S_ACTIVE)235 236 lco4 = life_cycle_process.LifeCycleObject()237 yield lcop.register_life_cycle_object(lco4)238 self.assertEquals(lco4._get_state(), state_object.BasicStates.S_ACTIVE)239 # Process does not currently implement for deactivate!240 241 # Terminate the process and its objects242 yield lcop.terminate()243 self.assertEquals(lcop._get_state(), state_object.BasicStates.S_TERMINATED)244 self.assertEquals(lco1._get_state(), state_object.BasicStates.S_TERMINATED)245 self.assertEquals(lco2._get_state(), state_object.BasicStates.S_TERMINATED)246 self.assertEquals(lco3._get_state(), state_object.BasicStates.S_TERMINATED)247 self.assertEquals(lco4._get_state(), state_object.BasicStates.S_TERMINATED)248 249 # Can't seem to assert raises - not sure why not?250 #lco5 = life_cycle_process.LifeCycleObject()251 #self.assertRaises(ProcessError,lcop.register_life_cycle_object,lco5)252 #yield lcop.register_life_cycle_object(lco5)253 254class EchoProcess(Process):255 256 @defer.inlineCallbacks257 def op_echo(self, content, headers, msg):258 log.info("Message received: "+str(content))259 yield self.reply(msg, content=content)260 @defer.inlineCallbacks261 def op_echo_fail(self, content, headers, msg):...
apa102_api.py
Source:apa102_api.py
...45class StateControl(Resource):46 """47 Flask-RESTFul Resource defining APA102 state API.48 """49 def _get_state(self):50 """51 Helper method to return current APA102 state.52 """53 return {54 "contrast": apa102.contrast,55 "speed": apa102.animation_speed,56 "colors" : list(filter(None, apa102.color_buffer)), # Just colours. Empty color elements not returned.57 "animation": mode_to_text[apa102.mode]58 }59 def get(self):60 """61 GET Request returns current APA102 state.62 """63 return {64 "success": True,65 "state": self._get_state()66 }67class ColorControl(Resource):68 """69 Flask-RESTFul Resource defining APA102 color control API.70 """71 def __init__(self):72 """73 Constructor - Setup allowed arguments.74 """75 self.args_parser = reqparse.RequestParser(trim=True)76 self.args_parser.add_argument(name='colors', type=str, required=True, store_missing=False, help='Comma separated list of colors')77 self.args_parser.add_argument(name='pattern', type=str, required=False, default="n", choices=("y", "n", "yes", "no"), case_sensitive=False, store_missing=True, help='Apply colors as a repeating pattern')78 def _get_state(self):79 """80 Helper method to return current APA102 color buffer.81 """82 return {83 "colors" : list(filter(None, apa102.color_buffer)), # Just colours. Empty color elements not returned.84 }85 def get(self):86 """87 GET Request returns current APA102 color buffer.88 """89 return {90 "success": True,91 "state": self._get_state()92 }93 def post(self):94 """95 POST Request to set/update APA102 color buffer.96 """97 args = self.args_parser.parse_args()98 pattern = args['pattern'][0] == "y"99 colors = args['colors'].split(",")100 if pattern:101 apa102.set_pattern(colors)102 else:103 for color in colors:104 apa102.push_color(color)105 return {106 "success": True,107 "state": self._get_state()108 }109class ContrastControl(Resource):110 """111 Flask-RESTFul Resource defining APA102 contrast control API.112 """113 def __init__(self):114 """115 Constructor - Setup argument parser116 """117 self.args_parser = reqparse.RequestParser(trim=True)118 self.args_parser.add_argument(name='level', type=inputs.int_range(0, 255), required=False)119 def _get_state(self):120 """121 Helper method to return current APA102 contrast level.122 """123 return {124 "contrast": apa102.contrast125 }126 def get(self):127 """128 GET Request returns current APA102 contrast level.129 """130 return {131 "success": True,132 "state": self._get_state()133 }134 def post(self):135 """136 POST Request to set APA102 contrast level.137 """138 args = self.args_parser.parse_args()139 if 'level' in args and args['level'] is not None:140 level = int(args['level'])141 apa102.set_contrast(level)142 return {143 "success": True,144 "state": self._get_state()145 }146class AnimationControl(Resource):147 """148 Flask-RESTFul Resource defining APA102 LED animation control API.149 """150 def __init__(self):151 """152 Constructor - Setup argument parser153 """154 self.args_parser = reqparse.RequestParser(trim=True)155 self.args_parser.add_argument(name='mode', type=str, required=False, choices=("stop", "blink", "left", "right", "rainbow"), case_sensitive=False, help='Mode')156 self.args_parser.add_argument(name='speed', type=inputs.int_range(1, 10), required=False)157 def _get_state(self):158 """159 Helper method to return current APA102 contrast level.160 """161 return {162 "speed": apa102.animation_speed,163 "animation": mode_to_text[apa102.mode]164 }165 def get(self):166 """167 GET Request returns current APA102 animation mode.168 """169 return {170 "success": True,171 "state": self._get_state()172 }173 def post(self):174 """175 POST Request sets APA102 animation mode.176 """177 args = self.args_parser.parse_args()178 if 'mode' in args:179 mode = args['mode']180 self._set_mode(mode)181 if 'speed' in args and args['speed'] is not None:182 speed = int(args['speed'])183 apa102.set_animation_speed(speed)184 return {185 "success": True,186 "state": self._get_state()187 }188 def _set_mode(self, mode):189 """190 Helper method to set animation mode.191 """192 if mode == "stop":193 apa102.stop_animation()194 elif mode == "blink":195 apa102.blink()196 elif mode == "left":197 apa102.rotate_left()198 elif mode == "right":199 apa102.rotate_right()200 elif mode == "rainbow":...
buildbot_state_test.py
Source:buildbot_state_test.py
...38 'desired_state': self.desired_buildbot_state,39 'transition_time_utc': self.desired_transition_time,40 }41 return evidence42 def _get_state(self):43 return self.matcher.get_state(self._get_evidence())44 def _get_execution_list(self):45 return self.matcher.execution_list(self._get_evidence())46 def testPatternMatcherIsSane(self):47 self.assertTrue(self.matcher.is_correct)48 #### Tests.49 def testBuildbotIsOffline(self):50 self.buildbot_is_running = False51 state = self._get_state()['buildbot']52 self.assertEqual(state, 'offline')53 def testBuildbotIsStarting(self):54 self.utcnow = self.last_boot + 4 * 6055 self.accepting_builds = None56 state = self._get_state()['buildbot']57 self.assertEqual(state, 'starting')58 def testBuildbotIsRunning(self):59 state = self._get_state()['buildbot']60 self.assertEqual(state, 'running')61 def testBuildbotIsDraining(self):62 self.accepting_builds = False63 self.utcnow = self.last_no_new_builds + 4 * 6064 state = self._get_state()['buildbot']65 self.assertEqual(state, 'draining')66 def testBuildbotIsDrainedNoBuilds(self):67 self.accepting_builds = False68 self.current_running_builds = 069 self.utcnow = self.last_no_new_builds + 4 * 6070 state = self._get_state()['buildbot']71 self.assertEqual(state, 'drained')72 def testBuildbotIsDrainedTimeout(self):73 self.accepting_builds = False74 state = self._get_state()['buildbot']75 self.assertEqual(state, 'drained')76 def testBuildbotIsCrashing(self):77 self.accepting_builds = None78 state = self._get_state()['buildbot']79 self.assertEqual(state, 'crashed')80 def testDesiredOffline(self):81 self.desired_buildbot_state = 'offline'82 state = self._get_state()['desired_buildbot_state']83 self.assertEqual(state, 'offline')84 def testDesiredInvalid(self):85 self.desired_buildbot_state = 'delicious'86 with self.assertRaises(ValueError):87 _ = self._get_state()88 def testDesiredUpToDate(self):89 state = self._get_state()['desired_transition_time']90 self.assertEqual(state, 'hold_steady')91 def testDesiredFuture(self):92 self.desired_transition_time = 300093 with self.assertRaises(ValueError):94 _ = self._get_state()95 def testDesiredReboot(self):96 self.desired_transition_time = 110097 state = self._get_state()['desired_transition_time']98 self.assertEqual(state, 'ready_to_fire')99 def testNoLastBoot(self):100 self.last_boot = None101 state = self._get_state()['desired_transition_time']102 self.assertEqual(state, 'ready_to_fire')103 def testOfflineStaysOffline(self):104 self.desired_buildbot_state = 'offline'105 self.buildbot_is_running = False106 _, _, execution_list = self._get_execution_list()107 self.assertEqual(execution_list, [])108 def testRestartKickedOff(self):109 self.desired_transition_time = 1100110 _, _, execution_list = self._get_execution_list()111 self.assertEqual(execution_list, [master.MakeNoNewBuilds])112 def testTurnDown(self):113 self.desired_buildbot_state = 'offline'114 self.accepting_builds = False115 _, _, execution_list = self._get_execution_list()...
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 localstack 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?
