(self)
| 125 | |
| 126 | @requires_subprocess() |
| 127 | def test_poll2(self): |
| 128 | cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done' |
| 129 | proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, |
| 130 | bufsize=0) |
| 131 | self.enterContext(proc) |
| 132 | p = proc.stdout |
| 133 | pollster = select.poll() |
| 134 | pollster.register( p, select.POLLIN ) |
| 135 | for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10: |
| 136 | fdlist = pollster.poll(tout) |
| 137 | if (fdlist == []): |
| 138 | continue |
| 139 | fd, flags = fdlist[0] |
| 140 | if flags & select.POLLHUP: |
| 141 | line = p.readline() |
| 142 | if line != b"": |
| 143 | self.fail('error: pipe seems to be closed, but still returns data') |
| 144 | continue |
| 145 | |
| 146 | elif flags & select.POLLIN: |
| 147 | line = p.readline() |
| 148 | if not line: |
| 149 | break |
| 150 | self.assertEqual(line, b'testing...\n') |
| 151 | continue |
| 152 | else: |
| 153 | self.fail('Unexpected return value from select.poll: %s' % fdlist) |
| 154 | |
| 155 | def test_poll3(self): |
| 156 | # test int overflow |
nothing calls this directly
no test coverage detected