Mercurial > hg-stable
view tests/test-cappedreader.py @ 40479:197f092b2cd9 stable 4.8.2
server: always close http socket if responding with an error (issue6033)
It's possible for hgweb to respond _very_ early with an error if we're
catching certain types of errors. When we do, we need to tell the client
the socket is toast when there's a POST involved because otherwise there
can be lingering POST data on the socket that will confuse any future
requests on the socket. This manifested as a flaky failure on Linux in an
lfs extension test and a reliable failure on FreeBSD. With this patch
applied, test-lfs-serve-access.t now passes for me on FreeBSD.
Differential Revision: https://phab.mercurial-scm.org/D5498
author | Augie Fackler <raf@durin42.com> |
---|---|
date | Sun, 06 Jan 2019 14:58:54 -0500 |
parents | 01e29e885600 |
children | 2372284d9457 |
line wrap: on
line source
from __future__ import absolute_import, print_function import io import unittest from mercurial import ( util, ) class CappedReaderTests(unittest.TestCase): def testreadfull(self): source = io.BytesIO(b'x' * 100) reader = util.cappedreader(source, 10) res = reader.read(10) self.assertEqual(res, b'x' * 10) self.assertEqual(source.tell(), 10) source.seek(0) reader = util.cappedreader(source, 15) res = reader.read(16) self.assertEqual(res, b'x' * 15) self.assertEqual(source.tell(), 15) source.seek(0) reader = util.cappedreader(source, 100) res = reader.read(100) self.assertEqual(res, b'x' * 100) self.assertEqual(source.tell(), 100) source.seek(0) reader = util.cappedreader(source, 50) res = reader.read() self.assertEqual(res, b'x' * 50) self.assertEqual(source.tell(), 50) source.seek(0) def testreadnegative(self): source = io.BytesIO(b'x' * 100) reader = util.cappedreader(source, 20) res = reader.read(-1) self.assertEqual(res, b'x' * 20) self.assertEqual(source.tell(), 20) source.seek(0) reader = util.cappedreader(source, 100) res = reader.read(-1) self.assertEqual(res, b'x' * 100) self.assertEqual(source.tell(), 100) source.seek(0) def testreadmultiple(self): source = io.BytesIO(b'x' * 100) reader = util.cappedreader(source, 10) for i in range(10): res = reader.read(1) self.assertEqual(res, b'x') self.assertEqual(source.tell(), i + 1) self.assertEqual(source.tell(), 10) res = reader.read(1) self.assertEqual(res, b'') self.assertEqual(source.tell(), 10) source.seek(0) reader = util.cappedreader(source, 45) for i in range(4): res = reader.read(10) self.assertEqual(res, b'x' * 10) self.assertEqual(source.tell(), (i + 1) * 10) res = reader.read(10) self.assertEqual(res, b'x' * 5) self.assertEqual(source.tell(), 45) def readlimitpasteof(self): source = io.BytesIO(b'x' * 100) reader = util.cappedreader(source, 1024) res = reader.read(1000) self.assertEqual(res, b'x' * 100) self.assertEqual(source.tell(), 100) res = reader.read(1000) self.assertEqual(res, b'') self.assertEqual(source.tell(), 100) if __name__ == '__main__': import silenttestrunner silenttestrunner.main(__name__)