##############################################################################
#
# Copyright (c) 2004 Enfold Systems LLC. All rights reserved.
# Copyright (c) 2005-2006 Brian Sutherland. All rights reserved.
#
# This software is distributed under the terms of the Zope Public
# License (ZPL) v2.1. See COPYING.txt for more information.
#
##############################################################################
"""
$Id$
"""

import unittest
import threading

from transaction import get, begin
from zope.app.testing.functional import BrowserTestCase
import zope.component
from zope.rdb.interfaces import IZopeDatabaseAdapter

from sqlos.interfaces import IConnectionName
from sqlos.testing.sampleperson import SamplePerson, SampleDog

__metaclass__ = type

class TestTransaction(BrowserTestCase):

    def setUp(self):
        super(TestTransaction, self).setUp()
        SamplePerson.createTable(ifNotExists=True)
        SampleDog.createTable(ifNotExists=True)
        person = SamplePerson(fullname='Sidnei da Silva',
                              username='sidnei',
                              password='test')
        self.personid = person.id
        # Commit what we have done
        get().commit()
        begin()

    def supportTransactions(self):
        return SamplePerson._connection.supportTransactions

    def testChange(self):
        person = SamplePerson.get(self.personid)
        self.assertEqual(person.fullname, 'Sidnei da Silva')
        self.assertEqual(person.username, 'sidnei')
        self.assertEqual(person.password, 'test')
        person.fullname = 'Sidnei Silva'
        person.username = 'dreamcatcher'
        person.password = 'pass'
        person.syncUpdate() # XXX - This sync should not be necessary but is
                            # unless _cacheValues=False is removed from
                            # SamplePerson - jinty
        self.assertEqual(person.fullname, 'Sidnei Silva')
        self.assertEqual(person.username, 'dreamcatcher')
        self.assertEqual(person.password, 'pass')

    def testCommit(self):
        person = SamplePerson.get(self.personid)
        self.assertEqual(person.fullname, 'Sidnei da Silva')
        self.assertEqual(person.username, 'sidnei')
        self.assertEqual(person.password, 'test')
        person.fullname = 'Sidnei Silva'
        person.username = 'dreamcatcher'
        person.password = 'pass'
        get().commit()
        begin()
        person = SamplePerson.get(self.personid)
        self.assertEqual(person.fullname, 'Sidnei Silva')
        self.assertEqual(person.username, 'dreamcatcher')
        self.assertEqual(person.password, 'pass')

    def testAbort(self):
        person = SamplePerson.get(self.personid)
        self.assertEqual(person.fullname, 'Sidnei da Silva')
        self.assertEqual(person.username, 'sidnei')
        self.assertEqual(person.password, 'test')
        person.fullname = 'Sidnei Silva'
        person.username = 'dreamcatcher'
        person.password = 'pass'
        assert person.dirty is True
        person.sync() # Sync to ensure that the DB is sent the SQL statements
        get().abort()
        begin()
        person = SamplePerson.get(self.personid)
        if self.supportTransactions():
            self.assertEqual(person.fullname, 'Sidnei da Silva')
            self.assertEqual(person.username, 'sidnei')
            self.assertEqual(person.password, 'test')
            # commit the next transaction so we can paranoically check
            get().commit()
            begin()
            person = SamplePerson.get(self.personid)
            self.assertEqual(person.fullname, 'Sidnei da Silva')
        else:
            # ya well no fine
            import warnings
            warnings.warn('SQLObject connections for this database do not '
                          'support transactions. Not testing if transaction '
                          'abort works.')
            self.assertEqual(person.fullname, 'Sidnei Silva')
            self.assertEqual(person.username, 'dreamcatcher')
            self.assertEqual(person.password, 'pass')

    def testAbortAndCommitDirty(self):
        """Test the commit after an abort in the presence of dirty SQLObjects.

        Here we commit the next transaction with another dirty object
        just to make sure that remnants from the previous transaction are
        not committed.
        Yeah, this is paranoid, but sometimes it pays to be paranoid.
        """
        if self.supportTransactions():
            # make a dirty object
            person = SamplePerson.get(self.personid)
            self.assertEqual(person.fullname, 'Sidnei da Silva')
            person.fullname = 'Sidnei Silva'
            assert person.dirty is True
            person.sync() # Sync to ensure that the DB is sent the SQL
            # abort transaction and start a new one
            get().abort()
            begin()
            # make another dirty object
            person = SamplePerson.get(self.personid)
            self.assertEqual(person.fullname, 'Sidnei da Silva')
            brian = SamplePerson(fullname='Brian Sutherland',
                                 username='brian',
                                 password='test')
            brian.fullname = "B. Sutherland" # make the object dirty
            assert brian.dirty is True
            brianid = brian.id
            # commit the second transaction
            get().commit()
            begin()
            person = SamplePerson.get(self.personid)
            brian = SamplePerson.get(brianid)
            self.assertEqual(person.fullname, 'Sidnei da Silva')
            self.assertEqual(brian.fullname, 'B. Sutherland')
        else:
            # ya well no fine
            import warnings
            warnings.warn('SQLObject connections for this database do not '
                          'support transactions. Not testing if transaction '
                          'abort works.')

    def testCacheThreadIsolation(self):
        """Tests that the changes we make in one thread don't appear in another.

        This is a regression test for if the cache makes breaks the isolation
        between threads.
        """
        ut = zope.component.getUtility(IConnectionName)
        adapter = zope.component.queryUtility(IZopeDatabaseAdapter, ut.name)
        if adapter.getDSN() == 'dbi://:memory:':
            import warnings
            warnings.warn('Warning, not testing Cache Isolation')
            return # this test is NOT going to work against the default test
                   # database
        # warm up the cache by getting an instance
        person2 = SamplePerson.get(self.personid)
        self.assertEqual(person2.fullname, 'Sidnei da Silva')
        self.assertEqual(person2.username, 'sidnei')
        self.assertEqual(person2.password, 'test')
        def changePerson():
            person1 = SamplePerson.get(self.personid)
            person1.fullname = 'Another Person'
            person1.username = 'notsidnei'
            person1.password = 'nottest'
        thread = threading.Thread(target=changePerson)
        thread.start()
        thread.join()
        self.assertEqual(person2.fullname, 'Sidnei da Silva')
        self.assertEqual(person2.username, 'sidnei')
        self.assertEqual(person2.password, 'test')

    def testCleanCacheOnOverTransaction(self):
        """Test that the cache is being cleared on every transaction.

        We test this in a different thread to the main one to make sure that
        whatever needs to happen to clean out the cache over transactions will
        happen even if the thread is not the main thread. (Just in case some
        dodo only registers the cache clearer in the main thread.)
        """
        ut = zope.component.getUtility(IConnectionName)
        adapter = zope.component.queryUtility(IZopeDatabaseAdapter, ut.name)
        if adapter.getDSN() == 'dbi://:memory:':
            import warnings
            warnings.warn('Warning, not testing Cache Isolation')
            return # this test is NOT going to work against the default test
                   # database
        log = {}
        def testCache():
            # warm up the cache by getting an instance
            person2 = SamplePerson.get(self.personid)
            self.assertEqual(person2.fullname, 'Sidnei da Silva')
            self.assertEqual(person2.username, 'sidnei')
            self.assertEqual(person2.password, 'test')
            # comit this transaction "just in case"
            get().commit()
            # get change and commit the person in another thread
            def changePerson():
                person1 = SamplePerson.get(self.personid)
                person1.fullname = 'Another Person'
                person1.username = 'notsidnei'
                person1.password = 'nottest'
                get().commit()
            thread = threading.Thread(target=changePerson)
            thread.start()
            thread.join()
            # start a new transaction in this thread
            get().commit()
            begin()
            # If we get a new person, we should see the changed person
            person3 = SamplePerson.get(self.personid)
            log['fullname'] = person3.fullname
            log['username'] = person3.username
            log['password'] = person3.password
        thread = threading.Thread(target=testCache)
        thread.start()
        thread.join()
        self.assertEqual(log['fullname'], 'Another Person')
        self.assertEqual(log['username'], 'notsidnei')
        self.assertEqual(log['password'], 'nottest')

    def tearDown(self):
        person = SamplePerson.get(self.personid)
        person.destroySelf()
        super(TestTransaction, self).tearDown()
        SamplePerson.dropTable()
        get().commit()
        begin()

def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(TestTransaction))
    return suite


if __name__ == '__main__':
    unittest.main()