Loading python/mozbuild/mozbuild/configure/util.py +62 −0 Original line number Diff line number Diff line Loading @@ -5,6 +5,9 @@ from __future__ import absolute_import, print_function, unicode_literals import itertools import logging import os import sys from distutils.version import LooseVersion Loading Loading @@ -34,3 +37,62 @@ class Version(LooseVersion): if isinstance(other, unicode): other = other.encode('ascii') return LooseVersion.__cmp__(self, other) class ConfigureOutputHandler(logging.Handler): '''A logging handler class that sends info messages to stdout and other messages to stderr. Messages sent to stdout are not formatted with the attached Formatter. Additionally, if they end with '... ', no newline character is printed, making the next message printed follow the '... '. ''' def __init__(self, stdout=sys.stdout, stderr=sys.stderr): super(ConfigureOutputHandler, self).__init__() self._stdout, self._stderr = stdout, stderr try: fd1 = self._stdout.fileno() fd2 = self._stderr.fileno() self._same_output = self._is_same_output(fd1, fd2) except AttributeError: self._same_output = self._stdout == self._stderr self._stdout_waiting = None @staticmethod def _is_same_output(fd1, fd2): if fd1 == fd2: return True stat1 = os.fstat(fd1) stat2 = os.fstat(fd2) return stat1.st_ino == stat2.st_ino and stat1.st_dev == stat2.st_dev WAITING = 1 INTERRUPTED = 2 def emit(self, record): try: if record.levelno == logging.INFO: stream = self._stdout msg = record.getMessage() if (self._stdout_waiting == self.INTERRUPTED and self._same_output): msg = ' ... %s' % msg self._stdout_waiting = msg.endswith('... ') if msg.endswith('... '): self._stdout_waiting = self.WAITING else: self._stdout_waiting = None msg = '%s\n' % msg else: if self._stdout_waiting == self.WAITING and self._same_output: self._stdout_waiting = self.INTERRUPTED self._stdout.write('\n') self._stdout.flush() stream = self._stderr msg = '%s\n' % self.format(record) stream.write(msg) stream.flush() except (KeyboardInterrupt, SystemExit): raise except: self.handleError(record) python/mozbuild/mozbuild/test/configure/test_util.py +166 −1 Original line number Diff line number Diff line Loading @@ -4,11 +4,176 @@ from __future__ import absolute_import, print_function, unicode_literals import logging import os import tempfile import unittest import sys from StringIO import StringIO from mozunit import main from mozbuild.configure.util import Version from mozbuild.configure.util import ( ConfigureOutputHandler, Version, ) class TestConfigureOutputHandler(unittest.TestCase): def test_separation(self): out = StringIO() err = StringIO() name = '%s.test_separation' % self.__class__.__name__ logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) logger.addHandler(ConfigureOutputHandler(out, err)) logger.error('foo') logger.warning('bar') logger.info('baz') logger.debug('qux') self.assertEqual(out.getvalue(), 'baz\n') self.assertEqual(err.getvalue(), 'foo\nbar\nqux\n') def test_format(self): out = StringIO() err = StringIO() name = '%s.test_format' % self.__class__.__name__ logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) handler = ConfigureOutputHandler(out, err) handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s')) logger.addHandler(handler) logger.error('foo') logger.warning('bar') logger.info('baz') logger.debug('qux') self.assertEqual(out.getvalue(), 'baz\n') self.assertEqual( err.getvalue(), 'ERROR:foo\n' 'WARNING:bar\n' 'DEBUG:qux\n' ) def test_continuation(self): out = StringIO() name = '%s.test_continuation' % self.__class__.__name__ logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) handler = ConfigureOutputHandler(out, out) handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s')) logger.addHandler(handler) logger.info('foo') logger.info('checking bar... ') logger.info('yes') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... yes\n' 'qux\n' ) out.seek(0) out.truncate() logger.info('foo') logger.info('checking bar... ') logger.warning('hoge') logger.info('no') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... \n' 'WARNING:hoge\n' ' ... no\n' 'qux\n' ) out.seek(0) out.truncate() logger.info('foo') logger.info('checking bar... ') logger.warning('hoge') logger.warning('fuga') logger.info('no') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... \n' 'WARNING:hoge\n' 'WARNING:fuga\n' ' ... no\n' 'qux\n' ) out.seek(0) out.truncate() err = StringIO() logger.removeHandler(handler) handler = ConfigureOutputHandler(out, err) handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s')) logger.addHandler(handler) logger.info('foo') logger.info('checking bar... ') logger.warning('hoge') logger.warning('fuga') logger.info('no') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... no\n' 'qux\n' ) self.assertEqual( err.getvalue(), 'WARNING:hoge\n' 'WARNING:fuga\n' ) def test_is_same_output(self): fd1 = sys.stderr.fileno() fd2 = os.dup(fd1) try: self.assertTrue(ConfigureOutputHandler._is_same_output(fd1, fd2)) finally: os.close(fd2) fd2, path = tempfile.mkstemp() try: self.assertFalse(ConfigureOutputHandler._is_same_output(fd1, fd2)) fd3 = os.dup(fd2) try: self.assertTrue(ConfigureOutputHandler._is_same_output(fd2, fd3)) finally: os.close(fd3) with open(path, 'a') as fh: fd3 = fh.fileno() self.assertTrue( ConfigureOutputHandler._is_same_output(fd2, fd3)) finally: os.close(fd2) os.remove(path) class TestVersion(unittest.TestCase): Loading Loading
python/mozbuild/mozbuild/configure/util.py +62 −0 Original line number Diff line number Diff line Loading @@ -5,6 +5,9 @@ from __future__ import absolute_import, print_function, unicode_literals import itertools import logging import os import sys from distutils.version import LooseVersion Loading Loading @@ -34,3 +37,62 @@ class Version(LooseVersion): if isinstance(other, unicode): other = other.encode('ascii') return LooseVersion.__cmp__(self, other) class ConfigureOutputHandler(logging.Handler): '''A logging handler class that sends info messages to stdout and other messages to stderr. Messages sent to stdout are not formatted with the attached Formatter. Additionally, if they end with '... ', no newline character is printed, making the next message printed follow the '... '. ''' def __init__(self, stdout=sys.stdout, stderr=sys.stderr): super(ConfigureOutputHandler, self).__init__() self._stdout, self._stderr = stdout, stderr try: fd1 = self._stdout.fileno() fd2 = self._stderr.fileno() self._same_output = self._is_same_output(fd1, fd2) except AttributeError: self._same_output = self._stdout == self._stderr self._stdout_waiting = None @staticmethod def _is_same_output(fd1, fd2): if fd1 == fd2: return True stat1 = os.fstat(fd1) stat2 = os.fstat(fd2) return stat1.st_ino == stat2.st_ino and stat1.st_dev == stat2.st_dev WAITING = 1 INTERRUPTED = 2 def emit(self, record): try: if record.levelno == logging.INFO: stream = self._stdout msg = record.getMessage() if (self._stdout_waiting == self.INTERRUPTED and self._same_output): msg = ' ... %s' % msg self._stdout_waiting = msg.endswith('... ') if msg.endswith('... '): self._stdout_waiting = self.WAITING else: self._stdout_waiting = None msg = '%s\n' % msg else: if self._stdout_waiting == self.WAITING and self._same_output: self._stdout_waiting = self.INTERRUPTED self._stdout.write('\n') self._stdout.flush() stream = self._stderr msg = '%s\n' % self.format(record) stream.write(msg) stream.flush() except (KeyboardInterrupt, SystemExit): raise except: self.handleError(record)
python/mozbuild/mozbuild/test/configure/test_util.py +166 −1 Original line number Diff line number Diff line Loading @@ -4,11 +4,176 @@ from __future__ import absolute_import, print_function, unicode_literals import logging import os import tempfile import unittest import sys from StringIO import StringIO from mozunit import main from mozbuild.configure.util import Version from mozbuild.configure.util import ( ConfigureOutputHandler, Version, ) class TestConfigureOutputHandler(unittest.TestCase): def test_separation(self): out = StringIO() err = StringIO() name = '%s.test_separation' % self.__class__.__name__ logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) logger.addHandler(ConfigureOutputHandler(out, err)) logger.error('foo') logger.warning('bar') logger.info('baz') logger.debug('qux') self.assertEqual(out.getvalue(), 'baz\n') self.assertEqual(err.getvalue(), 'foo\nbar\nqux\n') def test_format(self): out = StringIO() err = StringIO() name = '%s.test_format' % self.__class__.__name__ logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) handler = ConfigureOutputHandler(out, err) handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s')) logger.addHandler(handler) logger.error('foo') logger.warning('bar') logger.info('baz') logger.debug('qux') self.assertEqual(out.getvalue(), 'baz\n') self.assertEqual( err.getvalue(), 'ERROR:foo\n' 'WARNING:bar\n' 'DEBUG:qux\n' ) def test_continuation(self): out = StringIO() name = '%s.test_continuation' % self.__class__.__name__ logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) handler = ConfigureOutputHandler(out, out) handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s')) logger.addHandler(handler) logger.info('foo') logger.info('checking bar... ') logger.info('yes') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... yes\n' 'qux\n' ) out.seek(0) out.truncate() logger.info('foo') logger.info('checking bar... ') logger.warning('hoge') logger.info('no') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... \n' 'WARNING:hoge\n' ' ... no\n' 'qux\n' ) out.seek(0) out.truncate() logger.info('foo') logger.info('checking bar... ') logger.warning('hoge') logger.warning('fuga') logger.info('no') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... \n' 'WARNING:hoge\n' 'WARNING:fuga\n' ' ... no\n' 'qux\n' ) out.seek(0) out.truncate() err = StringIO() logger.removeHandler(handler) handler = ConfigureOutputHandler(out, err) handler.setFormatter(logging.Formatter('%(levelname)s:%(message)s')) logger.addHandler(handler) logger.info('foo') logger.info('checking bar... ') logger.warning('hoge') logger.warning('fuga') logger.info('no') logger.info('qux') self.assertEqual( out.getvalue(), 'foo\n' 'checking bar... no\n' 'qux\n' ) self.assertEqual( err.getvalue(), 'WARNING:hoge\n' 'WARNING:fuga\n' ) def test_is_same_output(self): fd1 = sys.stderr.fileno() fd2 = os.dup(fd1) try: self.assertTrue(ConfigureOutputHandler._is_same_output(fd1, fd2)) finally: os.close(fd2) fd2, path = tempfile.mkstemp() try: self.assertFalse(ConfigureOutputHandler._is_same_output(fd1, fd2)) fd3 = os.dup(fd2) try: self.assertTrue(ConfigureOutputHandler._is_same_output(fd2, fd3)) finally: os.close(fd3) with open(path, 'a') as fh: fd3 = fh.fileno() self.assertTrue( ConfigureOutputHandler._is_same_output(fd2, fd3)) finally: os.close(fd2) os.remove(path) class TestVersion(unittest.TestCase): Loading