Bug 1257516 - Add a file-like class that sends writes to a callback. r?ted
--- a/python/mozbuild/mozbuild/configure/util.py
+++ b/python/mozbuild/mozbuild/configure/util.py
@@ -91,8 +91,42 @@ class ConfigureOutputHandler(logging.Han
stream = self._stderr
msg = '%s\n' % self.format(record)
stream.write(msg)
stream.flush()
except (KeyboardInterrupt, SystemExit):
raise
except:
self.handleError(record)
+
+
+class LineIO(object):
+ '''File-like class that sends each line of the written data to a callback
+ (without carriage returns).
+ '''
+ def __init__(self, callback):
+ self._callback = callback
+ self._buf = ''
+
+ def write(self, buf):
+ lines = buf.splitlines()
+ if not lines:
+ return
+ if self._buf:
+ lines[0] = self._buf + lines[0]
+ self._buf = ''
+ if not buf.endswith('\n'):
+ self._buf = lines[-1]
+ lines = lines[:-1]
+
+ for line in lines:
+ self._callback(line)
+
+ def close(self):
+ if self._buf:
+ self._callback(self._buf)
+ self._buf = ''
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ self.close()
--- a/python/mozbuild/mozbuild/test/configure/test_util.py
+++ b/python/mozbuild/mozbuild/test/configure/test_util.py
@@ -11,16 +11,17 @@ import unittest
import sys
from StringIO import StringIO
from mozunit import main
from mozbuild.configure.util import (
ConfigureOutputHandler,
+ LineIO,
Version,
)
class TestConfigureOutputHandler(unittest.TestCase):
def test_separation(self):
out = StringIO()
err = StringIO()
@@ -171,16 +172,60 @@ class TestConfigureOutputHandler(unittes
self.assertTrue(
ConfigureOutputHandler._is_same_output(fd2, fd3))
finally:
os.close(fd2)
os.remove(path)
+class TestLineIO(unittest.TestCase):
+ def test_lineio(self):
+ lines = []
+ l = LineIO(lambda l: lines.append(l))
+
+ l.write('a')
+ self.assertEqual(lines, [])
+
+ l.write('b')
+ self.assertEqual(lines, [])
+
+ l.write('\n')
+ self.assertEqual(lines, ['ab'])
+
+ l.write('cdef')
+ self.assertEqual(lines, ['ab'])
+
+ l.write('\n')
+ self.assertEqual(lines, ['ab', 'cdef'])
+
+ l.write('ghi\njklm')
+ self.assertEqual(lines, ['ab', 'cdef', 'ghi'])
+
+ l.write('nop\nqrst\nuv\n')
+ self.assertEqual(lines, ['ab', 'cdef', 'ghi', 'jklmnop', 'qrst', 'uv'])
+
+ l.write('wx\nyz')
+ self.assertEqual(lines, ['ab', 'cdef', 'ghi', 'jklmnop', 'qrst', 'uv',
+ 'wx'])
+
+ l.close()
+ self.assertEqual(lines, ['ab', 'cdef', 'ghi', 'jklmnop', 'qrst', 'uv',
+ 'wx', 'yz'])
+
+ def test_lineio_contextmanager(self):
+ lines = []
+ with LineIO(lambda l: lines.append(l)) as l:
+ l.write('a\nb\nc')
+
+ self.assertEqual(lines, ['a', 'b'])
+
+ self.assertEqual(lines, ['a', 'b', 'c'])
+
+
class TestVersion(unittest.TestCase):
def test_version_simple(self):
v = Version('1')
self.assertEqual(v, '1')
self.assertLess(v, '2')
self.assertGreater(v, '0.5')
self.assertEqual(v.major, 1)
self.assertEqual(v.minor, 0)