diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2007-02-13 22:53:05 +0000 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2007-02-13 22:53:05 +0000 |
commit | e7aaeb28527d27814cb192e1ddc4af228a43d816 (patch) | |
tree | ed36612431a3f845cb580f5b2d25e4398f44213e /test/engine/execute.py | |
parent | 238e8620ee2556d86ecd2ca6a430de990951d514 (diff) | |
download | sqlalchemy-e7aaeb28527d27814cb192e1ddc4af228a43d816.tar.gz |
- fixed argument passing to straight textual execute() on engine, connection.
can handle *args or a list instance for positional, **kwargs or a dict instance
for named args, or a list of list or dicts to invoke executemany()
Diffstat (limited to 'test/engine/execute.py')
-rw-r--r-- | test/engine/execute.py | 56 |
1 files changed, 56 insertions, 0 deletions
diff --git a/test/engine/execute.py b/test/engine/execute.py new file mode 100644 index 000000000..ce60a21c7 --- /dev/null +++ b/test/engine/execute.py @@ -0,0 +1,56 @@ + +import testbase +import unittest, sys, datetime +import tables +db = testbase.db +from sqlalchemy import * + + +class ExecuteTest(testbase.PersistTest): + def setUpAll(self): + global users, metadata + metadata = BoundMetaData(testbase.db) + users = Table('users', metadata, + Column('user_id', INT, primary_key = True), + Column('user_name', VARCHAR(20)), + mysql_engine='InnoDB' + ) + metadata.create_all() + + def tearDown(self): + testbase.db.connect().execute(users.delete()) + def tearDownAll(self): + metadata.drop_all() + + @testbase.supported('sqlite') + def test_raw_qmark(self): + for conn in (testbase.db, testbase.db.connect()): + conn.execute("insert into users (user_id, user_name) values (?, ?)", [1,"jack"]) + conn.execute("insert into users (user_id, user_name) values (?, ?)", [2,"ed"], [3,"horse"]) + conn.execute("insert into users (user_id, user_name) values (?, ?)", 4, 'sally') + res = conn.execute("select * from users") + assert res.fetchall() == [(1, "jack"), (2, "ed"), (3, "horse"), (4, 'sally')] + conn.execute("delete from users") + + @testbase.supported('mysql') + def test_raw_sprintf(self): + for conn in (testbase.db, testbase.db.connect()): + conn.execute("insert into users (user_id, user_name) values (%s, %s)", [1,"jack"]) + conn.execute("insert into users (user_id, user_name) values (%s, %s)", [2,"ed"], [3,"horse"]) + conn.execute("insert into users (user_id, user_name) values (%s, %s)", 4, 'sally') + res = conn.execute("select * from users") + assert res.fetchall() == [(1, "jack"), (2, "ed"), (3, "horse"), (4, 'sally')] + conn.execute("delete from users") + + @testbase.supported('postgres') + def test_raw_python(self): + for conn in (testbase.db, testbase.db.connect()): + conn.execute("insert into users (user_id, user_name) values (%(id)s, %(name)s)", {'id':1, 'name':'jack'}) + conn.execute("insert into users (user_id, user_name) values (%(id)s, %(name)s)", {'id':2, 'name':'ed'}, {'id':3, 'name':'horse'}) + conn.execute("insert into users (user_id, user_name) values (%(id)s, %(name)s)", id=4, name='sally') + res = conn.execute("select * from users") + assert res.fetchall() == [(1, "jack"), (2, "ed"), (3, "horse"), (4, 'sally')] + conn.execute("delete from users") + +if __name__ == "__main__": + testbase.main() |