summaryrefslogtreecommitdiff
path: root/test/sql
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2020-02-19 17:59:52 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2020-02-19 18:05:33 -0500
commit1149a81d53ff6825048dd8dea09fb95c803a8944 (patch)
tree7ef3cc7210c01db5604673f10babdcc9dcb75bb8 /test/sql
parent8c9537d37292459d348214fb8befa85d9cb64059 (diff)
downloadsqlalchemy-1149a81d53ff6825048dd8dea09fb95c803a8944.tar.gz
Modernize test_rowcount and move to dialect suite
Amazingly there are no "rowcount" tests in suite, so these tests should definitely be there. Change-Id: Ib4c595fe6e16b457680ce4ee01180ccc8ddb6a40
Diffstat (limited to 'test/sql')
-rw-r--r--test/sql/test_rowcount.py157
1 files changed, 0 insertions, 157 deletions
diff --git a/test/sql/test_rowcount.py b/test/sql/test_rowcount.py
deleted file mode 100644
index 8cff8c98f..000000000
--- a/test/sql/test_rowcount.py
+++ /dev/null
@@ -1,157 +0,0 @@
-from sqlalchemy import bindparam
-from sqlalchemy import Column
-from sqlalchemy import Integer
-from sqlalchemy import MetaData
-from sqlalchemy import Sequence
-from sqlalchemy import String
-from sqlalchemy import Table
-from sqlalchemy import testing
-from sqlalchemy import text
-from sqlalchemy.testing import AssertsExecutionResults
-from sqlalchemy.testing import eq_
-from sqlalchemy.testing import fixtures
-
-
-class FoundRowsTest(fixtures.TestBase, AssertsExecutionResults):
-
- """tests rowcount functionality"""
-
- __requires__ = ("sane_rowcount",)
- __backend__ = True
-
- @classmethod
- def setup_class(cls):
- global employees_table, metadata
- metadata = MetaData(testing.db)
-
- employees_table = Table(
- "employees",
- metadata,
- Column(
- "employee_id",
- Integer,
- Sequence("employee_id_seq", optional=True),
- primary_key=True,
- ),
- Column("name", String(50)),
- Column("department", String(1)),
- )
- metadata.create_all()
-
- 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()
-
- @classmethod
- def teardown_class(cls):
- metadata.drop_all()
-
- def test_basic(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
-
- @testing.skip_if(
- testing.requires.oracle5x, "unknown DBAPI error fixed in later version"
- )
- @testing.requires.sane_rowcount_w_returning
- def test_update_rowcount_return_defaults(self):
- department = employees_table.c.department
- stmt = (
- employees_table.update(department == "C")
- .values(name=employees_table.c.department + "Z")
- .return_defaults()
- )
-
- r = stmt.execute()
- assert r.rowcount == 3
-
- def test_raw_sql_rowcount(self):
- # test issue #3622, make sure eager rowcount is called for text
- with testing.db.connect() as conn:
- result = conn.execute(
- "update employees set department='Z' where department='C'"
- )
- eq_(result.rowcount, 3)
-
- def test_text_rowcount(self):
- # test issue #3622, make sure eager rowcount is called for text
- with testing.db.connect() as conn:
- result = conn.execute(
- text(
- "update employees set department='Z' "
- "where department='C'"
- )
- )
- eq_(result.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
-
- @testing.requires.sane_multi_rowcount
- def test_multi_update_rowcount(self):
- stmt = (
- employees_table.update()
- .where(employees_table.c.name == bindparam("emp_name"))
- .values(department="C")
- )
-
- r = testing.db.execute(
- stmt,
- [
- {"emp_name": "Bob"},
- {"emp_name": "Cynthia"},
- {"emp_name": "nonexistent"},
- ],
- )
-
- eq_(r.rowcount, 2)
-
- @testing.requires.sane_multi_rowcount
- def test_multi_delete_rowcount(self):
- stmt = employees_table.delete().where(
- employees_table.c.name == bindparam("emp_name")
- )
-
- r = testing.db.execute(
- stmt,
- [
- {"emp_name": "Bob"},
- {"emp_name": "Cynthia"},
- {"emp_name": "nonexistent"},
- ],
- )
-
- eq_(r.rowcount, 2)