test_models.py

from twisted.trial import unittest
from twisted.internet import defer
from storm.twisted.transact import transact, Transactor

from storm.twisted.testing import FakeThreadPool

class Antani(object):

  @transact
  def antani(self):
      d = defer.Deferred()
      d.success('somestuff')
      return d

class TestAntani(unittest.TestCase):
    def setUp(self):
        self.threadpool = FakeThreadPool()
        self.transactor = Transactor(self.threadpool)

    def test_antani(self):
        def cb(result):
            # This result is actually a deferred
            # so you would have to do
            def cb2(result2):
                self.assertEqual(result2, "somestuff")

            result.addCallback(cb2)
            self.assertEqual(result, "foobar")

        def eb(failure):
            print "errback"
            print failure

        a = Antani()
        a.transactor = self.transactor
        d = a.antani()
        d.addCallback(cb) # never called
        d.addErrback(eb)   # never called
        return d