test_update_delete.py 1.6 KB

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