KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use check_write method in autotest

Best Python code snippet using autotest_python

test_elf_multithread.py

Source:test_elf_multithread.py

...21        self.assertEqual("child", ql.loader.argv[0])22        del QL_TEST23        del ql24    def test_multithread_elf_linux_x86(self):25        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):26            nonlocal buf_out27            try:28                buf = ql.mem.read(write_buf, write_count)29                buf = buf.decode()30                buf_out = buf31            except:32                pass33        buf_out = None34        ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_multithreading"], "../examples/rootfs/x86_linux", multithread=True, verbose=QL_VERBOSE.DEBUG)35        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)36        ql.run()37        self.assertTrue("thread 2 ret val is" in buf_out)38        del ql39    def test_multithread_elf_linux_arm64(self):40        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):41            nonlocal buf_out42            try:43                buf = ql.mem.read(write_buf, write_count)44                buf = buf.decode()45                buf_out = buf46            except:47                pass48        buf_out = None49        ql = Qiling(["../examples/rootfs/arm64_linux/bin/arm64_multithreading"], "../examples/rootfs/arm64_linux", multithread=True, verbose=QL_VERBOSE.DEBUG)50        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)51        ql.run()52        self.assertTrue("thread 2 ret val is" in buf_out)53        del ql54    def test_multithread_elf_linux_x8664(self):55        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):56            nonlocal buf_out57            try:58                buf = ql.mem.read(write_buf, write_count)59                buf = buf.decode()60                buf_out = buf61            except:62                pass63        buf_out = None64        ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_multithreading"], "../examples/rootfs/x8664_linux", multithread=True, profile= "profiles/append_test.ql")65        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)66        ql.run()67        self.assertTrue("thread 2 ret val is" in buf_out)68        del ql69    def test_multithread_elf_linux_mips32el(self):70        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):71            nonlocal buf_out72            try:73                buf = ql.mem.read(write_buf, write_count)74                buf = buf.decode()75                buf_out = buf76            except:77                pass78        buf_out = None79        ql = Qiling(["../examples/rootfs/mips32el_linux/bin/mips32el_multithreading"], "../examples/rootfs/mips32el_linux", multithread=True, verbose=QL_VERBOSE.DEBUG)80        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)81        ql.run()82        self.assertTrue("thread 2 ret val is" in buf_out)83        del ql84    def test_multithread_elf_linux_arm(self):85        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):86            nonlocal buf_out87            try:88                buf = ql.mem.read(write_buf, write_count)89                buf = buf.decode()90                buf_out = buf91            except:92                pass93        buf_out = None94        ql = Qiling(["../examples/rootfs/arm_linux/bin/arm_multithreading"], "../examples/rootfs/arm_linux", multithread=True, verbose=QL_VERBOSE.DEBUG)95        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)96        ql.run()97        self.assertTrue("thread 2 ret val is" in buf_out)98        del ql99    def test_tcp_elf_linux_x86(self):100        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):101            try:102                buf = ql.mem.read(write_buf, write_count)103                buf = buf.decode()104                if buf.startswith("server send()"):105                    ql.buf_out = buf106            except:107                pass108        ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_tcp_test","20001"], "../examples/rootfs/x86_linux", multithread=True)109        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)110        ql.run()111        self.assertEqual("server send() 14 return 14.\n", ql.buf_out)112        del ql113    def test_tcp_elf_linux_x8664(self):114        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):115            try:116                buf = ql.mem.read(write_buf, write_count)117                buf = buf.decode()118                if buf.startswith("server send()"):119                    ql.buf_out = buf120            except:121                pass122        ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_tcp_test","20002"], "../examples/rootfs/x8664_linux", multithread=True)123        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)124        ql.run()125        self.assertEqual("server send() 14 return 14.\n", ql.buf_out)126        del ql127    def test_tcp_elf_linux_arm(self):128        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):129            try:130                buf = ql.mem.read(write_buf, write_count)131                buf = buf.decode()132                if buf.startswith("server write()"):133                    ql.buf_out = buf134            except:135                pass136        ql = Qiling(["../examples/rootfs/arm_linux/bin/arm_tcp_test","20003"], "../examples/rootfs/arm_linux", multithread=True)137        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)138        ql.run()139        self.assertEqual("server write() 14 return 14.\n", ql.buf_out)140        del ql141    def test_tcp_elf_linux_arm64(self):142        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):143            try:144                buf = ql.mem.read(write_buf, write_count)145                buf = buf.decode()146                if buf.startswith("server send()"):147                    ql.buf_out = buf148            except:149                pass150        ql = Qiling(["../examples/rootfs/arm64_linux/bin/arm64_tcp_test","20004"], "../examples/rootfs/arm64_linux", multithread=True)151        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)152        ql.run()153        self.assertEqual("server send() 14 return 14.\n", ql.buf_out)154        del ql155    def test_tcp_elf_linux_mips32el(self):156        ql = Qiling(["../examples/rootfs/mips32el_linux/bin/mips32el_tcp_test","20005"], "../examples/rootfs/mips32el_linux", multithread=True)157        ql.run()158        del ql159    def test_udp_elf_linux_x86(self):160        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):161            try:162                buf = ql.mem.read(write_buf, write_count)163                buf = buf.decode()164                if buf.startswith("server sendto()"):165                    ql.buf_out = buf166            except:167                pass168        ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_udp_test","20007"], "../examples/rootfs/x86_linux", multithread=True)169        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)170        ql.run()171        self.assertEqual("server sendto() 14 return 14.\n", ql.buf_out)172        del ql173    def test_udp_elf_linux_x8664(self):174        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):175            try:176                buf = ql.mem.read(write_buf, write_count)177                buf = buf.decode()178                if buf.startswith("server sendto()"):179                    ql.buf_out = buf180            except:181                pass182        ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_udp_test","20008"], "../examples/rootfs/x8664_linux", multithread=True)183        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)184        ql.run()185        self.assertEqual("server sendto() 14 return 14.\n", ql.buf_out)186        del ql187    def test_udp_elf_linux_arm64(self):188        def check_write(ql, write_fd, write_buf, write_count, *args, **kw):189            try:190                buf = ql.mem.read(write_buf, write_count)191                buf = buf.decode()192                if buf.startswith("server sendto()"):193                    ql.buf_out = buf194            except:195                pass196        ql = Qiling(["../examples/rootfs/arm64_linux/bin/arm64_udp_test","20009"], "../examples/rootfs/arm64_linux", multithread=True)197        ql.set_syscall("write", check_write, QL_INTERCEPT.ENTER)198        ql.run()199        self.assertEqual("server sendto() 14 return 14.\n", ql.buf_out)200        del ql201if __name__ == "__main__":202    unittest.main()

Project1.py

Source:Project1.py

1# Hootan Hosseinzadeganbushehri2import sys3import Project1_First4def control_output(my_input):5    sys.stdout = open('output.txt', 'w')6    check_write = 07    nxt_line = False8    read_c = 19    for l_input in my_input:10        if 'in' == l_input[0]:11            current_p = Project1_First.FileProcess()12            if not nxt_line:13                read_c = 014            else:15                check_write = 016                print()17            nxt_line = True18            read_c += 119            print(current_p.source_start.index, end=' ')20        elif 'de' == l_input[0]:21            check_new_setup = False22            for n in current_p.source_start.pos_set:23                if n.index != int(l_input[1]):24                    check_write += 125                else:26                    check_new_setup = True27                    check_write = 028                    break29            if not check_new_setup:30                print('-1', end=' ')31                check_write += 132                current_r = False33            else:34                current_r = current_p.delete_process(int(l_input[1]))35                check_write -= 136            if current_r is None or current_r:37                read_c += 138                print(current_p.source_start.index, end=' ')39        elif '' == l_input[0]:40            check_write = 041            continue42        elif 'to' == l_input[0]:43            current_p.latency_run()44            read_c = 045            print(current_p.source_start.index, end=' ')46        elif 'cr' == l_input[0]:47            current_r = current_p.new_setup(int(l_input[1]))48            if current_r is not None:49                read_c -= 150                check_write -= 151            else:52                check_write += 153                print(current_p.source_start.index, end=' ')54        elif 'rl' == l_input[0]:55            current_r = current_p.process_flow(int(l_input[1]), int(l_input[2]))56            check_write = 057            if current_r is not None:58                check_write -= 159            else:60                check_write += 161                print(current_p.source_start.index, end=' ')62        elif 'rq' == l_input[0]:63            current_r = current_p.develop_process(int(l_input[1]), int(l_input[2]))64            check_write = 065            if current_r is not None:66                check_write -= 167                read_c = 068            else:69                check_write += 170                print(current_p.source_start.index, end=' ')71def new_process():72    lines = []73    if len(sys.argv) > 1:74        in_file = open(sys.argv[1])75        while 1:76            in_line = in_file.readline()77            if in_line == '':78                break79            else:80                lines.append(in_line)81        my_list = [line.strip().split(' ') for line in lines]82        in_file.close()83    control_output(my_list)84if __name__ == '__main__':...

test_utils.py

Source:test_utils.py

...9    write_ushort,10)11__author__ = "Matthieu Gallet"12class Test(TestCase):13    def check_write(self, fn, obj):14        fd = io.BytesIO()15        fn(fd, obj)16        return fd.getvalue()17    def check_read(self, fn, content):18        fd = io.BytesIO(content)19        return fn(fd)20    def test_write_ushort(self):21        content = self.check_write(write_ushort, 1)22        self.assertEqual(b"\x01\x00", content)23    def test_write_sbyte(self):24        content = self.check_write(write_sbyte, 1)25        self.assertEqual(b"\x01", content)26        content = self.check_write(write_sbyte, -1)27        self.assertEqual(b"\xff", content)28    def test_write_ubyte(self):29        content = self.check_write(write_ubyte, 1)30        self.assertEqual(b"\x01", content)31        content = self.check_write(write_ubyte, 255)32        self.assertEqual(b"\xff", content)33    def test_read_ushort(self):34        content = self.check_read(read_ushort, b"\x01\x00")35        self.assertEqual(1, content)36    def test_read_sbyte(self):37        content = self.check_read(read_sbyte, b"\x01")38        self.assertEqual(1, content)39        content = self.check_read(read_sbyte, b"\xff")40        self.assertEqual(-1, content)41    def test_read_ubyte(self):42        content = self.check_read(read_ubyte, b"\x01")43        self.assertEqual(1, content)44        content = self.check_read(read_ubyte, b"\xff")45        self.assertEqual(255, content)...

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.