How to unittest a database wrapper?

Question

I've written a database wrapper for RethinkDB, in Python, a wrapper that introduces models (something similar to what Django delivers regarding models and managers). How do I write unittests for it? Actually, how do I test whether the database has been updated with the values I've given to a model?

I thought of actually querying the database, which indeed works, but this means I need to create a connection to the database (and also set up a database) for each tests run. Is there a way to mock a database or connection so I get this working?

Currently I create a Connection object in a test's setUp() method, create a database for the tests, and remove the above operations in the tearDown() method.

Answer 1

You can use unittest.mock to mock the low-level API you are using to implement the wrapper and then use asserts to check calls to the API from the wrapper.

I don't know much about django models or rethinkdb but it could look somethink like this.

import uuid import unittest import unittest.mock as mock

import wrapper # your wrapper


class Person(wrapper.Model):
    name = wrapper.CharField()

class Tests(unittest.TestCase):
    def setUp(self):
        # you can mock the whole low-level API module
        self.mock_r = mock.Mock()
        self.r_patcher = mock.patch.dict('rethinkdb', rethinkdb=self.mock_r):
        self.r_patcher.start()
        wrapper.connect('localhost', 1234, 'db')

    def tearDown(self):
        self.r_patcher.stop()

    def test_create(self):
        """Test successful document creation."""
        id = uuid.uuid4()
        # rethinkdb.table('persons').insert(...) will return this
        self.mock_r.table().insert.return_value = {
            "deleted": 0,
            "errors": 0,
            "generated_keys": [
                id
            ],
            "inserted": 1,
            "replaced": 0,
            "skipped": 0,
            "unchanged": 0
        }

        name = 'Smith'
        person = wrapper.Person(name=name)
        person.save()

        # checking for a call like rethinkdb.table('persons').insert(...)
        self.mock_r.table.assert_called_once_with('persons')
        expected = {'name': name}
        self.mock_r.table().insert.assert_called_once_with(expected)

        # checking the generated id
        self.assertEqual(person.id, id)

    def test_create_error(self):
        """Test error during document creation."""
        error_msg = "boom!"
        self.mock_r.table().insert.return_value = {
            "deleted": 0,
            "errors": 1,
            "first_error": error_msg
            "inserted": 0,
            "replaced": 0,
            "skipped": 0,
            "unchanged": 0
        }

        name = 'Smith'
        person = wrapper.Person(name=name)

        # expecting error
        with self.assertRaises(wrapper.Error) as error:
            person.save()
        # checking the error message
        self.assertEqual(str(error), error_msg)

This code is quite rough, but I hope you get the idea.

Edit: Added return_value and test for error.