63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
# -*- 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)
|