upstream/mercurial-mirror Files · tests/test-cappedreader.py

tests: fix test-sparse-revlog...

tests: fix test-sparse-revlog This one is not covered by the CIbecause I requires an expensive artifact to be cached. So it goes out of think on regular basis (we should fix that…) The test ouput was affected by as we filtering now happens sooner, removing for the output.

Gregory Szorc - - Load All Authors

File last commit:

r49730:6000f5b2 default


                r50521:da636e7a

default

Download file

             test-cappedreader.py
        
                    90 lines
            
             | 2.5 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / tests / test-cappedreader.py
          
                    History
                
                 |
                  Source
                 | Raw
                 |Copy content
                 |Copy permalink

        Gregory Szorc
    
util: add a file object proxy that can read at most N bytes...

              r36382
            
      import io

      import unittest

        Augie Fackler
    
formatting: blacken the codebase...

              r43346
            
      from mercurial import util

        Gregory Szorc
    
util: add a file object proxy that can read at most N bytes...

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

        Augie Fackler
    
formatting: blacken the codebase...

              r43346
            
        Gregory Szorc
    
util: add a file object proxy that can read at most N bytes...

              r36382
            
      if __name__ == '__main__':

          import silenttestrunner

        Augie Fackler
    
formatting: blacken the codebase...

              r43346
            
        Gregory Szorc
    
util: add a file object proxy that can read at most N bytes...

              r36382
            
          silenttestrunner.main(__name__)

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

Gregory Szorc util: add a file object proxy that can read at most N bytes...	r36382	import io
		import unittest

Augie Fackler formatting: blacken the codebase...	r43346	from mercurial import util

Gregory Szorc util: add a file object proxy that can read at most N bytes...	r36382
		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)

Augie Fackler formatting: blacken the codebase...	r43346
Gregory Szorc util: add a file object proxy that can read at most N bytes...	r36382	if __name__ == '__main__':
		import silenttestrunner
Augie Fackler formatting: blacken the codebase...	r43346
Gregory Szorc util: add a file object proxy that can read at most N bytes...	r36382	silenttestrunner.main(__name__)