stream_test.py 文件源码

def testInputStreams(self):
        input_spec = {
            'mode': 'http',
            'method': 'GET',
            'url': 'http://mockedhost'
        }

        @httmock.urlmatch(netloc='^mockedhost$', method='GET')
        def mock_fetch(url, request):
            return 'hello\nworld'

        adapters = {
            _pipepath: make_stream_fetch_adapter(input_spec)
        }
        cmd = [sys.executable, _iscript, _pipepath]

        try:
            with captureOutput() as stdpipes, httmock.HTTMock(mock_fetch):
                run_process(cmd, input_pipes=adapters)
        except Exception:
            print('Stdout/stderr from exception: ')
            print(stdpipes)
            raise
        self.assertEqual(stdpipes, ['olleh\ndlrow\n', ''])