tests/test-cappedreader.py

phases: invalidate the phases set less often on retract boundary

We already have the information to update the phase set, so we do so directly
instead of invalidating the cache.

This show a sizeable speedup in our `perf::unbundle` benchmark on the
many-draft mozilla-try repository.

### data-env-vars.name            = mozilla-try-2023-03-22-zstd-sparse-revlog
  # benchmark.name                = hg.perf.perf-unbundle
  # bin-env-vars.hg.flavor        = no-rust
  # bin-env-vars.hg.py-re2-module = default
  # benchmark.variants.issue6528  = disabled

  # benchmark.variants.revs = last-10
before: 2.055259 seconds
after:  1.887064 seconds (-8.18%)
  # benchmark.variants.revs = last-100
before: 2.409239 seconds
after:  2.222429 seconds (-7.75%)
  # benchmark.variants.revs = last-1000
before: 3.945648 seconds
after:  3.762480 seconds (-4.64%)

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__)