nilmdb/tests/test_mustclose.py

import nilmdb
from nilmdb.utils.printf import *

import nose
from nose.tools import *
from nose.tools import assert_raises

from testutil.helpers import *

import sys
import io
import gc

import inspect

err = io.StringIO()

@nilmdb.utils.must_close(errorfile = err)
class Foo:
    def __init__(self, arg):
        fprintf(err, "Init %s\n", arg)

    def __del__(self):
        fprintf(err, "Deleting\n")

    def close(self):
        fprintf(err, "Closing\n")

@nilmdb.utils.must_close(errorfile = err, wrap_verify = True)
class Bar:
    def __init__(self):
        fprintf(err, "Init\n")

    def __del__(self):
        fprintf(err, "Deleting\n")

    @classmethod
    def baz(self):
        fprintf(err, "Baz\n")

    def close(self):
        fprintf(err, "Closing\n")

    def blah(self, arg):
        fprintf(err, "Blah %s\n", arg)

@nilmdb.utils.must_close(errorfile = err)
class Baz:
    pass

class TestMustClose(object):
    def test(self):

        # Note: this test might fail if the Python interpreter doesn't
        # garbage collect the object (and call its __del__ function)
        # right after a "del x".

        # Trigger error
        err.truncate()
        x = Foo("hi")
        # Verify that the arg spec was maintained
        eq_(inspect.getargspec(x.__init__),
            inspect.ArgSpec(args = ['self', 'arg'],
                            varargs = None, keywords = None, defaults = None))
        del x
        gc.collect()
        eq_(err.getvalue(),
            "Init hi\n"
            "error: Foo.close() wasn't called!\n"
            "Deleting\n")

        # No error
        err.truncate(0)
        err.seek(0)
        y = Foo("bye")
        y.close()
        del y
        gc.collect()
        eq_(err.getvalue(),
            "Init bye\n"
            "Closing\n"
            "Deleting\n")

        # Verify function calls when wrap_verify is True
        err.truncate(0)
        err.seek(0)
        z = Bar()
        eq_(inspect.getargspec(z.blah),
            inspect.ArgSpec(args = ['self', 'arg'],
                            varargs = None, keywords = None, defaults = None))
        z.blah("boo")
        z.close()
        with assert_raises(AssertionError) as e:
            z.blah("hello")
        in_("called <function Bar.blah at 0x", str(e.exception))
        in_("> after close", str(e.exception))
        # Since the most recent assertion references 'z',
        # we need to raise another assertion here so that
        # 'z' will get properly deleted.
        with assert_raises(AssertionError):
            raise AssertionError()
        del z
        gc.collect()
        eq_(err.getvalue(),
            "Init\n"
            "Blah boo\n"
            "Closing\n"
            "Deleting\n")

        # Class with missing methods
        err.truncate(0)
        err.seek(0)
        w = Baz()
        w.close()
        del w
        eq_(err.getvalue(), "")

        # Test errors during __del__, by closing stderr so the fprintf fails
        r = Foo("hi")
        err.close()
        del r