From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Wed, 1 Feb 2006 23:27:36 +0000 (+0000)
Subject: unit tests for proxy engine
X-Git-Tag: rel_0_1_0~80
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=581d1dbdef46eb9de6e4f56530ca7ef2ddd27ff0;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

unit tests for proxy engine
---

diff --git a/test/proxy_engine.py b/test/proxy_engine.py
new file mode 100644
index 0000000000..7d508481e5
--- /dev/null
+++ b/test/proxy_engine.py
@@ -0,0 +1,240 @@
+from sqlalchemy import *
+from sqlalchemy.ext.proxy import ProxyEngine
+
+from testbase import PersistTest
+import testbase
+
+#
+# Define an engine, table and mapper at the module level, to show that the
+# table and mapper can be used with different real engines in multiple threads
+#
+module_engine = ProxyEngine()
+
+users = Table('users', module_engine, 
+              Column('user_id', Integer, primary_key=True),
+              Column('user_name', String(16)),
+              Column('password', String(20))
+              )
+
+class User(object):
+    pass
+
+assign_mapper(User, users)
+
+class ProxyEngineTest(PersistTest):
+
+    def setUp(self):
+        objectstore.clear()
+        
+    def test_engine_connect(self):
+
+        # connect to a real engine
+        module_engine.connect(testbase.db_uri)
+        users.create()
+
+        objectstore.begin()
+
+        user = User()
+        user.user_name='fred'
+        user.password='*'
+        objectstore.commit()
+        
+        # select
+        sqluser = User.select_by(user_name='fred')[0]
+        assert sqluser.user_name == 'fred'
+        
+        # modify
+        sqluser.user_name = 'fred jones'
+
+        # commit - saves everything that changed
+        objectstore.commit()
+
+        allusers = [ user.user_name for user in User.select() ]
+        assert allusers == [ 'fred jones' ]
+        users.drop()
+        
+    def test_multi_thread(self):
+        from threading import Thread
+        from Queue import Queue
+        
+        # start 2 threads with different connection params
+        # and perform simultaneous operations, showing that the
+        # 2 threads don't share a connection
+        qa = Queue()
+        qb = Queue()
+        def run(db_uri, uname, queue):
+            def test():
+                try:
+                    module_engine.connect(db_uri)
+                    users.create()
+
+                    objectstore.begin()
+                    
+                    all = User.select()[:]
+                    assert all == []
+
+                    u = User()
+                    u.user_name = uname
+                    u.password = 'whatever'
+                    objectstore.commit()
+
+                    names = [ us.user_name for us in User.select() ]
+                    assert names == [ uname ]                    
+                    users.drop()
+                    
+                except Exception, e:
+                    import traceback
+                    traceback.print_exc()
+                    queue.put(e)
+                else:
+                    queue.put(False)
+            return test
+
+        # NOTE: I'm not sure how to give the test runner the option to
+        # override these uris, or how to safely clear them after test runs
+        a = Thread(target=run('sqlite://filename=threadtesta.db', 'jim', qa))
+        b = Thread(target=run('sqlite://filename=threadtestb.db', 'joe', qb))
+        
+        a.start()
+        b.start()
+
+        # block and wait for the threads to push their results
+        res = qa.get(True)
+        if res != False:
+            raise res
+
+        res = qb.get(True)
+        if res != False:
+            raise res
+
+    def test_table_singleton_a(self):
+        """set up for table singleton check
+        """
+
+        #
+        # For this 'test', create a proxy engine instance, connect it
+        # to a real engine, and make it do some work
+        #
+        engine = ProxyEngine()
+        cats = Table('cats', engine,
+                     Column('cat_id', Integer, primary_key=True),
+                     Column('cat_name', String))
+
+        engine.connect(testbase.db_uri)
+        cats.create()
+        cats.drop()
+
+        ProxyEngineTest.cats_table_a = cats
+        assert isinstance(cats, Table)
+
+    def test_table_singleton_b(self):
+        """check that a table on a 2nd proxy engine instance gets 2nd table
+        instance
+        """
+
+        #
+        # Now create a new proxy engine instance and attach the same
+        # table as the first test. This should result in 2 table instances,
+        # since different proxy engine instances can't attach to the
+        # same table instance
+        #
+        engine = ProxyEngine()
+        cats = Table('cats', engine,
+                     Column('cat_id', Integer, primary_key=True),
+                     Column('cat_name', String))
+        assert id(cats) != id(ProxyEngineTest.cats_table_a)
+
+        # the real test -- if we're still using the old engine reference,
+        # this will fail because the old reference's local storage will
+        # not have the default attributes
+        engine.connect(testbase.db_uri)
+        cats.create()
+        cats.drop()
+
+    def test_type_engine_caching(self):
+        from sqlalchemy.engine import SQLEngine
+        import sqlalchemy.types as sqltypes
+
+        class EngineA(SQLEngine):
+            def __init__(self):
+                pass
+
+            def hash_key(self):
+                return 'a'
+            
+            def type_descriptor(self, typeobj):
+                if typeobj == type(1):
+                    return TypeEngineX2()
+                else:
+                    return TypeEngineSTR()
+            
+        class EngineB(SQLEngine):
+            def __init__(self):
+                pass
+
+            def hash_key(self):
+                return 'b'
+            
+            def type_descriptor(self, typeobj):
+                return TypeEngineMonkey()
+
+        class TypeEngineX2(sqltypes.TypeEngine):
+            def convert_bind_param(self, value, engine):
+                return value * 2
+
+        class TypeEngineSTR(sqltypes.TypeEngine):
+            def convert_bind_param(self, value, engine):
+                return repr(str(value))
+
+        class TypeEngineMonkey(sqltypes.TypeEngine):
+            def convert_bind_param(self, value, engine):
+                return 'monkey'
+            
+        engine = ProxyEngine()
+        engine.storage.engine = EngineA()
+
+        a = engine.type_descriptor(type(1))
+        assert a.convert_bind_param(12, engine) == 24
+        assert a.convert_bind_param([1,2,3], engine) == [1, 2, 3, 1, 2, 3]
+
+        a2 = engine.type_descriptor(type('hi'))
+        assert a2.convert_bind_param(12, engine) == "'12'"
+        assert a2.convert_bind_param([1,2,3], engine) == "'[1, 2, 3]'"
+        
+        engine.storage.engine = EngineB()
+        b = engine.type_descriptor(type(1))
+        assert b.convert_bind_param(12, engine) == 'monkey'
+        assert b.convert_bind_param([1,2,3], engine) == 'monkey'
+        
+        
+if __name__ == "__main__":
+    testbase.main()
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/test/testbase.py b/test/testbase.py
index e6e2e8ef57..b84c08e273 100644
--- a/test/testbase.py
+++ b/test/testbase.py
@@ -9,7 +9,7 @@ import sqlalchemy.databases.postgres as postgres
 echo = True
 #echo = 'debug'
 db = None
-
+db_uri = None
 
 def parse_argv():
     # we are using the unittest main runner, so we are just popping out the 
@@ -20,21 +20,25 @@ def parse_argv():
     else:
         DBTYPE = 'sqlite'
 
-    global db
+    global db, db_uri
     if DBTYPE == 'sqlite':
         try:
-            db = engine.create_engine('sqlite://filename=:memory:', echo=echo, default_ordering=True)
+            db_uri = 'sqlite://filename=:memory:'
+            db = engine.create_engine(db_uri, echo=echo, default_ordering=True)
         except:
             raise "Could not create sqlite engine.  specify --db <sqlite|sqlite_file|postgres|mysql|oracle> to test runner."
     elif DBTYPE == 'sqlite_file':
-        db = engine.create_engine('sqlite://filename=querytest.db', echo=echo, default_ordering=True)
+        db_uri = 'sqlite://filename=querytest.db'
+        db = engine.create_engine(db_uri, echo=echo, default_ordering=True)
     elif DBTYPE == 'postgres':
-        db = engine.create_engine('postgres://database=test&port=5432&host=127.0.0.1&user=scott&password=tiger',
-         echo=echo, default_ordering=True)
+        db_uri = 'postgres://database=test&port=5432&host=127.0.0.1&user=scott&password=tiger'
+        db = engine.create_engine(db_uri, echo=echo, default_ordering=True)
     elif DBTYPE == 'mysql':
-        db = engine.create_engine('mysql://db=test&host=127.0.0.1&user=scott&passwd=tiger', echo=echo, default_ordering=True)
+        db_uri = 'mysql://db=test&host=127.0.0.1&user=scott&passwd=tiger'
+        db = engine.create_engine(db_uri, echo=echo, default_ordering=True)
     elif DBTYPE == 'oracle':
-        db = engine.create_engine('oracle://user=scott&password=tiger', echo=echo, default_ordering=True)
+        db_uri = 'oracle://user=scott&password=tiger'
+        db = engine.create_engine(db_uri, echo=echo, default_ordering=True)
     db = EngineAssert(db)
 
 class PersistTest(unittest.TestCase):
@@ -222,3 +226,5 @@ def runTests(suite):
     
 def main():
     unittest.main()
+
+
diff --git a/test/wsgi_test.py b/test/wsgi_test.py
new file mode 100644
index 0000000000..1330f88b63
--- /dev/null
+++ b/test/wsgi_test.py
@@ -0,0 +1,122 @@
+"""Interactive wsgi test
+
+Small WSGI application that uses a table and mapper defined at the module
+level, with per-application uris enabled by the ProxyEngine.
+
+Requires the wsgiutils package from:
+
+http://www.owlfish.com/software/wsgiutils/
+
+Run the script with python wsgi_test.py, then visit http://localhost:8080/a
+and http://localhost:8080/b with a browser. You should see two distinct lists
+of colors.
+"""
+
+from sqlalchemy import *
+from sqlalchemy.ext.proxy import ProxyEngine
+from wsgiutils import wsgiServer
+
+engine = ProxyEngine()
+
+colors = Table('colors', engine,
+               Column('id', Integer, primary_key=True),
+               Column('name', String(32)),
+               Column('hex', String(6)))
+
+class Color(object):
+    pass
+
+assign_mapper(Color, colors)
+
+data = { 'a': (('fff','white'), ('aaa','gray'), ('000','black'),
+               ('f00', 'red'), ('0f0', 'green')),
+         'b': (('00f','blue'), ('ff0', 'yellow'), ('0ff','purple')) }
+
+db_uri = { 'a': 'sqlite://filename=wsgi_db_a.db',
+           'b': 'sqlite://filename=wsgi_db_b.db' }
+
+def app(dataset):
+    print '... connecting to database %s: %s' % (dataset, db_uri[dataset])
+    engine.connect(db_uri[dataset], echo=True, echo_pool=True)
+    colors.create()
+
+    print '... populating data into %s' % db_uri[dataset]
+    for hex, name in data[dataset]:
+        c = Color()
+        c.hex = hex
+        c.name = name
+    objectstore.commit()
+    objectstore.clear()
+    
+    def call(environ, start_response):
+        engine.connect(db_uri[dataset], echo=True, echo_pool=True)
+
+        # NOTE: must clear objectstore on each request, or you'll see
+        # objects from another thread here
+        objectstore.clear()
+        objectstore.begin()
+        
+        c = Color.select()
+
+        start_response('200 OK', [('content-type','text/html')])
+        yield '<html><head><title>Test dataset %s</title></head>' % dataset
+        yield '<body>'
+        yield '<p>uri: %s</p>' % db_uri[dataset]
+        yield '<p>engine: <xmp>%s</xmp></p>' % engine.engine
+        yield '<p>Colors!</p>'
+        for color in c:
+            yield '<div style="background: #%s">%s</div>' % (color.hex,
+                                                             color.name)
+        yield '</body></html>'
+    return call
+
+def cleanup():
+    for uri in db_uri.values():
+        print "Cleaning db %s" % uri
+        engine.connect(uri)
+        colors.drop()
+
+def run_server(apps, host='localhost', port=8080):
+    print "Serving test app at http://%s:%s/" % (host, port)
+    print "Visit http://%(host)s:%(port)s/a and " \
+        "http://%(host)s:%(port)s/b to test apps" % {'host': host,
+                                                     'port': port}
+    
+    server = wsgiServer.WSGIServer((host, port), apps, serveFiles=False)
+    try:
+        server.serve_forever()
+    except:
+        cleanup()
+        raise
+    
+if __name__ == '__main__':
+    run_server({'/a':app('a'), '/b':app('b')})
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+