Source

django-denormalize / denormalize / tests / backend.py

from pprint import pprint
import os
from functools import partial

from ..models import *
from ..backend.locmem import LocMemBackend
from ..context import delay_sync, sync_together

from django.conf import settings
from unittest import skipIf

import logging
log = logging.getLogger(__name__)

# TODO: add rollback test

class BackendTestMixin(object):
    SUPPORTS_SYNC_COLLECTION = True

    @classmethod
    def _create_backend(cls):
        raise NotImplementedError()

    @classmethod
    def setUpClass(cls):
        cls.create_test_tables()
        cls.bookcol = cls.collection()
        cls.backend = cls._create_backend()
        cls.backend.register(cls.bookcol)
        with sync_together():
            cls.models.create_test_data()

    @classmethod
    def tearDownClass(cls):
        cls.backend.deregister_all()
        cls.delete_test_tables()

    def get_book_doc(self, id):
        return self.backend.get_doc(self.bookcol, id)

    def test1_dump(self):
        # Test data
        doc = self.get_book_doc(1)
        self.assertEqual(doc['id'], 1)
        self.assertIn('chapter_set', doc)
        self.assertIn('extra_info', doc)
        self.assertIn('tags', doc)

        # Change a one to many link
        log.info("update book chapter (one-to-many)")
        self.update_chapter()
        doc = self.get_book_doc(1)
        chapter = doc['chapter_set'][0]
        self.assertEqual(chapter['id'], 1)
        self.assertTrue(chapter['title'].endswith('!!!'), chapter)

        # Change something (m2m)
        log.info("update book author (many-to-many)")
        self.update_author()
        doc = self.get_book_doc(1)
        self.assertEqual(doc['authors'][0]['name'], 'Another Name')

        # Change something that's shared (m2m)
        log.info("update book/publisher tag (multiple many-to-many)")
        self.update_tag()
        doc = self.get_book_doc(1)
        self.assertIn('tech', doc['tags'])

        # Add a chapter to a book
        log.info("insert chapter (one-to-many)")
        book1, chapter = self.insert_chapter()
        doc = self.get_book_doc(1)
        self.assertIn("Conclusion", [x['title'] for x in doc['chapter_set']])
        self.assertEqual(chapter.book, book1)

        # Move a chapter to another book (FK change!)
        log.info("move chapter")
        book2 = self.move_chapter(chapter)
        doc = self.get_book_doc(1)
        self.assertNotIn("Conclusion", [x['title'] for x in doc['chapter_set']])
        doc = self.get_book_doc(2)
        self.assertIn("Conclusion", [x['title'] for x in doc['chapter_set']])

        # Delete the chapter
        log.info("delete chapter (one-to-many)")
        self.delete_chapter(chapter)
        doc = self.get_book_doc(2)
        self.assertNotIn("Conclusion", [x['title'] for x in doc['chapter_set']])

        # Add a tag (m2m updated!!!) FIXME: not supported yet
        log.info("add tag to book (many-to-many)")
        tag = self.add_tag(book2)

        # Remove the tag from the book from the other side (reverse=True)
        log.info("remove tag from book (reverse many-to-many)")
        self.remove_tag_from_reverse(tag, book2)

        if self.SUPPORTS_SYNC_COLLECTION:
            # TODO: This code is fairly LocMemBackend specific, we need a
            #       cleaner way of testing.

            # Finally, test syncing the collection. First without special events.
            self.backend.sync_collection(self.bookcol)

            # Next, with dirty records
            inject_dirty = partial(self.inject_dirty, self.backend, book2)
            self.backend._sync_collection_before_handling_dirty = inject_dirty
            self.backend.sync_collection(self.bookcol)

            # Check that we ended up with a new book, even though it was added
            # during the full sync.
            self.assertEqual(len(self.backend.data['books']), 3)

class LocMemBackendTestMixin(BackendTestMixin):
    @classmethod
    def _create_backend(cls):
        return LocMemBackend()


# The MongoDB database to use for tests (required)
TEST_MONGO_DB = getattr(settings, 'DENORMALIZE_TEST_MONGO_DB', None)
# Optional, defaults to localhost
TEST_MONGO_URI = getattr(settings, 'DENORMALIZE_TEST_MONGO_URI', None)

@skipIf(not TEST_MONGO_DB, "settings.DENORMALIZE_TEST_MONGO_DB is not set!")
class MongoBackendMixin(BackendTestMixin):
    SUPPORTS_SYNC_COLLECTION = False # FIXME

    @classmethod
    def _create_backend(cls):
        from ..backend.mongodb import MongoBackend
        backend = MongoBackend(db_name=TEST_MONGO_DB,
                               connection_uri=TEST_MONGO_URI)

        log.info("dropping mongo database: %s" % TEST_MONGO_DB)
        connection = backend.connection
        connection.drop_database(TEST_MONGO_DB)
        return backend