Skip to content

Commit

Permalink
Improved the Gaphor transaction unit tests.
Browse files Browse the repository at this point in the history
  • Loading branch information
adamboduch committed Apr 25, 2011
1 parent d0999ba commit 7de78f3
Showing 1 changed file with 82 additions and 53 deletions.
135 changes: 82 additions & 53 deletions gaphor/tests/test_transaction.py
@@ -1,6 +1,9 @@
"""Unit tests for transactions in Gaphor."""


from unittest import TestCase from unittest import TestCase

from zope.component.globalregistry import base from zope.component.globalregistry import base

from gaphor.application import Application from gaphor.application import Application
from gaphor.transaction import Transaction, transactional, TransactionError from gaphor.transaction import Transaction, transactional, TransactionError
from gaphor.event import TransactionBegin, TransactionCommit, TransactionRollback from gaphor.event import TransactionBegin, TransactionCommit, TransactionRollback
Expand All @@ -9,126 +12,152 @@
commits = [] commits = []
rollbacks = [] rollbacks = []


def handle_begins(ev): def handle_begins(event):
begins.append(ev) """Store TransactionBegin events in begins."""
begins.append(event)


def handle_commits(ev): def handle_commits(event):
commits.append(ev) """Store TransactionCommit events in commits."""

commits.append(event)
def handle_rollback(ev):
rollbacks.append(ev)


def handle_rollback(event):
"""Store TransactionRollback events in rollbacks."""
rollbacks.append(event)


class TransactionTestCase(TestCase): class TransactionTestCase(TestCase):
"""Test case for transactions with the component registry
enabled."""


def setUp(self): def setUp(self):
"""Initialize Gaphor services and register transaction event
handlers."""

Application.init(services=['component_registry']) Application.init(services=['component_registry'])

component_registry = Application.get_service('component_registry') component_registry = Application.get_service('component_registry')

component_registry.register_handler(handle_begins, [TransactionBegin]) component_registry.register_handler(handle_begins, [TransactionBegin])
component_registry.register_handler(handle_commits, [TransactionCommit]) component_registry.register_handler(handle_commits, [TransactionCommit])
component_registry.register_handler(handle_rollback, [TransactionRollback]) component_registry.register_handler(handle_rollback, [TransactionRollback])

del begins[:] del begins[:]
del commits[:] del commits[:]
del rollbacks[:] del rollbacks[:]
assert not begins
assert not commits
assert not rollbacks


def tearDown(self): def tearDown(self):
"""Finished with the test case. Unregister event handlers that
store transaction events."""

component_registry = Application.get_service('component_registry') component_registry = Application.get_service('component_registry')

component_registry.unregister_handler(handle_begins, [TransactionBegin]) component_registry.unregister_handler(handle_begins, [TransactionBegin])
component_registry.unregister_handler(handle_commits, [TransactionCommit]) component_registry.unregister_handler(handle_commits, [TransactionCommit])
component_registry.unregister_handler(handle_rollback, [TransactionRollback]) component_registry.unregister_handler(handle_rollback, [TransactionRollback])




def test_transaction_commit(self): def test_transaction_commit(self):
"""Test committing a transaction."""

tx = Transaction() tx = Transaction()
self.assertTrue(tx._stack)
self.assertEquals(1, len(begins)) self.assertTrue(tx._stack, 'Transaction has no stack')
self.assertEquals(0, len(commits)) self.assertEquals(1, len(begins), 'Incorrect number of TrasactionBegin events')
self.assertEquals(0, len(rollbacks)) self.assertEquals(0, len(commits), 'Incorrect number of TransactionCommit events')
self.assertEquals(0, len(rollbacks), 'Incorrect number of TransactionRollback events')


tx.commit() tx.commit()
self.assertEquals(1, len(begins))
self.assertEquals(1, len(commits)) self.assertEquals(1, len(begins), 'Incorrect number of TrasactionBegin events')
self.assertEquals(0, len(rollbacks)) self.assertEquals(1, len(commits), 'Incorrect number of TransactionCommit events')

self.assertEquals(0, len(rollbacks), 'Incorrect number of TransactionRollback events')
self.assertFalse(tx._stack) self.assertFalse(tx._stack, 'Transaction stack is not empty')


try: try:
tx.commit() tx.commit()
except TransactionError: except TransactionError:
pass # ok pass
else: else:
assert False, 'should not be reached' self.fail('Commit should not have succeeded')




def test_transaction_rollback(self): def test_transaction_rollback(self):
"""Test rolling back a transaction."""

tx = Transaction() tx = Transaction()
self.assertTrue(tx._stack)
self.assertEquals(1, len(begins)) self.assertTrue(tx._stack, 'Transaction has no stack')
self.assertEquals(0, len(commits)) self.assertEquals(1, len(begins), 'Incorrect number of TrasactionBegin events')
self.assertEquals(0, len(rollbacks)) self.assertEquals(0, len(commits), 'Incorrect number of TransactionCommit events')
self.assertEquals(0, len(rollbacks), 'Incorrect number of TransactionRollback events')


tx.rollback() tx.rollback()
self.assertEquals(1, len(begins))
self.assertEquals(0, len(commits)) self.assertEquals(1, len(begins), 'Incorrect number of TrasactionBegin events')
self.assertEquals(1, len(rollbacks)) self.assertEquals(0, len(commits), 'Incorrect number of TransactionCommit events')
self.assertEquals(1, len(rollbacks), 'Incorrect number of TransactionRollback events')


self.assertFalse(tx._stack) self.assertFalse(tx._stack, 'Transaction stack is not empty')




def test_transaction_commit_after_rollback(self): def test_transaction_commit_after_rollback(self):
tx = Transaction() """Test committing one transaction after rolling back another
transaction."""

tx1 = Transaction()
tx2 = Transaction() tx2 = Transaction()


tx2.rollback() tx2.rollback()

tx1.commit()
tx.commit()
self.assertEquals(1, len(begins)) self.assertEquals(1, len(begins), 'Incorrect number of TrasactionBegin events')
self.assertEquals(0, len(commits)) self.assertEquals(0, len(commits), 'Incorrect number of TransactionCommit events')
self.assertEquals(1, len(rollbacks)) self.assertEquals(1, len(rollbacks), 'Incorrect number of TransactionRollback events')




def test_transaction_stack(self): def test_transaction_stack(self):
tx = Transaction() """Test the transaction stack."""

tx1 = Transaction()
tx2 = Transaction() tx2 = Transaction()


try: try:
tx.commit() tx1.commit()
except TransactionError, e: except TransactionError, e:
self.assertEquals('Transaction on stack is not the transaction being closed.', str(e)) pass
else: else:
assert False, 'should not be reached' self.fail('Commit should not have succeeded')




def test_transaction_context(self): def test_transaction_context(self):
"""Test the transaction context manager."""

with Transaction() as tx: with Transaction() as tx:
self.assertTrue(isinstance(tx, Transaction))
self.assertTrue(Transaction._stack) self.assertTrue(isinstance(tx, Transaction), 'Context is not a Transaction instance')
self.assertFalse(Transaction._stack) self.assertTrue(Transaction._stack, 'Transaction instance has no stack inside a context')

self.assertFalse(Transaction._stack, 'Transaction stack should be empty')




def test_transaction_context_error(self): def test_transaction_context_error(self):
"""Test the transaction context manager with errors."""

try: try:
with Transaction(): with Transaction():
self.assertTrue(Transaction._stack) raise TypeError('transaction error')
raise TypeError('some error')
except TypeError, e: except TypeError, e:
self.assertEquals('some error', str(e)) self.assertEquals('transaction error', str(e), 'Transaction context manager did no raise correct exception')
self.assertFalse(Transaction._stack)
else: else:
self.assertFalse(Transaction._stack) self.fail('Transaction context manager did not raise exception when it should have')
assert False, 'should not be reached'

class TransactionWithoutComponentRegistryTestCase(TestCase): class TransactionWithoutComponentRegistryTestCase(TestCase):
"""Test case for transactions with no component registry."""


def test_transaction(self): def test_transaction(self):
"""Test basic transaction functionality."""

tx = Transaction() tx = Transaction()
tx.rollback() tx.rollback()


tx = Transaction() tx = Transaction()
tx.commit() tx.commit()

# vim:sw=4:et:ai

0 comments on commit 7de78f3

Please sign in to comment.