(self)
| 202 | self.assertTrue(logged[-1].startswith('thread 2 ret val is')) |
| 203 | |
| 204 | def test_tcp_elf_linux_x86(self): |
| 205 | logged: List[str] = [] |
| 206 | |
| 207 | def check_write(ql: Qiling, fd: int, write_buf, count: int): |
| 208 | if fd == 2: |
| 209 | content = ql.mem.read(write_buf, count) |
| 210 | |
| 211 | logged.extend(content.decode().splitlines()) |
| 212 | |
| 213 | ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_tcp_test', '20000'], X86_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG) |
| 214 | |
| 215 | ql.os.stats = QlOsNullStats() |
| 216 | ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER) |
| 217 | ql.run() |
| 218 | |
| 219 | self.assertGreaterEqual(len(logged), 2) |
| 220 | self.assertTrue(logged[-2].startswith('server recv()')) |
| 221 | self.assertTrue(logged[-1].startswith('server send()')) |
| 222 | |
| 223 | # the server is expected to send the value it received, for example: |
| 224 | # 'server recv() return 14.\n' |
| 225 | # 'server send() 14 return 14.\n' |
| 226 | |
| 227 | m = re.search(r'(?P<num>\d+)\.\Z', logged[-2]) |
| 228 | self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"') |
| 229 | |
| 230 | num = m.group('num') |
| 231 | msg = logged[-1].strip() |
| 232 | |
| 233 | self.assertTrue(msg.endswith(f'{num} return {num}.')) |
| 234 | |
| 235 | def test_tcp_elf_linux_x8664(self): |
| 236 | logged: List[str] = [] |
nothing calls this directly
no test coverage detected