summaryrefslogtreecommitdiff
path: root/test/engine/execute.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2007-02-13 22:53:05 +0000
committerMike Bayer <mike_mp@zzzcomputing.com>2007-02-13 22:53:05 +0000
commite7aaeb28527d27814cb192e1ddc4af228a43d816 (patch)
treeed36612431a3f845cb580f5b2d25e4398f44213e /test/engine/execute.py
parent238e8620ee2556d86ecd2ca6a430de990951d514 (diff)
downloadsqlalchemy-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.py56
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()