wutta-continuum/tests/test_util.py

# -*- coding: utf-8; -*-

from unittest import TestCase

import sqlalchemy_continuum as continuum

from wutta_continuum import util as mod
from wutta_continuum.testing import VersionTestCase


class TestRenderOperationType(TestCase):

    def test_basic(self):
        self.assertEqual(
            mod.render_operation_type(continuum.Operation.INSERT), "INSERT"
        )
        self.assertEqual(
            mod.render_operation_type(continuum.Operation.UPDATE), "UPDATE"
        )
        self.assertEqual(
            mod.render_operation_type(continuum.Operation.DELETE), "DELETE"
        )


class TestModelTransactionQuery(VersionTestCase):

    def test_inner_join(self):
        model = self.app.model

        user = model.User(username="fred")
        self.session.add(user)
        self.session.commit()

        query = mod.model_transaction_query(user)
        self.assertEqual(query.count(), 1)
        txn = query.one()

        UserVersion = continuum.version_class(model.User)
        version = self.session.query(UserVersion).one()
        self.assertIs(version.transaction, txn)

    def test_outer_joins(self):
        model = self.app.model

        person = model.Person(full_name="Fred Flintstone")
        self.session.add(person)
        user = model.User(username="fred", person=person)
        self.session.add(user)
        self.session.commit()

        query = mod.model_transaction_query(
            user, joins=[(model.Person, "uuid", "person_uuid")]
        )
        self.assertEqual(query.count(), 1)
        txn = query.one()

        vercls_user = continuum.version_class(model.User)
        version = self.session.query(vercls_user).one()
        self.assertIs(version.transaction, txn)

        vercls_person = continuum.version_class(model.Person)
        version = self.session.query(vercls_person).one()
        self.assertIs(version.transaction, txn)