diff options
Diffstat (limited to 'Lib')
| -rw-r--r-- | Lib/logging/__init__.py | 60 | ||||
| -rw-r--r-- | Lib/subprocess.py | 71 | ||||
| -rw-r--r-- | Lib/test/test_logging.py | 31 | ||||
| -rw-r--r-- | Lib/test/test_subprocess.py | 34 |
4 files changed, 179 insertions, 17 deletions
diff --git a/Lib/logging/__init__.py b/Lib/logging/__init__.py index 499c954..cce6ea0 100644 --- a/Lib/logging/__init__.py +++ b/Lib/logging/__init__.py @@ -23,7 +23,8 @@ Copyright (C) 2001-2008 Vinay Sajip. All Rights Reserved. To use, simply 'import logging' and log away! """ -import sys, os, time, io, traceback +import sys, os, time, io, traceback, warnings + __all__ = ['BASIC_FORMAT', 'BufferingFormatter', 'CRITICAL', 'DEBUG', 'ERROR', 'FATAL', 'FileHandler', 'Filter', 'Filterer', 'Formatter', 'Handler', 'INFO', 'LogRecord', 'Logger', 'Manager', 'NOTSET', 'PlaceHolder', @@ -42,8 +43,8 @@ except ImportError: __author__ = "Vinay Sajip <vinay_sajip@red-dove.com>" __status__ = "production" -__version__ = "0.5.0.5" -__date__ = "24 January 2008" +__version__ = "0.5.0.6" +__date__ = "03 December 2008" #--------------------------------------------------------------------------- # Miscellaneous module data @@ -1483,3 +1484,56 @@ except ImportError: # for Python versions < 2.0 old_exit(status) sys.exit = exithook + +# Null handler + +class NullHandler(Handler): + """ + This handler does nothing. It's intended to be used to avoid the + "No handlers could be found for logger XXX" one-off warning. This is + important for library code, which may contain code to log events. If a user + of the library does not configure logging, the one-off warning might be + produced; to avoid this, the library developer simply needs to instantiate + a NullHandler and add it to the top-level logger of the library module or + package. + """ + def emit(self, record): + pass + +# Warnings integration + +_warnings_showwarning = None + +def _showwarning(message, category, filename, lineno, file=None, line=None): + """ + Implementation of showwarnings which redirects to logging, which will first + check to see if the file parameter is None. If a file is specified, it will + delegate to the original warnings implementation of showwarning. Otherwise, + it will call warnings.formatwarning and will log the resulting string to a + warnings logger named "py.warnings" with level logging.WARNING. + """ + if file is not None: + if _warnings_showwarning is not None: + _warnings_showwarning(message, category, filename, lineno, file, line) + else: + s = warnings.formatwarning(message, category, filename, lineno, line) + logger = getLogger("py.warnings") + if not logger.handlers: + logger.addHandler(NullHandler()) + logger.warning("%s", s) + +def captureWarnings(capture): + """ + If capture is true, redirect all warnings to the logging package. + If capture is False, ensure that warnings are not redirected to logging + but to their original destinations. + """ + global _warnings_showwarning + if capture: + if _warnings_showwarning is None: + _warnings_showwarning = warnings.showwarning + warnings.showwarning = _showwarning + else: + if _warnings_showwarning is not None: + warnings.showwarning = _warnings_showwarning + _warnings_showwarning = None diff --git a/Lib/subprocess.py b/Lib/subprocess.py index ac92185..b47ebb8 100644 --- a/Lib/subprocess.py +++ b/Lib/subprocess.py @@ -104,7 +104,7 @@ appearance of the main window and priority for the new process. (Windows only) -This module also defines four shortcut functions: +This module also defines some shortcut functions: call(*popenargs, **kwargs): Run command with arguments. Wait for command to complete, then @@ -151,6 +151,17 @@ getoutput(cmd): >>> subprocess.getoutput('ls /bin/ls') '/bin/ls' +check_output(*popenargs, **kwargs): + Run command with arguments and return its output as a byte string. + + If the exit code was non-zero it raises a CalledProcessError. The + CalledProcessError object will have the return code in the returncode + attribute and output in the output attribute. + + The arguments are the same as for the Popen constructor. Example: + + output = subprocess.check_output(["ls", "-l", "/dev/null"]) + Exceptions ---------- @@ -166,8 +177,8 @@ should prepare for OSErrors. A ValueError will be raised if Popen is called with invalid arguments. -check_call() will raise CalledProcessError, if the called process -returns a non-zero return code. +check_call() and check_output() will raise CalledProcessError, if the +called process returns a non-zero return code. Security @@ -321,12 +332,15 @@ import signal # Exception classes used by this module. class CalledProcessError(Exception): - """This exception is raised when a process run by check_call() returns - a non-zero exit status. The exit status will be stored in the - returncode attribute.""" - def __init__(self, returncode, cmd): + """This exception is raised when a process run by check_call() or + check_output() returns a non-zero exit status. + The exit status will be stored in the returncode attribute; + check_output() will also store the output in the output attribute. + """ + def __init__(self, returncode, cmd, output=None): self.returncode = returncode self.cmd = cmd + self.output = output def __str__(self): return "Command '%s' returned non-zero exit status %d" % (self.cmd, self.returncode) @@ -364,7 +378,7 @@ else: import pickle __all__ = ["Popen", "PIPE", "STDOUT", "call", "check_call", "getstatusoutput", - "getoutput", "CalledProcessError"] + "getoutput", "check_output", "CalledProcessError"] try: MAXFD = os.sysconf("SC_OPEN_MAX") @@ -410,12 +424,45 @@ def check_call(*popenargs, **kwargs): check_call(["ls", "-l"]) """ retcode = call(*popenargs, **kwargs) - cmd = kwargs.get("args") - if cmd is None: - cmd = popenargs[0] if retcode: + cmd = kwargs.get("args") + if cmd is None: + cmd = popenargs[0] raise CalledProcessError(retcode, cmd) - return retcode + return 0 + + +def check_output(*popenargs, **kwargs): + """Run command with arguments and return its output as a byte string. + + If the exit code was non-zero it raises a CalledProcessError. The + CalledProcessError object will have the return code in the returncode + attribute and output in the output attribute. + + The arguments are the same as for the Popen constructor. Example: + + >>> check_output(["ls", "-l", "/dev/null"]) + 'crw-rw-rw- 1 root root 1, 3 Oct 18 2007 /dev/null\n' + + The stdout argument is not allowed as it is used internally. + To capture standard error in the result, use stderr=subprocess.STDOUT. + + >>> check_output(["/bin/sh", "-c", + "ls -l non_existant_file ; exit 0"], + stderr=subprocess.STDOUT) + 'ls: non_existant_file: No such file or directory\n' + """ + if 'stdout' in kwargs: + raise ValueError('stdout argument not allowed, it will be overridden.') + process = Popen(*popenargs, stdout=PIPE, **kwargs) + output, unused_err = process.communicate() + retcode = process.poll() + if retcode: + cmd = kwargs.get("args") + if cmd is None: + cmd = popenargs[0] + raise CalledProcessError(retcode, cmd, output=output) + return output def list2cmdline(seq): diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py index c99a650..c179d14 100644 --- a/Lib/test/test_logging.py +++ b/Lib/test/test_logging.py @@ -18,7 +18,7 @@ """Test harness for the logging module. Run all tests. -Copyright (C) 2001-2002 Vinay Sajip. All Rights Reserved. +Copyright (C) 2001-2008 Vinay Sajip. All Rights Reserved. """ import logging @@ -44,6 +44,7 @@ import threading import time import types import unittest +import warnings import weakref @@ -885,6 +886,32 @@ class EncodingTest(BaseTest): if os.path.isfile(fn): os.remove(fn) +class WarningsTest(BaseTest): + def test_warnings(self): + logging.captureWarnings(True) + warnings.filterwarnings("always", category=UserWarning) + try: + file = io.StringIO() + h = logging.StreamHandler(file) + logger = logging.getLogger("py.warnings") + logger.addHandler(h) + warnings.warn("I'm warning you...") + logger.removeHandler(h) + s = file.getvalue() + h.close() + self.assertTrue(s.find("UserWarning: I'm warning you...\n") > 0) + + #See if an explicit file uses the original implementation + file = io.StringIO() + warnings.showwarning("Explicit", UserWarning, "dummy.py", 42, file, + "Dummy line") + s = file.getvalue() + file.close() + self.assertEqual(s, "dummy.py:42: UserWarning: Explicit\n Dummy line\n") + finally: + warnings.resetwarnings() + logging.captureWarnings(False) + # Set the locale to the platform-dependent default. I have no idea # why the test does this, but in any case we save the current locale # first and restore it at the end. @@ -893,7 +920,7 @@ def test_main(): run_unittest(BuiltinLevelsTest, BasicFilterTest, CustomLevelsAndFiltersTest, MemoryHandlerTest, ConfigFileTest, SocketHandlerTest, MemoryTest, - EncodingTest) + EncodingTest, WarningsTest) if __name__ == "__main__": test_main() diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index f4f1cd5..eb4e759 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -73,6 +73,40 @@ class ProcessTestCase(unittest.TestCase): else: self.fail("Expected CalledProcessError") + def test_check_output(self): + # check_output() function with zero return code + output = subprocess.check_output( + [sys.executable, "-c", "print('BDFL')"]) + self.assertTrue(b'BDFL' in output) + + def test_check_output_nonzero(self): + # check_call() function with non-zero return code + try: + subprocess.check_output( + [sys.executable, "-c", "import sys; sys.exit(5)"]) + except subprocess.CalledProcessError as e: + self.assertEqual(e.returncode, 5) + else: + self.fail("Expected CalledProcessError") + + def test_check_output_stderr(self): + # check_output() function stderr redirected to stdout + output = subprocess.check_output( + [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"], + stderr=subprocess.STDOUT) + self.assertTrue(b'BDFL' in output) + + def test_check_output_stdout_arg(self): + # check_output() function stderr redirected to stdout + try: + output = subprocess.check_output( + [sys.executable, "-c", "print('will not be run')"], + stdout=sys.stdout) + except ValueError as e: + self.assertTrue('stdout' in e.args[0]) + else: + self.fail("Expected ValueError when stdout arg supplied.") + def test_call_kwargs(self): # call() function with keyword args newenv = os.environ.copy() |