rbbratta/test_suppress.py

## test_suppress.py
#!/usr/bin/env python
# Copyright (c) 2016-2017 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from contextlib import contextmanager
from functools import wraps
import logging
import traceback
import sys

log_level = logging.DEBUG

logging.basicConfig(level=logging.DEBUG)

class MyGeneratorContextManager(object):
    """Helper for @contextmanager decorator."""

    def __init__(self, gen):
        self.gen = gen

    def __enter__(self):
        try:
            return self.gen.next()
        except StopIteration:
            raise RuntimeError("generator didn't yield")

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            try:
                self.gen.next()
            except StopIteration:
                return
            else:
                raise RuntimeError("generator didn't stop")
        else:
            if exc_val is None:
                # Need to force instantiation so we can reliably
                # tell if we get the same exception back
                exc_val = exc_type()
            try:
                # output = traceback.format_exception(exc_type, exc_val, exc_tb)
                # output = traceback.format_exception(*sys.exc_info())
                # logging.error("\n".join(output))
                self.gen.throw(*sys.exc_info())
                # self.gen.throw(exc_type, exc_val, exc_tb)
                raise RuntimeError("generator didn't stop after throw()")
            except StopIteration, exc:
                # Suppress the exception *unless* it's the same exception that
                # was passed to throw().  This prevents a StopIteration
                # raised inside the "with" statement from being suppressed
                return exc is not exc_val
            except:
                # only re-raise if it's *not* the exception that was
                # passed to throw(), because __exit__() must not raise
                # an exception unless __exit__() itself failed.  But throw()
                # has to raise the exception to signal propagation, so this
                # fixes the impedance mismatch between the throw() protocol
                # and the __exit__() protocol.
                #
                if sys.exc_info()[1] is not exc_val:
                    raise


def my_contextmanager(func):
    """@contextmanager decorator.

    Typical usage:

        @contextmanager
        def some_generator(<arguments>):
            <setup>
            try:
                yield <value>
            finally:
                <cleanup>

    This makes this:

        with some_generator(<arguments>) as <variable>:
            <body>

    equivalent to this:

        <setup>
        try:
            <variable> = <value>
            <body>
        finally:
            <cleanup>

    """
    @wraps(func)
    def helper(*args, **kwds):
        return MyGeneratorContextManager(func(*args, **kwds))
    return helper

# class suppress(object):
#
#     def __init__(self, *args):
#         super(suppress, self).__init__()
#         self.exceptions = args
#
#     def __enter__(self):
#         pass
#
#     def __exit__(self, exc_type, exc_val, exc_tb):
#         if exc_type in self.exceptions:
#             output = traceback.format_exception(exc_type, exc_val, exc_tb)
#             logging.error("\n".join(output))

@contextmanager
def suppress(*args):
    try:
        yield
    except args as e:
        if log_level <= logging.DEBUG:
            logging.exception(e)


def reboot_switch_snmp(a, b, c, d, **kwargs):
    logging.debug((a, b, c, d))
    logging.debug(kwargs)


def reboot_switch(**kwargs):
    with suppress(TypeError):
        return reboot_switch_snmp(**kwargs)


def main():
    reboot_switch(a=1)


if __name__ == "__main__":
    main()

# bad traceback, using @contextmanager supress
# Note how the traceback stops at the yield statement, it does not show the actual calling location

#ERROR:root:reboot_switch_snmp() takes exactly 4 arguments (1 given)
#Traceback (most recent call last):
#  File "test_supress.py", line 125, in suppress
#    yield
#  File "test_supress.py", line 138, in reboot_switch
#    return reboot_switch_snmp(**kwargs)
#TypeError: reboot_switch_snmp() takes exactly 4 arguments (1 given)


# good traceback using suppress class
# Note how the traceback show the calling location in main(), reboot_switch()

#ERROR:root:Traceback (most recent call last):
#
#  File "test_supress.py", line 138, in reboot_switch
#    return reboot_switch_snmp(**kwargs)
#
#TypeError: reboot_switch_snmp() takes exactly 4 arguments (1 given)
#
#Traceback (most recent call last):
#  File "test_supress.py", line 146, in <module>
#    main()
#  File "test_supress.py", line 142, in main
#    reboot_switch(a=1)
#  File "test_supress.py", line 138, in reboot_switch
#    return reboot_switch_snmp(**kwargs)
#TypeError: reboot_switch_snmp() takes exactly 4 arguments (1 given)
	#!/usr/bin/env python
	# Copyright (c) 2016-2017 Intel Corporation
	#
	# Licensed under the Apache License, Version 2.0 (the "License");
	# you may not use this file except in compliance with the License.
	# You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.

	from contextlib import contextmanager
	from functools import wraps
	import logging
	import traceback
	import sys

	log_level = logging.DEBUG

	logging.basicConfig(level=logging.DEBUG)

	class MyGeneratorContextManager(object):
	"""Helper for @contextmanager decorator."""

	def __init__(self, gen):
	self.gen = gen

	def __enter__(self):
	try:
	return self.gen.next()
	except StopIteration:
	raise RuntimeError("generator didn't yield")

	def __exit__(self, exc_type, exc_val, exc_tb):
	if exc_type is None:
	try:
	self.gen.next()
	except StopIteration:
	return
	else:
	raise RuntimeError("generator didn't stop")
	else:
	if exc_val is None:
	# Need to force instantiation so we can reliably
	# tell if we get the same exception back
	exc_val = exc_type()
	try:
	# output = traceback.format_exception(exc_type, exc_val, exc_tb)
	# output = traceback.format_exception(*sys.exc_info())
	# logging.error("\n".join(output))
	self.gen.throw(*sys.exc_info())
	# self.gen.throw(exc_type, exc_val, exc_tb)
	raise RuntimeError("generator didn't stop after throw()")
	except StopIteration, exc:
	# Suppress the exception unless it's the same exception that
	# was passed to throw(). This prevents a StopIteration
	# raised inside the "with" statement from being suppressed
	return exc is not exc_val
	except:
	# only re-raise if it's not the exception that was
	# passed to throw(), because __exit__() must not raise
	# an exception unless __exit__() itself failed. But throw()
	# has to raise the exception to signal propagation, so this
	# fixes the impedance mismatch between the throw() protocol
	# and the __exit__() protocol.
	#
	if sys.exc_info()[1] is not exc_val:
	raise


	def my_contextmanager(func):
	"""@contextmanager decorator.

	Typical usage:

	@contextmanager
	def some_generator(<arguments>):
	<setup>
	try:
	yield <value>
	finally:
	<cleanup>

	This makes this:

	with some_generator(<arguments>) as <variable>:
	<body>

	equivalent to this:

	<setup>
	try:
	<variable> = <value>
	<body>
	finally:
	<cleanup>

	"""
	@wraps(func)
	def helper(args, *kwds):
	return MyGeneratorContextManager(func(args, *kwds))
	return helper

	# class suppress(object):
	#
	# def __init__(self, *args):
	# super(suppress, self).__init__()
	# self.exceptions = args
	#
	# def __enter__(self):
	# pass
	#
	# def __exit__(self, exc_type, exc_val, exc_tb):
	# if exc_type in self.exceptions:
	# output = traceback.format_exception(exc_type, exc_val, exc_tb)
	# logging.error("\n".join(output))

	@contextmanager
	def suppress(*args):
	try:
	yield
	except args as e:
	if log_level <= logging.DEBUG:
	logging.exception(e)


	def reboot_switch_snmp(a, b, c, d, **kwargs):
	logging.debug((a, b, c, d))
	logging.debug(kwargs)


	def reboot_switch(**kwargs):
	with suppress(TypeError):
	return reboot_switch_snmp(**kwargs)


	def main():
	reboot_switch(a=1)


	if __name__ == "__main__":
	main()

	# bad traceback, using @contextmanager supress
	# Note how the traceback stops at the yield statement, it does not show the actual calling location

	#ERROR:root:reboot_switch_snmp() takes exactly 4 arguments (1 given)
	#Traceback (most recent call last):
	# File "test_supress.py", line 125, in suppress
	# yield
	# File "test_supress.py", line 138, in reboot_switch
	# return reboot_switch_snmp(**kwargs)
	#TypeError: reboot_switch_snmp() takes exactly 4 arguments (1 given)



	# good traceback using suppress class
	# Note how the traceback show the calling location in main(), reboot_switch()

	#ERROR:root:Traceback (most recent call last):
	#
	# File "test_supress.py", line 138, in reboot_switch
	# return reboot_switch_snmp(**kwargs)
	#
	#TypeError: reboot_switch_snmp() takes exactly 4 arguments (1 given)
	#
	#Traceback (most recent call last):
	# File "test_supress.py", line 146, in <module>
	# main()
	# File "test_supress.py", line 142, in main
	# reboot_switch(a=1)
	# File "test_supress.py", line 138, in reboot_switch
	# return reboot_switch_snmp(**kwargs)
	#TypeError: reboot_switch_snmp() takes exactly 4 arguments (1 given)