2 files changed, 68 insertions, 0 deletions
diff --git a/test/sql/alltests.py b/test/sql/alltests.py
index c79d7b67e..98dbfa6a0 100644
--- a/test/sql/alltests.py
+++ b/test/sql/alltests.py
@@ -15,6 +15,7 @@ def suite():
         # assorted round-trip tests
         'sql.query',
         'sql.quote',
+        'sql.rowcount',
         
         # defaults, sequences (postgres/oracle)
         'sql.defaults',
diff --git a/test/sql/rowcount.py b/test/sql/rowcount.py
new file mode 100644
index 000000000..05d0f2110
--- /dev/null
+++ b/test/sql/rowcount.py
@@ -0,0 +1,67 @@
+from sqlalchemy import *
+import testbase
+
+class FoundRowsTest(testbase.AssertMixin):
+    """tests rowcount functionality"""
+    def setUpAll(self):
+        metadata = BoundMetaData(testbase.db)
+
+        global employees_table
+
+        employees_table = Table('employees', metadata,
+            Column('employee_id', Integer, primary_key=True),
+            Column('name', String(50)),
+            Column('department', String(1)),
+        )
+        employees_table.create()
+
+    def setUp(self):
+        global data
+        data = [ ('Angela', 'A'),
+                 ('Andrew', 'A'),
+                 ('Anand', 'A'),
+                 ('Bob', 'B'),
+                 ('Bobette', 'B'),
+                 ('Buffy', 'B'),
+                 ('Charlie', 'C'),
+                 ('Cynthia', 'C'),
+                 ('Chris', 'C') ]
+
+        i = employees_table.insert()
+        i.execute(*[{'name':n, 'department':d} for n, d in data])
+    def tearDown(self):
+        employees_table.delete().execute()
+        
+    def tearDownAll(self):
+        employees_table.drop()
+
+    def testbasic(self):
+        s = employees_table.select()
+        r = s.execute().fetchall()
+
+        assert len(r) == len(data)
+
+    def test_update_rowcount1(self):
+        # WHERE matches 3, 3 rows changed
+        department = employees_table.c.department
+        r = employees_table.update(department=='C').execute(department='Z')
+        assert r.rowcount == 3
+        
+    def test_update_rowcount2(self):
+        # WHERE matches 3, 0 rows changed
+        department = employees_table.c.department
+        r = employees_table.update(department=='C').execute(department='C')
+        assert r.rowcount == 3
+        
+    def test_delete_rowcount(self):
+        # WHERE matches 3, 3 rows deleted
+        department = employees_table.c.department
+        r = employees_table.delete(department=='C').execute()
+        assert r.rowcount == 3
+
+if __name__ == '__main__':
+    testbase.main()
+    
+
+
+