(bbot_scanner)
| 5 | |
| 6 | @pytest.mark.asyncio |
| 7 | async def test_command(bbot_scanner): |
| 8 | scan1 = bbot_scanner() |
| 9 | |
| 10 | # test timeouts |
| 11 | command = ["sleep", "3"] |
| 12 | start = time.time() |
| 13 | with pytest.raises(asyncio.exceptions.TimeoutError): |
| 14 | await scan1.helpers.run(command, idle_timeout=1) |
| 15 | end = time.time() |
| 16 | elapsed = end - start |
| 17 | assert 0 < elapsed < 2 |
| 18 | |
| 19 | start = time.time() |
| 20 | with pytest.raises(asyncio.exceptions.TimeoutError): |
| 21 | async for line in scan1.helpers.run_live(command, idle_timeout=1): |
| 22 | print(line) |
| 23 | end = time.time() |
| 24 | elapsed = end - start |
| 25 | assert 0 < elapsed < 2 |
| 26 | |
| 27 | # run |
| 28 | assert "plumbus\n" == (await scan1.helpers.run(["echo", "plumbus"])).stdout |
| 29 | assert b"plumbus\n" == (await scan1.helpers.run(["echo", "plumbus"], text=False)).stdout |
| 30 | result = (await scan1.helpers.run(["cat"], input="some\nrandom\nstdin")).stdout |
| 31 | assert result.splitlines() == ["some", "random", "stdin"] |
| 32 | result = (await scan1.helpers.run(["cat"], input=b"some\nrandom\nstdin", text=False)).stdout |
| 33 | assert result.splitlines() == [b"some", b"random", b"stdin"] |
| 34 | result = (await scan1.helpers.run(["cat"], input=["some", "random", "stdin"])).stdout |
| 35 | assert result.splitlines() == ["some", "random", "stdin"] |
| 36 | result = (await scan1.helpers.run(["cat"], input=[b"some", b"random", b"stdin"], text=False)).stdout |
| 37 | assert result.splitlines() == [b"some", b"random", b"stdin"] |
| 38 | |
| 39 | # test overflow - run |
| 40 | tmpfile_path = Path("/tmp/test_bigfile") |
| 41 | with open(tmpfile_path, "w") as f: |
| 42 | # write 2MB |
| 43 | f.write("A" * 1024 * 1024 * 2) |
| 44 | result = (await scan1.helpers.run(["cat", str(tmpfile_path)], limit=1024 * 64, text=False)).stdout |
| 45 | assert len(result) == 1024 * 1024 * 2 |
| 46 | tmpfile_path.unlink(missing_ok=True) |
| 47 | # test overflow - run_live |
| 48 | tmpfile_path = Path("/tmp/test_bigfile") |
| 49 | with open(tmpfile_path, "w") as f: |
| 50 | # write 2MB |
| 51 | f.write("A" * 10 + "\n") |
| 52 | f.write("B" * 1024 * 1024 * 2 + "\n") |
| 53 | f.write("C" * 10 + "\n") |
| 54 | lines = [] |
| 55 | async for line in scan1.helpers.run_live(["cat", str(tmpfile_path)], limit=1024 * 64): |
| 56 | lines.append(line) |
| 57 | # only a small bit of the overflowed line survives, that's okay. |
| 58 | assert lines == ["AAAAAAAAAA", "BBBBBBBBBBB", "CCCCCCCCCC"] |
| 59 | tmpfile_path.unlink(missing_ok=True) |
| 60 | |
| 61 | # run_live |
| 62 | lines = [] |
| 63 | async for line in scan1.helpers.run_live(["echo", "plumbus"]): |
| 64 | lines.append(line) |
nothing calls this directly
no test coverage detected
searching dependent graphs…