KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use assert_events method in localstack
Best Python code snippet using localstack_python
LaunchTest.py
Source:LaunchTest.py 
...37 self.evlist.clear()38 def result(self, compname, action, status):39 evname = '%s_%s_%s' % (compname, action, status)40 return self.evlist[evname]['result']41 def assert_events(self, compname, action, status_list):42 self.assertEqual(len(self.eh.evlist), len(status_list))43 for status in status_list:44 evname = '%s_%s_%s' % (compname, action, status)45 evlist = ', '.join(self.eh.evlist)46 self.assertTrue(evname in self.eh.evlist,47 "'%s' not in event list: %s" % (evname, evlist))48 def setUp(self):49 self.eh = self.ActionEH()50class ActionsTest(CommonTestCase):51 def format(self):52 self.tgt.format().launch()53 self.fs._run_actions()54 self.eh.clear()55 def setUp(self):56 self.eh = self.ActionEH()57 self.fs = FileSystem('action', event_handler=self.eh)58 srv1 = Server("localhost", ["localhost@tcp"])59 self.disk = Utils.make_disk()60 self.tgt = self.fs.new_target(srv1, 'mgt', 0, self.disk.name)61 def tearDown(self):62 self.tgt.lustre_check()63 if self.tgt.is_started():64 self.tgt.stop().launch()65 self.fs._run_actions()66 #67 # Format68 #69 @Utils.rootonly70 def test_format_ok(self):71 """Format a simple MGT"""72 act = self.tgt.format()73 act.launch()74 self.fs._run_actions()75 # Callback checks76 self.assert_events('mgt', 'format', ['start', 'done'])77 result = self.eh.result('mgt', 'format', 'done')78 self.assertEqual(result.retcode, 0)79 # Status checks80 self.assertEqual(self.tgt.state, OFFLINE)81 self.assertEqual(act.status(), ACT_OK)82 @Utils.rootonly83 def test_format_if_started(self):84 """Format when target is started failed"""85 def check_set_online(self):86 self.__class__.lustre_check(self)87 self.state = MOUNTED88 self.tgt.lustre_check = types.MethodType(check_set_online, self.tgt)89 act = self.tgt.format()90 act.launch()91 self.fs._run_actions()92 # Callback checks93 self.assert_events('mgt', 'format', ['start', 'failed'])94 result = self.eh.result('mgt', 'format', 'failed')95 self.assertEqual(result.retcode, None)96 # Status checks97 self.assertEqual(self.tgt.state, TARGET_ERROR)98 self.assertEqual(act.status(), ACT_ERROR)99 @Utils.rootonly100 def test_format_with_error(self):101 """Format with bad options is an error"""102 act = self.tgt.format(addopts="--BAD-OPTIONS")103 act.launch()104 self.fs._run_actions()105 # Callback checks106 self.assert_events('mgt', 'format', ['start', 'failed'])107 result = self.eh.result('mgt', 'format', 'failed')108 self.assertEqual(result.retcode, 22)109 # Status checks110 self.assertEqual(self.tgt.state, OFFLINE)111 self.assertEqual(act.status(), ACT_ERROR)112 # XXX: Add tests with Journal113 #114 # Fsck115 #116 @Utils.rootonly117 def test_fsck_ok(self):118 """Fsck on a freshly formatted target is ok"""119 self.format()120 act = self.tgt.fsck()121 act.launch()122 self.fs._run_actions()123 # Callback checks124 self.assert_events('mgt', 'fsck', ['start', 'progress', 'done'])125 result = self.eh.result('mgt', 'fsck', 'done')126 self.assertEqual(result.retcode, 0)127 # Status checks128 self.assertEqual(self.tgt.state, OFFLINE)129 self.assertEqual(act.status(), ACT_OK)130 @Utils.rootonly131 def test_fsck_repairs(self):132 """Fsck repairs a corruption"""133 self.format()134 # Corrupt FS135 self.disk.seek(1024)136 self.disk.write('\0' * 1024)137 self.disk.flush()138 act = self.tgt.fsck()139 act.launch()140 self.fs._run_actions()141 # Callback checks142 self.assert_events('mgt', 'fsck', ['start', 'progress', 'done'])143 result = self.eh.result('mgt', 'fsck', 'done')144 self.assertEqual(result.message, "Errors corrected")145 self.assertEqual(result.retcode, 1)146 # Status checks147 self.assertEqual(self.tgt.state, OFFLINE)148 self.assertEqual(act.status(), ACT_OK)149 @Utils.rootonly150 def test_fsck_no_repair(self):151 """Fsck detects but does not repair a corruption with -n"""152 self.format()153 # Corrupt FS154 self.disk.seek(1024)155 self.disk.write('\0' * 1024)156 self.disk.flush()157 act = self.tgt.fsck(addopts='-n')158 act.launch()159 self.fs._run_actions()160 # Callback checks161 self.assert_events('mgt', 'fsck', ['start', 'progress', 'done'])162 result = self.eh.result('mgt', 'fsck', 'done')163 self.assertEqual(result.message, "Errors found but NOT corrected")164 self.assertEqual(result.retcode, 4)165 # Status checks166 self.assertEqual(self.tgt.state, OFFLINE)167 self.assertEqual(act.status(), ACT_OK)168 @Utils.rootonly169 def test_fsck_with_error(self):170 """Fsck on an unformated device fails"""171 act = self.tgt.fsck()172 act.launch()173 self.fs._run_actions()174 # Callback checks175 self.assert_events('mgt', 'fsck', ['start', 'failed'])176 result = self.eh.result('mgt', 'fsck', 'failed')177 self.assertEqual(result.retcode, 8)178 # Status checks179 self.assertEqual(self.tgt.state, OFFLINE)180 self.assertEqual(act.status(), ACT_ERROR)181 #182 # Execute183 #184 def test_execute_ok(self):185 """Execute of a simple command is ok"""186 act = self.tgt.execute(addopts='/bin/echo %device', mountdata='never')187 act.launch()188 self.fs._run_actions()189 # Callback checks190 self.assert_events('mgt', 'execute', ['start', 'done'])191 result = self.eh.result('mgt', 'execute', 'done')192 self.assertEqual(result.retcode, 0)193 # Status checks194 self.assertEqual(self.tgt.state, OFFLINE)195 self.assertEqual(act.status(), ACT_OK)196 def test_execute_error(self):197 """Execute a bad command fails"""198 act = self.tgt.execute(addopts='/bin/false', mountdata='never')199 act.launch()200 self.fs._run_actions()201 # Callback checks202 self.assert_events('mgt', 'execute', ['start', 'failed'])203 result = self.eh.result('mgt', 'execute', 'failed')204 self.assertEqual(result.retcode, 1)205 # Status checks206 self.assertEqual(self.tgt.state, OFFLINE)207 self.assertEqual(act.status(), ACT_ERROR)208 @Utils.rootonly209 def test_execute_check_mountdata(self):210 """Execute a command with mountdata check"""211 self.format()212 act = self.tgt.execute(addopts="ls %device", mountdata='always')213 act.launch()214 self.fs._run_actions()215 # Callback checks216 self.assert_events('mgt', 'execute', ['start', 'done'])217 result = self.eh.result('mgt', 'execute', 'done')218 self.assertEqual(result.retcode, 0)219 # Status checks220 self.assertEqual(self.tgt.state, OFFLINE)221 #222 # Status223 #224 @Utils.rootonly225 def test_status_ok(self):226 """Status on a simple target"""227 self.format()228 act = self.tgt.status()229 act.launch()230 self.fs._run_actions()231 # Callback checks232 self.assert_events('mgt', 'status', ['start', 'done'])233 result = self.eh.result('mgt', 'status', 'done')234 self.assertEqual(result, None)235 # Status checks236 self.assertEqual(self.tgt.state, OFFLINE)237 self.assertEqual(act.status(), ACT_OK)238 def test_status_error(self):239 """Status on a not-formated target fails"""240 act = self.tgt.status()241 act.launch()242 self.fs._run_actions()243 # Callback checks244 self.assert_events('mgt', 'status', ['start', 'failed'])245 result = self.eh.result('mgt', 'status', 'failed')246 self.assertEqual(result.retcode, None)247 # Status checks248 self.assertEqual(self.tgt.state, TARGET_ERROR)249 # XXX: Should we set an error even if job was done correctly?250 self.assertEqual(act.status(), ACT_ERROR)251 #252 # Start Target253 #254 @Utils.rootonly255 def test_start_ok(self):256 """Start a simple target"""257 self.format()258 act = self.tgt.start()259 act.launch()260 self.fs._run_actions()261 # Callback checks262 self.assert_events('mgt', 'start', ['start', 'done'])263 result = self.eh.result('mgt', 'start', 'done')264 self.assertEqual(result.retcode, 0)265 # Status checks266 self.assertEqual(self.tgt.state, MOUNTED)267 self.assertEqual(act.status(), ACT_OK)268 @Utils.rootonly269 def test_start_already_done(self):270 """Start an already mounted target returns ok"""271 self.format()272 def is_started(self):273 self.state = MOUNTED274 return True275 self.tgt.is_started = types.MethodType(is_started, self.tgt)276 act = self.tgt.start()277 act.launch()278 self.fs._run_actions()279 # Callback checks280 self.assert_events('mgt', 'start', ['start', 'done'])281 result = self.eh.result('mgt', 'start', 'done')282 self.assertEqual(result.message, "MGS is already started")283 self.assertEqual(result.retcode, None)284 # Status checks285 self.assertEqual(self.tgt.state, MOUNTED)286 self.assertEqual(act.status(), ACT_OK)287 @Utils.rootonly288 def test_start_error(self):289 """Start an non-formated target fails"""290 act = self.tgt.start()291 act.launch()292 self.fs._run_actions()293 # Callback checks294 self.assert_events('mgt', 'start', ['start', 'failed'])295 result = self.eh.result('mgt', 'start', 'failed')296 self.assertEqual(result.retcode, None)297 # Status checks298 self.assertEqual(self.tgt.state, TARGET_ERROR)299 self.assertEqual(act.status(), ACT_ERROR)300 #301 # Stop Target302 #303 @Utils.rootonly304 def test_stop_ok(self):305 """Stop a simple target"""306 self.format()307 # Start the target, to be able to unmount it after308 act = self.tgt.start()309 act.launch()310 self.fs._run_actions()311 self.eh.clear()312 act = self.tgt.stop()313 act.launch()314 self.fs._run_actions()315 # Callback checks316 self.assert_events('mgt', 'stop', ['start', 'done'])317 result = self.eh.result('mgt', 'stop', 'done')318 self.assertEqual(result.retcode, 0)319 # Status checks320 self.assertEqual(self.tgt.state, OFFLINE)321 self.assertEqual(act.status(), ACT_OK)322 @Utils.rootonly323 def test_stop_already_done(self):324 """Stop an already stopped target fails"""325 self.format()326 act = self.tgt.stop()327 act.launch()328 self.fs._run_actions()329 # Callback checks330 self.assert_events('mgt', 'stop', ['start', 'done'])331 result = self.eh.result('mgt', 'stop', 'done')332 self.assertEqual(result.message, "MGS is already stopped")333 self.assertEqual(result.retcode, None)334 # Status checks335 self.assertEqual(self.tgt.state, OFFLINE)336 self.assertEqual(act.status(), ACT_OK)337 @Utils.rootonly338 def test_stop_error(self):339 """Stop target failure is an error"""340 self.format()341 # Start the target, to be able to unmount it after342 self.tgt.start().launch()343 self.fs._run_actions()344 self.eh.clear()345 act = self.tgt.stop()346 def _prepare_cmd(_self):347 return [ "/bin/false" ]348 act._prepare_cmd = types.MethodType(_prepare_cmd, act)349 act.launch()350 self.fs._run_actions()351 # Callback checks352 self.assert_events('mgt', 'stop', ['start', 'failed'])353 result = self.eh.result('mgt', 'stop', 'failed')354 self.assertEqual(result.retcode, 1)355 # Status checks356 self.assertEqual(self.tgt.state, MOUNTED)357 self.assertEqual(act.status(), ACT_ERROR)358 #359 # Server test (special)360 # This test needs a running MGS, it is better to put it here.361 #362 @Utils.rootonly363 def test_module_busy(self):364 """Unloading module with a started MGS is not considered as an error"""365 self.format()366 # Start the target, to set module busy367 self.tgt.start().launch()368 self.fs._run_actions()369 srv = self.tgt.server370 act = srv.unload_modules()371 act.launch()372 self.fs._run_actions()373 # Status check374 self.assertTrue(set(['ldiskfs', 'libcfs', 'lustre']).issubset(375 set(srv.modules.keys())))376 self.assertEqual(act.status(), ACT_OK)377class RouterActionTest(CommonTestCase):378 def net_up(self, options):379 self.srv1.load_modules(modname='lnet', options=options).launch()380 self.fs._run_actions()381 self.eh.clear()382 def setUp(self):383 self.eh = self.ActionEH()384 self.fs = FileSystem('action', event_handler=self.eh)385 self.srv1 = Server("localhost", ["localhost@tcp"])386 self.router = self.fs.new_router(self.srv1)387 self.srv1.unload_modules().launch()388 self.fs._run_actions()389 def tearDown(self):390 self.router.lustre_check()391 if self.router.is_started():392 self.router.stop().launch()393 self.fs._run_actions()394 self.srv1.unload_modules().launch()395 self.fs._run_actions()396 #397 # Start Router398 #399 @Utils.rootonly400 def test_start_router_ok(self):401 """Start a stopped router is ok"""402 self.net_up('forwarding=enabled')403 act = self.router.start()404 act.launch()405 self.fs._run_actions()406 # Callback checks407 self.assert_events('router', 'start', ['start', 'done'])408 result = self.eh.result('router', 'start', 'done')409 self.assertEqual(result.retcode, 0)410 # Status checks411 self.assertEqual(self.router.state, MOUNTED)412 self.assertEqual(act.status(), ACT_OK)413 @Utils.rootonly414 def test_start_router_already_done(self):415 """Start an already started router is ok"""416 self.net_up('forwarding=enabled')417 # Start the router418 self.router.start().launch()419 self.fs._run_actions()420 self.eh.clear()421 # Then try to restart it422 act = self.router.start()423 act.launch()424 self.fs._run_actions()425 # Callback checks426 self.assert_events('router', 'start', ['start', 'done'])427 result = self.eh.result('router', 'start', 'done')428 self.assertEqual(result.message, "router is already enabled")429 self.assertEqual(result.retcode, None)430 # Status checks431 self.assertEqual(self.router.state, MOUNTED)432 self.assertEqual(act.status(), ACT_OK)433 #434 # Stop Router435 #436 @Utils.rootonly437 def test_stop_router_ok(self):438 """Stop a started router is ok"""439 self.net_up('forwarding=enabled')440 # Start the router441 self.router.start().launch()442 self.fs._run_actions()443 self.eh.clear()444 # Then stop it445 act = self.router.stop()446 act.launch()447 self.fs._run_actions()448 # Callback checks449 self.assert_events('router', 'stop', ['start', 'done'])450 result = self.eh.result('router', 'stop', 'done')451 self.assertEqual(result.retcode, 0)452 # Status checks453 self.assertEqual(self.router.state, OFFLINE)454 self.assertEqual(act.status(), ACT_OK)455 @Utils.rootonly456 def test_stop_router_already_done(self):457 """Stop an already stopped router fails"""458 self.net_up('forwarding=enabled')459 act = self.router.stop()460 act.launch()461 self.fs._run_actions()462 # Callback checks463 self.assert_events('router', 'stop', ['start', 'done'])464 result = self.eh.result('router', 'stop', 'done')465 self.assertEqual(result.message, "router is already disabled")466 self.assertEqual(result.retcode, None)467 # Status checks468 self.assertEqual(self.router.state, OFFLINE)469 self.assertEqual(act.status(), ACT_OK)470class ServerActionTest(unittest.TestCase):471 def _clean_modules(self):472 """Remove all already loaded modules, before a test."""473 self.srv.lustre_check()474 if 'libcfs' in self.srv.modules or 'ldiskfs' in self.srv.modules:475 self.srv.unload_modules().launch()476 self.fs._run_actions()477 def setUp(self):...
duplication_tests.py
Source:duplication_tests.py
...4def test_simple():5 engine = SessionEngine()6 session = Session(rating_key=100, state='create', duration=50 * 1000, view_offset=0, part=1)7 # playing8 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 1000, 'part': 1})], [('start',)])9 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 3000, 'part': 1})], [])10 # paused11 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 3000, 'part': 1})], [('pause',)])12 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 3000, 'part': 1})], [])13 # playing14 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 4000, 'part': 1})], [('start',)])15 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 4000, 'part': 1})], [])16 # stopped17 assert_events(engine, session, [('stopped', {'rating_key': 100, 'view_offset': 5000, 'part': 1})], [('stop',)])18 assert_events(engine, session, [('stopped', {'rating_key': 100, 'view_offset': 5000, 'part': 1})], [])19 # media change20 assert_events(engine, session, [('playing', {'rating_key': 101, 'view_offset': 1000, 'part': 1})], [('start',)])21def test_finished():22 engine = SessionEngine()23 session = Session(rating_key=100, state='create', duration=50 * 1000, view_offset=0, part=1)24 # playing25 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 1000, 'part': 1})], [('start',)])26 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 2000, 'part': 1})], [])27 # stopped28 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 50 * 1000, 'part': 1})], [('stop',)])29 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 51 * 1000, 'part': 1})], [])30 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 52 * 1000, 'part': 1})], [])31 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 53 * 1000, 'part': 1})], [])32 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 54 * 1000, 'part': 1})], [])33def test_stopped():34 engine = SessionEngine()35 session = Session(rating_key=100, state='create', duration=50 * 1000, view_offset=0, part=1)36 # Start watching item37 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 1000, 'part': 1})], [('start',)])38 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 10000, 'part': 1})], [])39 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 20000, 'part': 1})], [])40 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 30000, 'part': 1})], [])41 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 40000, 'part': 1})], [])42 # Ensure "stop" actions aren't duplicated43 assert_events(engine, session, [('stopped', {'rating_key': 100, 'view_offset': 50000, 'part': 1})], [('stop',)])44 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 50000, 'part': 1})], [])45 assert_events(engine, session, [('stopped', {'rating_key': 100, 'view_offset': 50000, 'part': 1})], [])46 # Ensure item can be restarted47 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 10000, 'part': 1})], [('start',)])...
action_tests.py
Source:action_tests.py
...4def test_stopped():5 engine = SessionEngine()6 session = Session(rating_key=100, state='create', duration=50 * 1000, view_offset=0, part=1)7 # playing8 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 1 * 1000, 'part': 1})], [('start',)])9 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 3 * 1000, 'part': 1})], [])10 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 5 * 1000, 'part': 1})], [])11 # finished12 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 50 * 1000, 'part': 1})], [('stop',)])13 # stopped14 assert_events(engine, session, [('stopped', {'rating_key': 100, 'view_offset': 50 * 1000, 'part': 1})], [])15def test_paused():16 engine = SessionEngine()17 session = Session(rating_key=100, state='create', duration=50 * 1000, view_offset=0, part=1)18 # playing19 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 1 * 1000, 'part': 1})], [('start',)])20 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 3 * 1000, 'part': 1})], [])21 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 5 * 1000, 'part': 1})], [])22 # paused23 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 37 * 1000, 'part': 1})], [('pause',)])24 # playing25 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 38 * 1000, 'part': 1})], [('start',)])26 assert_events(engine, session, [('playing', {'rating_key': 100, 'view_offset': 39 * 1000, 'part': 1})], [])27 # paused28 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 40 * 1000, 'part': 1})], [('pause',)])29 # finished30 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 49 * 1000, 'part': 1})], [('stop',)])31 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 50 * 1000, 'part': 1})], [])32 # stopped33 assert_events(engine, session, [('stopped', {'rating_key': 100, 'view_offset': 50 * 1000, 'part': 1})], [])34def test_unplayed():35 engine = SessionEngine()36 session = Session(rating_key=100, state='create', duration=50 * 1000, view_offset=0, part=1)37 assert_events(engine, session, [('paused', {'rating_key': 100, 'view_offset': 50000, 'part': 1})], [])...
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?
