1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
| # mypy: ignore-errors
|
| from .. import fixtures
| from ..assertions import eq_
| from ..schema import Column
| from ..schema import Table
| from ... import Integer
| from ... import String
|
|
| class SimpleUpdateDeleteTest(fixtures.TablesTest):
| run_deletes = "each"
| __requires__ = ("sane_rowcount",)
| __backend__ = True
|
| @classmethod
| def define_tables(cls, metadata):
| Table(
| "plain_pk",
| metadata,
| Column("id", Integer, primary_key=True),
| Column("data", String(50)),
| )
|
| @classmethod
| def insert_data(cls, connection):
| connection.execute(
| cls.tables.plain_pk.insert(),
| [
| {"id": 1, "data": "d1"},
| {"id": 2, "data": "d2"},
| {"id": 3, "data": "d3"},
| ],
| )
|
| def test_update(self, connection):
| t = self.tables.plain_pk
| r = connection.execute(
| t.update().where(t.c.id == 2), dict(data="d2_new")
| )
| assert not r.is_insert
| assert not r.returns_rows
| assert r.rowcount == 1
|
| eq_(
| connection.execute(t.select().order_by(t.c.id)).fetchall(),
| [(1, "d1"), (2, "d2_new"), (3, "d3")],
| )
|
| def test_delete(self, connection):
| t = self.tables.plain_pk
| r = connection.execute(t.delete().where(t.c.id == 2))
| assert not r.is_insert
| assert not r.returns_rows
| assert r.rowcount == 1
| eq_(
| connection.execute(t.select().order_by(t.c.id)).fetchall(),
| [(1, "d1"), (3, "d3")],
| )
|
|
| __all__ = ("SimpleUpdateDeleteTest",)
|
|
