Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Moving signals tests into their own module and getting signals working

again
  • Loading branch information...
commit fbccdf244423a66abfc89ae68e817180fc71cada 1 parent b0c8236
Charles Leifer authored
8 peewee.py
View
@@ -1909,11 +1909,9 @@ def delete(cls):
@classmethod
def create(cls, **query):
- insert = cls.insert(**query)
- new_pk = insert.execute()
- if cls._meta.auto_increment:
- query[cls._meta.primary_key.name] = new_pk
- return cls(**query)
+ inst = cls(**query)
+ inst.save(force_insert=True)
+ return inst
@classmethod
def filter(cls, *dq, **query):
2  playhouse/signals.py
View
@@ -54,7 +54,7 @@ def prepared(self):
post_init.send(self)
def save(self, *args, **kwargs):
- created = not bool(self.get_pk())
+ created = not bool(self.get_id())
pre_save.send(self, created=created)
super(Model, self).save(*args, **kwargs)
post_save.send(self, created=created)
139 playhouse/tests_signals.py
View
@@ -0,0 +1,139 @@
+import unittest
+
+from peewee import *
+import signals
+
+
+db = SqliteDatabase(':memory:')
+
+class BaseSignalModel(signals.Model):
+ class Meta:
+ database = db
+
+class ModelA(BaseSignalModel):
+ a = CharField(default='')
+
+class ModelB(BaseSignalModel):
+ b = CharField(default='')
+
+
+class SignalsTestCase(unittest.TestCase):
+ def setUp(self):
+ ModelA.create_table(True)
+ ModelB.create_table(True)
+
+ def tearDown(self):
+ ModelA.drop_table()
+ ModelB.drop_table()
+ signals.pre_save._flush()
+ signals.post_save._flush()
+ signals.pre_delete._flush()
+ signals.post_delete._flush()
+ signals.pre_init._flush()
+ signals.post_init._flush()
+
+ def test_pre_save(self):
+ state = []
+
+ @signals.connect(signals.pre_save)
+ def pre_save(sender, instance, created):
+ state.append((sender, instance, instance.get_id(), created))
+ m = ModelA()
+ m.save()
+ self.assertEqual(state, [(ModelA, m, None, True)])
+
+ m.save()
+ self.assertTrue(m.id is not None)
+ self.assertEqual(state[-1], (ModelA, m, m.id, False))
+
+ def test_post_save(self):
+ state = []
+
+ @signals.connect(signals.post_save)
+ def post_save(sender, instance, created):
+ state.append((sender, instance, instance.get_id(), created))
+ m = ModelA()
+ m.save()
+
+ self.assertTrue(m.id is not None)
+ self.assertEqual(state, [(ModelA, m, m.id, True)])
+
+ m.save()
+ self.assertEqual(state[-1], (ModelA, m, m.id, False))
+
+ def test_pre_delete(self):
+ state = []
+
+ m = ModelA()
+ m.save()
+
+ @signals.connect(signals.pre_delete)
+ def pre_delete(sender, instance):
+ state.append((sender, instance, ModelA.select().count()))
+ m.delete_instance()
+ self.assertEqual(state, [(ModelA, m, 1)])
+
+ def test_post_delete(self):
+ state = []
+
+ m = ModelA()
+ m.save()
+
+ @signals.connect(signals.post_delete)
+ def post_delete(sender, instance):
+ state.append((sender, instance, ModelA.select().count()))
+ m.delete_instance()
+ self.assertEqual(state, [(ModelA, m, 0)])
+
+ def test_pre_init(self):
+ state = []
+
+ m = ModelA(a='a')
+ m.save()
+
+ @signals.connect(signals.pre_init)
+ def pre_init(sender, instance):
+ state.append((sender, instance.a))
+
+ ModelA.get()
+ self.assertEqual(state, [(ModelA, '')])
+
+ def test_post_init(self):
+ state = []
+
+ m = ModelA(a='a')
+ m.save()
+
+ @signals.connect(signals.post_init)
+ def post_init(sender, instance):
+ state.append((sender, instance.a))
+
+ ModelA.get()
+ self.assertEqual(state, [(ModelA, 'a')])
+
+ def test_sender(self):
+ state = []
+
+ @signals.connect(signals.post_save, sender=ModelA)
+ def post_save(sender, instance, created):
+ state.append(instance)
+
+ m = ModelA.create()
+ self.assertEqual(state, [m])
+
+ m2 = ModelB.create()
+ self.assertEqual(state, [m])
+
+ def test_connect_disconnect(self):
+ state = []
+
+ @signals.connect(signals.post_save, sender=ModelA)
+ def post_save(sender, instance, created):
+ state.append(instance)
+
+ m = ModelA.create()
+ self.assertEqual(state, [m])
+
+ signals.post_save.disconnect(post_save)
+ m2 = ModelA.create()
+ self.assertEqual(state, [m])
3  runtests.py
View
@@ -32,6 +32,9 @@ def get_option_parser():
if options.all or options.extra:
modules = [tests]
+ from playhouse import tests_signals
+ modules.append(tests_signals)
+
#from playhouse import tests as extras_tests
#modules.append(extras_tests)
try:
7 tests.py
View
@@ -835,6 +835,13 @@ def test_saving(self):
self.assertEqual(User.select().count(), 1)
+ def test_saving_via_create_gh111(self):
+ u = User.create(username='u')
+ b = Blog.create(title='foo', user=u)
+ last_sql, _ = self.queries()[-1]
+ self.assertFalse('pub_date' in last_sql)
+ self.assertEqual(b.pub_date, None)
+
def test_reading(self):
u1 = self.create_user('u1')
u2 = self.create_user('u2')
Please sign in to comment.
Something went wrong with that request. Please try again.