subprocess2_test.py 文件源码

def test_nag_timer(self):
    w = []
    l = logging.getLogger()
    class _Filter(logging.Filter):
      def filter(self, record):
        if record.levelno == logging.WARNING:
          w.append(record.getMessage().lstrip())
        return 0
    f = _Filter()
    l.addFilter(f)
    proc = subprocess2.Popen(
        self.exe + ['--stdout', '--sleep_first'], stdout=PIPE)
    res = proc.communicate(nag_timer=3), proc.returncode
    l.removeFilter(f)
    self._check_res(res, 'A\nBB\nCCC\n', None, 0)
    expected = ['No output for 3 seconds from command:', proc.cmd_str,
                'No output for 6 seconds from command:', proc.cmd_str,
                'No output for 9 seconds from command:', proc.cmd_str]
    self.assertEquals(w, expected)