Mercurial > hg
view tests/test-cappedreader.py @ 42583:b9bc47211cf5
abort: added support for rebase
This adds support of `rebase` to `hg abort` plan.
An independent abort logic for `rebase` is created
under `abortrebase()` function. For this a seperate
`rebaseruntime` object is created under the function to
handle an unfinished `rebasestate` and abort that using
abort logic under `_prepareabortorcontinue`.
Results of tests are shown.
Differential Revision: https://phab.mercurial-scm.org/D6568
author | Taapas Agrawal <taapas2897@gmail.com> |
---|---|
date | Sun, 23 Jun 2019 23:11:35 +0530 |
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__)