Testcase helper examples

testcase_helpers.py

"""
Testcase Helpers (pending a better name) make it easy to compose elements
which require setup/teardown on each testcase.

For example, consider a testcase which needs to clean a Redis database after
each run::

    class TestRedis(TestCase):
        def cleanup_redis(self):
            for key in self.cxn.keys("*"):
                self.cxn.delete(key)

        def setUp(self):
            self.cxn = get_redis_connection()
            self.cleanup_redis()

        def teardown(self):
            self.cleanup_redis()
            self.cxn.close()
            self.cxn = None

        def test_set(self):
            self.cxn.set("foo", "42")
            assert self.cxn.get("foo") == "42"

This would be fine if there was only one test class that uses Redis… but as
more and more test classes need Redis, one of three things will happen:

    (1) The setup/teardown code will be copy + pasted between test classes.
    (2) All testcases will inherit from one base class that does setup and
        teardown.
    (3) There will be a RedisTestMixin created that test classes can inherit
        from.

The problems with (1) are obvious.

(2) may be acceptable in some environments, but becomes less desirable as the
number of components which require setup + teardown grows.

(3) leads to annoying super(...) related bugs, makes individual components
harder (or, at least, less obvious) to configure, and large inheritance
lists can get unwieldy.

This is where Testcase Helpers (again, working title) come in: they make it
simple to compose setup + teardown behavior by encapsulating it in a class::

    class RedisHelper(TestCaseHelper):
        def cleanup(self):
            for key in self.cxn.keys("*"):
                self.cxn.delete(key)

        def setup(self):
            self.cxn = get_redis_connection()
            self.cleanup()

        def teardown(self):
            self.cleanup()
            self.cxn.close()
            self.cxn = None

Then adding an instance of that class to the test class::

    class RedisTest(TestCase):
        redis = RedisHelper()

        def test_set(self):
            self.redis.cxn.set("foo", "42")
            assert self.redis.cxn.get("foo") == "42"

You can start to see how this makes composition simpler, cleaner, and more
obvious, and it becomes really clear once slightly more complex helpers and
test cases are used::

    class MyTestCase(TestCase):
        mock = MockHelper({
            "myproj.MyObj.get_foo": {"return_value": 42},
        })
        mail = EmailHelper()
        responses = ResponsesHelper()

        def test_stuff(self):
            obj = MyObj()
            assert obj.get_foo() == 42

            obj.send_mail()
            self.mail.assert_mailed(to="[email protected]", body="hello, world")

            self.responses.add("GET", "/stuff", body="some-value")
            assert obj.get_stuff() == "some-value"

"""

class TestCaseHelper(object):
    def get_settings(self):
        return {}

    def setup(self):
        pass

    def teardown(self):
        pass


class HelperMixin(object):
    def __init__(self, *args, **kwargs):
        self.init_helpers()
        super(HelperMixin, self).__init__(*args, **kwargs)

    def setUp(self):
        self.setup_helpers()

    def tearDown(self):
        self.teardown_helpers()

    def init_helpers(self):
        self._helpers = []
        for attr in dir(self):
            val = getattr(self, attr)
            if isinstance(val, TestCaseHelper):
                self._helpers.append(val)

    def get_helper_settings(self):
        helper_settings = {}
        for helper in self._helpers:
            helper_settings.update(helper.get_settings())
        return helper_settings

    def setup_helpers(self):
        for helper in self._helpers:
            helper.setup()

    def teardown_helpers(self):
        for helper in reversed(self._helpers):
            helper.teardown()


class EmailHelper(TestCaseHelper):
    @property
    def outbox(self):
        return mail.outbox

    def assert_mailed(self, **kwargs):
        for message in self.outbox:
            try:
                self.assert_message_matches(message, **kwargs)
                return
            except AssertionError:
                pass
        raise AssertionError(
            "No emails (of %s sent) match %s: %s" %(
                len(self.outbox), kwargs,
                [{"subject": m.subject, "to": m.to} for m in self.outbox],
            )
        )

    def assert_not_mailed(self, **kwargs):
        for message in self.outbox:
            try:
                self.assert_message_matches(message, **kwargs)
                raise AssertionError("Message matches %s: %s" %(kwargs, message))
            except AssertionError:
                pass

    def assert_message_matches(self, msg, to=None, subject=None, body=None):
        _assert_field_contains(to, msg.to, "To")
        _assert_field_contains(subject, msg.subject, "Subject")
        _assert_field_contains(body, msg.body, "Body")


class DjangoCache(TestCaseHelper):
    def get_settings(self):
        locmem = {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        }
        cur_caches = dict(settings.CACHES)
        cur_caches.update({
            "default": locmem,
            "locmem": locmem,
        })
        return {
            "CACHES": cur_caches,
        }

    def setup(self):
        self.backend = cache.cache
        self.old_backend = self.backend.get_backend()
        self.backend.set_backend("locmem")
        self.clear()

    def teardown(self):
        self.backend.set_backend(self.old_backend)

    def clear(self):
        self.backend.clear()

import redis

class RedisHelper(TestCaseHelper):
    def __init__(self, kp="tc:"):
        self.kp = kp

    def setup(self):
        from akindi import redis
        self.cxn = redis.get_connection_raw()
        self.clear()

    def teardown(self):
        self.clear()
        self.cxn = None

    def clear(self):
        keys = self.cxn.keys(self.kp + "*")
        if keys:
            self.cxn.delete(*keys)

import mock

class MockHelper(TestCaseHelper):
    """ Example::

        mock = MockHelper({
            "foo.bar.Baz": {"return_value": 42},
            "stuff=foo.bar.AnotherThing": {"some_attr": 42},
        })

        ...

        self.mock.Baz.assert_called_with(16)
        assert_equal(self.mock.stuff.some_attr, 42)
    """

    def __init__(self, to_mock):
        self._to_mock = to_mock

    def setup(self):
        self._patches = {}
        for to_patch, attrs in self._to_mock.items():
            if "=" in to_patch:
                name, _, to_patch = to_patch.partition("=")
            else:
                name = to_patch.split(".")[-1]
            if name in self.__dict__:
                raise AssertionError("%r already has attribute %r"
                                     %(self, name))
            p = mock.patch(to_patch)
            m = p.__enter__()
            if attrs:
                m.configure_mock(**attrs)
            self._patches[name] = p
            self.__dict__[name] = m

    def teardown(self):
        for name, p in self._patches.items():
            p.__exit__(None, None, None)
            self.__dict__.pop(name)
        self._patches = None


class GeventHelper(TestCaseHelper):
    def setup(self):
        self.threads = []

    def spawn(self, *a, **kw):
        import gevent
        thread = gevent.spawn(*a, **kw)
        self.threads.append(thread)
        return thread

    def join(self, timeout=0.5):
        for thread in self.threads:
            thread.get(timeout=timeout)

    def teardown(self):
        for thread in self.threads:
            if thread:
                thread.kill()

from urllib3_mock import Responses

class ResponsesHelper(TestCaseHelper):
    _responses = [
        Responses("urllib3"),
        Responses("requests.packages.urllib3"),
    ]

    def _call(self, method, *args, **kwargs):
        return [
            getattr(r, method)(*args, **kwargs)
            for r in self._responses
        ]

    def setup(self):
        self._call("__enter__")

    def teardown(self):
        self._call("__exit__", None, None, None)

    def add(self, method, url, **kwargs):
        """ Examples:

            responses.add("POST", "/foo", body={"foo": 42})
            responses.add("GET", "/bar", callback=lambda request: "stuff")
        """
        if isinstance(url, basestring) and any(url.startswith(x) for x in ["http://", "https://"]):
            url = "/" + "/".join(url.split("/")[3:])
        if "body" in kwargs:
            body = kwargs["body"]
            if isinstance(body, (dict, list, int, float)):
                kwargs["body"] = json.dumps(body)
                kwargs["content_type"] = "application/json"

        func = "add"
        if "callback" in kwargs:
            func = "add_callback"

        self._call(func, method, url, **kwargs)

    @property
    def calls(self):
        res = []
        for r in self._responses:
            res.extend(r.calls)
        return res


class ContextManagerHelper(TestCaseHelper):
    def __init__(self, context_manager):
        self.context_manager = context_manager

    def setup(self):
        self.context_manager.__enter__()

    def teardown(self):
        self.context_manager.__exit__(None, None, None)