On Tue, 13 Dec 2005 10:08:17 -0500
[EMAIL PROTECTED] wrote.

> On Tuesday 13 December 2005 09:54, Tadashi Matsumoto wrote:
>
> If you write a test for this, I'll check it in.
> 

I have made a test. It works well (perhaps).
This test program is small, so I have attached in this mail.

----
Tadashi Matsumoto
[EMAIL PROTECTED]
import os, shutil
import tempfile, threading
from unittest import TestCase, TestSuite, main, makeSuite

from zope.app.rdb.gadflyda import GadflyAdapter, setGadflyRoot

class GadflyTestBase(TestCase):
    
    def setUp(self):
        TestCase.setUp(self)
        self.tempdir = None
    
    def tearDown(self):
        TestCase.tearDown(self)
        if self.tempdir:
            shutil.rmtree(self.tempdir)
        setGadflyRoot()

    def getGadflyRoot(self):
        if not self.tempdir:
            self.tempdir = tempfile.mkdtemp('gadfly')
        setGadflyRoot(self.tempdir)
        return self.tempdir

    def _create(self, *args):
        return GadflyAdapter(*args)

def exec_sql(adapter, sql, args, fetch=False):
    
    conn = adapter()
    cur =conn.cursor()
    cur.execute(sql, args)
    rows = []
    if fetch:
        rows = cur.fetchall()
    conn.commit()
    return rows

class TestPhantom(GadflyTestBase):
    
    def setUp(self):
        GadflyTestBase.setUp(self)
        dir = self.getGadflyRoot()
        os.mkdir(os.path.join(dir, "demo"))
        self.adapter = self._create("dbi://demo")
        conn = self.adapter()
        cur = conn.cursor()
        cur.execute("create table t1 (name varchar)")
        conn.commit()

    def test_Phantom(self):
        
        adapter = self.adapter
        insert = "insert into t1 values (?)"
        select = "select name from t1"
        delete = "delete from t1"

        count = 0
        for name in ('a', 'b', 'c'):
            t = threading.Thread(target=exec_sql,
                                 args=(adapter, insert, (name,)))
            t.start()
            t.join()
            rows = exec_sql(adapter, select, args=(), fetch=True)
            count += 1 
            self.assertEqual(len(rows), count)
        
        exec_sql(adapter, delete, args=())
        t = threading.Thread(target=exec_sql,
                             args=(adapter, delete, ()))
        t.start()
        t.join()
        rows = exec_sql(adapter, select, args=(), fetch=True)
        self.assertEqual(len(rows), 0)                                     

def test_suite():
    return TestSuite((
        makeSuite(TestPhantom),
        ))

if __name__=='__main__':
    
    main(defaultTest='test_suite')
          
_______________________________________________
Zope3-dev mailing list
Zope3-dev@zope.org
Unsub: http://mail.zope.org/mailman/options/zope3-dev/archive%40mail-archive.com

Reply via email to