vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge pull request #1574 from coreos-inc/generator-buffered

Browse source 
Fix GeneratorFile for working with BufferedReader
This commit is contained in:
josephschorr 2016-06-23 16:35:39 -04:00 committed by GitHub
commit 8232a14518
2 changed files with 31 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
test/test_util.py
View file

@ -1,4 +1,5 @@
import unittest
import magic
from itertools import islice
from semantic_version import Version, Spec
@ -7,6 +8,7 @@ from util.validation import generate_valid_usernames
from util.registry.generatorfile import GeneratorFile
from util.registry.dockerver import docker_version
from util import slash_join
from _pyio import BufferedReader
class TestGeneratorFile(unittest.TestCase):
def sample_generator(self):
@ -76,6 +78,32 @@ class TestGeneratorFile(unittest.TestCase):
self.assertEquals("thisisatest", f.read(60))
self.assertEquals(11, f.tell())
def test_with_bufferedreader(self):
with GeneratorFile(self.sample_generator()) as f:
buffered = BufferedReader(f)
self.assertEquals("thisisatest", buffered.peek(10))
self.assertEquals("thisisates", buffered.read(10))
def mimed_html_generator(self):
yield '<html>'
yield '<body>'
yield 'sometext' * 1024
yield '</body>'
yield '</html>'
def test_magic(self):
mgc = magic.Magic(mime=True)
with GeneratorFile(self.mimed_html_generator()) as f:
buffered = BufferedReader(f)
file_header_bytes = buffered.peek(1024)
self.assertEquals("text/html", mgc.from_buffer(file_header_bytes))
with GeneratorFile(self.sample_generator()) as f:
buffered = BufferedReader(f)
file_header_bytes = buffered.peek(1024)
self.assertEquals("text/plain", mgc.from_buffer(file_header_bytes))
class TestUsernameGenerator(unittest.TestCase):
def assert_generated_output(self, input_username, expected_output):

3
util/registry/generatorfile.py
View file

@ -33,6 +33,9 @@ class GeneratorFile(object):
raise StopIteration
return r
def readable(self):
return not self._closed
def readline(self):
buf = []
while True: