Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Adding some more tests based on the old versions

  • Loading branch information...
commit 3597298f47a35a9824d687f93f133c19357b9966 1 parent 120ab5c
@coleifer authored
Showing with 117 additions and 5 deletions.
  1. +20 −3 peewee.py
  2. +97 −2 tests.py
View
23 peewee.py
@@ -1418,12 +1418,12 @@ def is_closed(self):
def get_cursor(self):
return self.get_conn().cursor()
+ def _close(self, conn):
+ conn.close()
+
def _connect(self, database, **kwargs):
raise NotImplementedError
- def close(self, conn):
- return conn.close()
-
def last_insert_id(self, cursor, model):
if model._meta.auto_increment:
return cursor.lastrowid
@@ -1491,6 +1491,23 @@ def drop_table(self, model_class, fail_silently=False):
def transaction(self):
return transaction(self)
+ def commit_on_success(self, func):
+ def inner(*args, **kwargs):
+ orig = self.get_autocommit()
+ self.set_autocommit(False)
+ self.begin()
+ try:
+ res = func(*args, **kwargs)
+ self.commit()
+ except:
+ self.rollback()
+ raise
+ else:
+ return res
+ finally:
+ self.set_autocommit(orig)
+ return inner
+
class SqliteDatabase(Database):
op_overrides = {
View
99 tests.py
@@ -1200,8 +1200,94 @@ def test_db_column(self):
class TransactionTestCase(ModelTestCase):
- requires = []
- # TODO
+ requires = [User, Blog]
+
+ def tearDown(self):
+ super(TransactionTestCase, self).tearDown()
+ test_db.set_autocommit(True)
+
+ def test_autocommit(self):
+ test_db.set_autocommit(False)
+
+ u1 = User.create(username='u1')
+ u2 = User.create(username='u2')
+
+ # open up a new connection to the database, it won't register any blogs
+ # as being created
+ new_db = database_class(database_name)
+ res = new_db.execute_sql('select count(*) from users;')
+ self.assertEqual(res.fetchone()[0], 0)
+
+ # commit our blog inserts
+ test_db.commit()
+
+ # now the blogs are query-able from another connection
+ res = new_db.execute_sql('select count(*) from users;')
+ self.assertEqual(res.fetchone()[0], 2)
+
+ def test_commit_on_success(self):
+ self.assertTrue(test_db.get_autocommit())
+
+ @test_db.commit_on_success
+ def will_fail():
+ u = User.create(username='u1')
+ b = Blog.create() # no blog, will raise an error
+ return u, b
+
+ self.assertRaises(Exception, will_fail)
+ self.assertEqual(User.select().count(), 0)
+ self.assertEqual(Blog.select().count(), 0)
+
+ @test_db.commit_on_success
+ def will_succeed():
+ u = User.create(username='u1')
+ b = Blog.create(title='b1', user=u)
+ return u, b
+
+ u, b = will_succeed()
+ self.assertEqual(User.select().count(), 1)
+ self.assertEqual(Blog.select().count(), 1)
+
+ def test_context_mgr(self):
+ def will_fail():
+ u = User.create(username='u1')
+ b = Blog.create() # no blog, will raise an error
+ return u, b
+
+ def do_will_fail():
+ with transaction(test_db):
+ will_fail()
+
+ def do_will_fail2():
+ with test_db.transaction():
+ will_fail()
+
+ self.assertRaises(Exception, do_will_fail)
+ self.assertEqual(Blog.select().count(), 0)
+
+ self.assertRaises(Exception, do_will_fail2)
+ self.assertEqual(Blog.select().count(), 0)
+
+ def will_succeed():
+ u = User.create(username='u1')
+ b = Blog.create(title='b1', user=u)
+ return u, b
+
+ def do_will_succeed():
+ with transaction(test_db):
+ will_succeed()
+
+ def do_will_succeed2():
+ with test_db.transaction():
+ will_succeed()
+
+ do_will_succeed()
+ self.assertEqual(User.select().count(), 1)
+ self.assertEqual(Blog.select().count(), 1)
+
+ do_will_succeed2()
+ self.assertEqual(User.select().count(), 2)
+ self.assertEqual(Blog.select().count(), 2)
class ConcurrencyTestCase(ModelTestCase):
@@ -1240,6 +1326,15 @@ class Meta:
self.assertTrue(deferred_db.deferred)
+class ConnectionStateTestCase(BasePeeweeTestCase):
+ def test_connection_state(self):
+ conn = test_db.get_conn()
+ self.assertFalse(test_db.is_closed())
+ test_db.close()
+ self.assertTrue(test_db.is_closed())
+ conn = test_db.get_conn()
+ self.assertFalse(test_db.is_closed())
+
class ForUpdateTestCase(ModelTestCase):
requires = []
Please sign in to comment.
Something went wrong with that request. Please try again.