mirror of
https://github.com/caronc/apprise.git
synced 2024-11-22 16:13:12 +01:00
LogCapture() class for those wishing to capture notification output (#311)
This commit is contained in:
Chris Caron
GitHub
parent
4d6cd73452
commit
52a8317356
@ -25,7 +25,7 @@
|
||||
|
||||
import re
|
||||
import six
|
||||
import logging
|
||||
from .logger import logger
|
||||
from time import sleep
|
||||
from datetime import datetime
|
||||
from xml.sax.saxutils import escape as sax_escape
|
||||
@ -115,8 +115,8 @@ class URLBase(object):
|
||||
# Secure sites should be verified against a Certificate Authority
|
||||
verify_certificate = True
|
||||
|
||||
# Logging
|
||||
logger = logging.getLogger(__name__)
|
||||
# Logging to our global logger
|
||||
logger = logger
|
||||
|
||||
# Define a default set of template arguments used for dynamically building
|
||||
# details about our individual plugins for developers.
|
||||
|
||||
@ -57,10 +57,13 @@ from .AppriseAsset import AppriseAsset
|
||||
from .AppriseConfig import AppriseConfig
|
||||
from .AppriseAttachment import AppriseAttachment
|
||||
|
||||
# Inherit our logging with our additional entries added to it
|
||||
from .logger import logging
|
||||
from .logger import logger
|
||||
from .logger import LogCapture
|
||||
|
||||
# Set default logging handler to avoid "No handler found" warnings.
|
||||
import logging
|
||||
from logging import NullHandler
|
||||
logging.getLogger(__name__).addHandler(NullHandler())
|
||||
logging.getLogger(__name__).addHandler(logging.NullHandler())
|
||||
|
||||
__all__ = [
|
||||
# Core
|
||||
@ -74,4 +77,7 @@ __all__ = [
|
||||
'ContentIncludeMode', 'CONTENT_INCLUDE_MODES',
|
||||
'ContentLocation', 'CONTENT_LOCATIONS',
|
||||
'PrivacyMode',
|
||||
|
||||
# Logging
|
||||
'logging', 'logger', 'LogCapture',
|
||||
]
|
||||
|
||||
@ -23,7 +23,12 @@
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import os
|
||||
import logging
|
||||
from io import StringIO
|
||||
|
||||
# The root identifier needed to monitor 'apprise' logging
|
||||
LOGGER_NAME = 'apprise'
|
||||
|
||||
# Define a verbosity level that is a noisier then debug mode
|
||||
logging.TRACE = logging.DEBUG - 1
|
||||
@ -57,5 +62,136 @@ def deprecate(self, message, *args, **kwargs):
|
||||
logging.Logger.trace = trace
|
||||
logging.Logger.deprecate = deprecate
|
||||
|
||||
# Create ourselve a generic logging reference
|
||||
logger = logging.getLogger('apprise')
|
||||
# Create ourselve a generic (singleton) logging reference
|
||||
logger = logging.getLogger(LOGGER_NAME)
|
||||
|
||||
|
||||
class LogCapture(object):
|
||||
"""
|
||||
A class used to allow one to instantiate loggers that write to
|
||||
memory for temporary purposes. e.g.:
|
||||
|
||||
1. with LogCapture() as captured:
|
||||
2.
|
||||
3. # Send our notification(s)
|
||||
4. aobj.notify("hello world")
|
||||
5.
|
||||
6. # retrieve our logs produced by the above call via our
|
||||
7. # `captured` StringIO object we have access to within the `with`
|
||||
8. # block here:
|
||||
9. print(captured.getvalue())
|
||||
|
||||
"""
|
||||
def __init__(self, path=None, level=None, name=LOGGER_NAME, delete=True,
|
||||
fmt='%(asctime)s - %(levelname)s - %(message)s'):
|
||||
"""
|
||||
Instantiate a temporary log capture object
|
||||
|
||||
If a path is specified, then log content is sent to that file instead
|
||||
of a StringIO object.
|
||||
|
||||
You can optionally specify a logging level such as logging.INFO if you
|
||||
wish, otherwise by default the script uses whatever logging has been
|
||||
set globally. If you set delete to `False` then when using log files,
|
||||
they are not automatically cleaned up afterwards.
|
||||
|
||||
Optionally over-ride the fmt as well if you wish.
|
||||
|
||||
"""
|
||||
# Our memory buffer placeholder
|
||||
self.__buffer_ptr = StringIO()
|
||||
|
||||
# Store our file path as it will determine whether or not we write to
|
||||
# memory and a file
|
||||
self.__path = path
|
||||
self.__delete = delete
|
||||
|
||||
# Our logging level tracking
|
||||
self.__level = level
|
||||
self.__restore_level = None
|
||||
|
||||
# Acquire a pointer to our logger
|
||||
self.__logger = logging.getLogger(name)
|
||||
|
||||
# Prepare our handler
|
||||
self.__handler = logging.StreamHandler(self.__buffer_ptr) \
|
||||
if not self.__path else logging.FileHandler(
|
||||
self.__path, mode='a', encoding='utf-8')
|
||||
|
||||
# Use the specified level, otherwise take on the already
|
||||
# effective level of our logger
|
||||
self.__handler.setLevel(
|
||||
self.__level if self.__level is not None
|
||||
else self.__logger.getEffectiveLevel())
|
||||
|
||||
# Prepare our formatter
|
||||
self.__handler.setFormatter(logging.Formatter(fmt))
|
||||
|
||||
def __enter__(self):
|
||||
"""
|
||||
Allows logger manipulation within a 'with' block
|
||||
"""
|
||||
|
||||
if self.__level is not None:
|
||||
# Temporary adjust our log level if required
|
||||
self.__restore_level = self.__logger.getEffectiveLevel()
|
||||
if self.__restore_level > self.__level:
|
||||
# Bump our log level up for the duration of our `with`
|
||||
self.__logger.setLevel(self.__level)
|
||||
|
||||
else:
|
||||
# No restoration required
|
||||
self.__restore_level = None
|
||||
|
||||
else:
|
||||
# Do nothing but enforce that we have nothing to restore to
|
||||
self.__restore_level = None
|
||||
|
||||
if self.__path:
|
||||
# If a path has been identified, ensure we can write to the path
|
||||
# and that the file exists
|
||||
with open(self.__path, 'a'):
|
||||
os.utime(self.__path, None)
|
||||
|
||||
# Update our buffer pointer
|
||||
self.__buffer_ptr = open(self.__path, 'r')
|
||||
|
||||
# Add our handler
|
||||
self.__logger.addHandler(self.__handler)
|
||||
|
||||
# return our memory pointer
|
||||
return self.__buffer_ptr
|
||||
|
||||
def __exit__(self, exc_type, exc_value, tb):
|
||||
"""
|
||||
removes the handler gracefully when the with block has completed
|
||||
"""
|
||||
|
||||
# Flush our content
|
||||
self.__handler.flush()
|
||||
self.__buffer_ptr.flush()
|
||||
|
||||
# Drop our handler
|
||||
self.__logger.removeHandler(self.__handler)
|
||||
|
||||
if self.__restore_level is not None:
|
||||
# Restore level
|
||||
self.__logger.setLevel(self.__restore_level)
|
||||
|
||||
if self.__path:
|
||||
# Close our file pointer
|
||||
self.__buffer_ptr.close()
|
||||
if self.__delete:
|
||||
try:
|
||||
# Always remove file afterwards
|
||||
os.unlink(self.__path)
|
||||
|
||||
except OSError:
|
||||
# It's okay if the file does not exist
|
||||
pass
|
||||
|
||||
if exc_type is not None:
|
||||
# pass exception on if one was generated
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
@ -23,10 +23,18 @@
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import re
|
||||
import os
|
||||
import sys
|
||||
import mock
|
||||
import pytest
|
||||
from apprise import Apprise
|
||||
from apprise import URLBase
|
||||
from apprise.logger import LogCapture
|
||||
|
||||
# Disable logging for a cleaner testing output
|
||||
import logging
|
||||
from apprise.logger import logging
|
||||
from apprise.logger import logger
|
||||
|
||||
|
||||
def test_apprise_logger():
|
||||
@ -58,3 +66,557 @@ def test_apprise_logger():
|
||||
|
||||
# Disable Logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major <= 2, reason="Requires Python 3.x+")
|
||||
def test_apprise_log_memory_captures_py3():
|
||||
"""
|
||||
API: Apprise() Python v3 Log Captures
|
||||
|
||||
"""
|
||||
|
||||
# Ensure we're not running in a disabled state
|
||||
logging.disable(logging.NOTSET)
|
||||
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
with LogCapture(level=logging.TRACE) as stream:
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
|
||||
# We have a log entry for each of the 6 logs we generated above
|
||||
assert 'trace' in stream.getvalue()
|
||||
assert 'debug' in stream.getvalue()
|
||||
assert 'info' in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
assert len(logs) == 6
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.CRITICAL
|
||||
|
||||
logger.setLevel(logging.TRACE)
|
||||
with LogCapture(level=logging.DEBUG) as stream:
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
# We have a log entry for 5 of the log entries we generated above
|
||||
# There will be no 'trace' entry
|
||||
assert 'trace' not in stream.getvalue()
|
||||
assert 'debug' in stream.getvalue()
|
||||
assert 'info' in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 5
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.TRACE
|
||||
|
||||
logger.setLevel(logging.ERROR)
|
||||
with LogCapture(level=logging.WARNING) as stream:
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
# We have a log entry for 3 of the log entries we generated above
|
||||
# There will be no 'trace', 'debug', or 'info' entry
|
||||
assert 'trace' not in stream.getvalue()
|
||||
assert 'debug' not in stream.getvalue()
|
||||
assert 'info' not in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 3
|
||||
|
||||
# Set a global level of ERROR
|
||||
logger.setLevel(logging.ERROR)
|
||||
|
||||
# Use the default level of None (by not specifying one); we then
|
||||
# use whatever has been defined globally
|
||||
with LogCapture() as stream:
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
assert 'trace' not in stream.getvalue()
|
||||
assert 'debug' not in stream.getvalue()
|
||||
assert 'info' not in stream.getvalue()
|
||||
assert 'warning' not in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 2
|
||||
|
||||
# Verify that we did not lose our effective log level
|
||||
assert logger.getEffectiveLevel() == logging.ERROR
|
||||
|
||||
with LogCapture(level=logging.TRACE) as stream:
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
# We have a log entry for each of the 6 logs we generated above
|
||||
assert 'trace' in stream.getvalue()
|
||||
assert 'debug' in stream.getvalue()
|
||||
assert 'info' in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 6
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.ERROR
|
||||
|
||||
# Test capture where our notification throws an unhandled exception
|
||||
obj = Apprise.instantiate('json://user:password@example.com')
|
||||
with mock.patch('requests.post', side_effect=NotImplementedError()):
|
||||
with pytest.raises(NotImplementedError):
|
||||
# Our exception gets caught in side our with() block
|
||||
# and although raised, all graceful handling of the log
|
||||
# is reverted as it was
|
||||
with LogCapture(level=logging.TRACE) as stream:
|
||||
obj.send("hello world")
|
||||
|
||||
# Disable Logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major >= 3, reason="Requires Python 2.x+")
|
||||
def test_apprise_log_memory_captures_py2():
|
||||
"""
|
||||
API: Apprise() Python v2 Log Captures
|
||||
|
||||
"""
|
||||
|
||||
# Ensure we're not running in a disabled state
|
||||
logging.disable(logging.NOTSET)
|
||||
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
with LogCapture(level=logging.TRACE) as stream:
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
|
||||
# We have a log entry for each of the 6 logs we generated above
|
||||
assert 'trace' in stream.getvalue()
|
||||
assert 'debug' in stream.getvalue()
|
||||
assert 'info' in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
assert len(logs) == 6
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.CRITICAL
|
||||
|
||||
logger.setLevel(logging.TRACE)
|
||||
with LogCapture(level=logging.DEBUG) as stream:
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
# We have a log entry for 5 of the log entries we generated above
|
||||
# There will be no 'trace' entry
|
||||
assert 'trace' not in stream.getvalue()
|
||||
assert 'debug' in stream.getvalue()
|
||||
assert 'info' in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 5
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.TRACE
|
||||
|
||||
logger.setLevel(logging.ERROR)
|
||||
with LogCapture(level=logging.WARNING) as stream:
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
# We have a log entry for 3 of the log entries we generated above
|
||||
# There will be no 'trace', 'debug', or 'info' entry
|
||||
assert 'trace' not in stream.getvalue()
|
||||
assert 'debug' not in stream.getvalue()
|
||||
assert 'info' not in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 3
|
||||
|
||||
# Set a global level of ERROR
|
||||
logger.setLevel(logging.ERROR)
|
||||
|
||||
# Use the default level of None (by not specifying one); we then
|
||||
# use whatever has been defined globally
|
||||
with LogCapture() as stream:
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
assert 'trace' not in stream.getvalue()
|
||||
assert 'debug' not in stream.getvalue()
|
||||
assert 'info' not in stream.getvalue()
|
||||
assert 'warning' not in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 2
|
||||
|
||||
# Verify that we did not lose our effective log level
|
||||
assert logger.getEffectiveLevel() == logging.ERROR
|
||||
|
||||
with LogCapture(level=logging.TRACE) as stream:
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
# We have a log entry for each of the 6 logs we generated above
|
||||
assert 'trace' in stream.getvalue()
|
||||
assert 'debug' in stream.getvalue()
|
||||
assert 'info' in stream.getvalue()
|
||||
assert 'warning' in stream.getvalue()
|
||||
assert 'error' in stream.getvalue()
|
||||
assert 'deprecate' in stream.getvalue()
|
||||
|
||||
logs = re.split(r'\r*\n', stream.getvalue().rstrip())
|
||||
assert len(logs) == 6
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.ERROR
|
||||
|
||||
# Test capture where our notification throws an unhandled exception
|
||||
obj = Apprise.instantiate('json://user:password@example.com')
|
||||
with mock.patch('requests.post', side_effect=NotImplementedError()):
|
||||
with pytest.raises(NotImplementedError):
|
||||
# Our exception gets caught in side our with() block
|
||||
# and although raised, all graceful handling of the log
|
||||
# is reverted as it was
|
||||
with LogCapture(level=logging.TRACE) as stream:
|
||||
obj.send("hello world")
|
||||
|
||||
# Disable Logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major <= 2, reason="Requires Python 3.x+")
|
||||
def test_apprise_log_file_captures_py3(tmpdir):
|
||||
"""
|
||||
API: Apprise() Python v3 Logfile Captures
|
||||
|
||||
"""
|
||||
|
||||
# Ensure we're not running in a disabled state
|
||||
logging.disable(logging.NOTSET)
|
||||
|
||||
log_file = tmpdir.join('capture.log')
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
with LogCapture(path=str(log_file), level=logging.TRACE) as fp:
|
||||
# The file will exit now
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
content = fp.read().rstrip()
|
||||
logs = re.split(r'\r*\n', content)
|
||||
|
||||
# We have a log entry for each of the 6 logs we generated above
|
||||
assert 'trace' in content
|
||||
assert 'debug' in content
|
||||
assert 'info' in content
|
||||
assert 'warning' in content
|
||||
assert 'error' in content
|
||||
assert 'deprecate' in content
|
||||
assert len(logs) == 6
|
||||
|
||||
# The file is automatically cleaned up afterwards
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.CRITICAL
|
||||
|
||||
logger.setLevel(logging.TRACE)
|
||||
with LogCapture(path=str(log_file), level=logging.DEBUG) as fp:
|
||||
# The file will exit now
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
content = fp.read().rstrip()
|
||||
logs = re.split(r'\r*\n', content)
|
||||
|
||||
# We have a log entry for 5 of the log entries we generated above
|
||||
# There will be no 'trace' entry
|
||||
assert 'trace' not in content
|
||||
assert 'debug' in content
|
||||
assert 'info' in content
|
||||
assert 'warning' in content
|
||||
assert 'error' in content
|
||||
assert 'deprecate' in content
|
||||
|
||||
assert len(logs) == 5
|
||||
|
||||
# Remove our file before we exit the with clause
|
||||
# this causes our delete() call to throw gracefully inside
|
||||
os.unlink(str(log_file))
|
||||
|
||||
# Verify file is gone
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.TRACE
|
||||
|
||||
logger.setLevel(logging.ERROR)
|
||||
with LogCapture(path=str(log_file), delete=False,
|
||||
level=logging.WARNING) as fp:
|
||||
|
||||
# Verify exists
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
logger.trace("trace")
|
||||
logger.debug("debug")
|
||||
logger.info("info")
|
||||
logger.warning("warning")
|
||||
logger.error("error")
|
||||
logger.deprecate("deprecate")
|
||||
|
||||
content = fp.read().rstrip()
|
||||
logs = re.split(r'\r*\n', content)
|
||||
|
||||
# We have a log entry for 3 of the log entries we generated above
|
||||
# There will be no 'trace', 'debug', or 'info' entry
|
||||
assert 'trace' not in content
|
||||
assert 'debug' not in content
|
||||
assert 'info' not in content
|
||||
assert 'warning' in content
|
||||
assert 'error' in content
|
||||
assert 'deprecate' in content
|
||||
|
||||
assert len(logs) == 3
|
||||
|
||||
# Verify the file still exists (because delete was set to False)
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
# remove it now
|
||||
os.unlink(str(log_file))
|
||||
|
||||
# Enure it's been removed
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
# Set a global level of ERROR
|
||||
logger.setLevel(logging.ERROR)
|
||||
|
||||
# Test case where we can't open the file
|
||||
with mock.patch('builtins.open', side_effect=OSError()):
|
||||
# Use the default level of None (by not specifying one); we then
|
||||
# use whatever has been defined globally
|
||||
with pytest.raises(OSError):
|
||||
with LogCapture(path=str(log_file)) as fp:
|
||||
# we'll never get here because we'll fail to open the file
|
||||
pass
|
||||
|
||||
# Disable Logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major >= 3, reason="Requires Python 2.x+")
|
||||
def test_apprise_log_file_captures_py2(tmpdir):
|
||||
"""
|
||||
API: Apprise() Python v2 Logfile Captures
|
||||
|
||||
"""
|
||||
|
||||
# Ensure we're not running in a disabled state
|
||||
logging.disable(logging.NOTSET)
|
||||
|
||||
log_file = tmpdir.join('capture.log')
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
with LogCapture(path=str(log_file), level=logging.TRACE) as fp:
|
||||
# The file will exit now
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
content = fp.read().rstrip()
|
||||
logs = re.split(r'\r*\n', content)
|
||||
|
||||
# We have a log entry for each of the 6 logs we generated above
|
||||
assert 'trace' in content
|
||||
assert 'debug' in content
|
||||
assert 'info' in content
|
||||
assert 'warning' in content
|
||||
assert 'error' in content
|
||||
assert 'deprecate' in content
|
||||
assert len(logs) == 6
|
||||
|
||||
# The file is automatically cleaned up afterwards
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.CRITICAL
|
||||
|
||||
logger.setLevel(logging.TRACE)
|
||||
with LogCapture(path=str(log_file), level=logging.DEBUG) as fp:
|
||||
# The file will exit now
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
content = fp.read().rstrip()
|
||||
logs = re.split(r'\r*\n', content)
|
||||
|
||||
# We have a log entry for 5 of the log entries we generated above
|
||||
# There will be no 'trace' entry
|
||||
assert 'trace' not in content
|
||||
assert 'debug' in content
|
||||
assert 'info' in content
|
||||
assert 'warning' in content
|
||||
assert 'error' in content
|
||||
assert 'deprecate' in content
|
||||
|
||||
assert len(logs) == 5
|
||||
|
||||
# Remove our file before we exit the with clause
|
||||
# this causes our delete() call to throw gracefully inside
|
||||
os.unlink(str(log_file))
|
||||
|
||||
# Verify file is gone
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
# Verify that we did not lose our effective log level even though
|
||||
# the above steps the level up for the duration of the capture
|
||||
assert logger.getEffectiveLevel() == logging.TRACE
|
||||
|
||||
logger.setLevel(logging.ERROR)
|
||||
with LogCapture(path=str(log_file), delete=False,
|
||||
level=logging.WARNING) as fp:
|
||||
|
||||
# Verify exists
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
logger.trace(u"trace")
|
||||
logger.debug(u"debug")
|
||||
logger.info(u"info")
|
||||
logger.warning(u"warning")
|
||||
logger.error(u"error")
|
||||
logger.deprecate(u"deprecate")
|
||||
|
||||
content = fp.read().rstrip()
|
||||
logs = re.split(r'\r*\n', content)
|
||||
|
||||
# We have a log entry for 3 of the log entries we generated above
|
||||
# There will be no 'trace', 'debug', or 'info' entry
|
||||
assert 'trace' not in content
|
||||
assert 'debug' not in content
|
||||
assert 'info' not in content
|
||||
assert 'warning' in content
|
||||
assert 'error' in content
|
||||
assert 'deprecate' in content
|
||||
|
||||
assert len(logs) == 3
|
||||
|
||||
# Verify the file still exists (because delete was set to False)
|
||||
assert os.path.isfile(str(log_file))
|
||||
|
||||
# remove it now
|
||||
os.unlink(str(log_file))
|
||||
|
||||
# Enure it's been removed
|
||||
assert not os.path.isfile(str(log_file))
|
||||
|
||||
# Set a global level of ERROR
|
||||
logger.setLevel(logging.ERROR)
|
||||
|
||||
# Test case where we can't open the file
|
||||
with mock.patch('__builtin__.open', side_effect=OSError()):
|
||||
# Use the default level of None (by not specifying one); we then
|
||||
# use whatever has been defined globally
|
||||
with pytest.raises(OSError):
|
||||
with LogCapture(path=str(log_file)) as fp:
|
||||
# we'll never get here because we'll fail to open the file
|
||||
pass
|
||||
|
||||
# Disable Logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user