← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/base/test_events.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/base/test_events.py
1
1
"""Test event registration and listening."""
2
2
 
3
 
from test.lib.testing import eq_, assert_raises, assert_raises_message, \
 
3
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message, \
4
4
    is_, is_not_
5
 
from sqlalchemy import event, exc, util
6
 
from test.lib import fixtures
7
 
 
8
 
class TestEvents(fixtures.TestBase):
 
5
from sqlalchemy import event, exc
 
6
from sqlalchemy.testing import fixtures
 
7
from sqlalchemy.testing.util import gc_collect
 
8
from sqlalchemy.testing.mock import Mock, call
 
9
 
 
10
 
 
11
class EventsTest(fixtures.TestBase):
9
12
    """Test class- and instance-level event registration."""
10
13
 
11
14
    def setUp(self):
168
171
                meth
169
172
            )
170
173
 
171
 
class TestClsLevelListen(fixtures.TestBase):
 
174
class ClsLevelListenTest(fixtures.TestBase):
172
175
 
173
176
 
174
177
    def tearDown(self):
189
192
    def test_lis_subcalss_lis(self):
190
193
        @event.listens_for(self.TargetOne, "event_one")
191
194
        def handler1(x, y):
192
 
            print 'handler1'
 
195
            pass
193
196
 
194
197
        class SubTarget(self.TargetOne):
195
198
            pass
206
209
    def test_lis_multisub_lis(self):
207
210
        @event.listens_for(self.TargetOne, "event_one")
208
211
        def handler1(x, y):
209
 
            print 'handler1'
 
212
            pass
210
213
 
211
214
        class SubTarget(self.TargetOne):
212
215
            pass
249
252
        assert handler2 not in s2.dispatch.event_one
250
253
 
251
254
 
252
 
class TestAcceptTargets(fixtures.TestBase):
 
255
class AcceptTargetsTest(fixtures.TestBase):
253
256
    """Test default target acceptance."""
254
257
 
255
258
    def setUp(self):
320
323
            [listen_two, listen_four]
321
324
        )
322
325
 
323
 
class TestCustomTargets(fixtures.TestBase):
 
326
class CustomTargetsTest(fixtures.TestBase):
324
327
    """Test custom target acceptance."""
325
328
 
326
329
    def setUp(self):
359
362
            listen, "event_one", self.Target
360
363
        )
361
364
 
362
 
class TestListenOverride(fixtures.TestBase):
 
365
class SubclassGrowthTest(fixtures.TestBase):
 
366
    """test that ad-hoc subclasses are garbage collected."""
 
367
 
 
368
    def setUp(self):
 
369
        class TargetEvents(event.Events):
 
370
            def some_event(self, x, y):
 
371
                pass
 
372
 
 
373
        class Target(object):
 
374
            dispatch = event.dispatcher(TargetEvents)
 
375
 
 
376
        self.Target = Target
 
377
 
 
378
    def test_subclass(self):
 
379
        class SubTarget(self.Target):
 
380
            pass
 
381
 
 
382
        st = SubTarget()
 
383
        st.dispatch.some_event(1, 2)
 
384
        del st
 
385
        del SubTarget
 
386
        gc_collect()
 
387
        eq_(self.Target.__subclasses__(), [])
 
388
 
 
389
 
 
390
class ListenOverrideTest(fixtures.TestBase):
363
391
    """Test custom listen functions which change the listener function signature."""
364
392
 
365
393
    def setUp(self):
385
413
        event._remove_dispatcher(self.Target.__dict__['dispatch'].events)
386
414
 
387
415
    def test_listen_override(self):
388
 
        result = []
389
 
        def listen_one(x):
390
 
            result.append(x)
391
 
 
392
 
        def listen_two(x, y):
393
 
            result.append((x, y))
 
416
        listen_one = Mock()
 
417
        listen_two = Mock()
394
418
 
395
419
        event.listen(self.Target, "event_one", listen_one, add=True)
396
420
        event.listen(self.Target, "event_one", listen_two)
399
423
        t1.dispatch.event_one(5, 7)
400
424
        t1.dispatch.event_one(10, 5)
401
425
 
402
 
        eq_(result,
403
 
            [
404
 
                12, (5, 7), 15, (10, 5)
405
 
            ]
 
426
        eq_(
 
427
            listen_one.mock_calls,
 
428
            [call(12), call(15)]
 
429
        )
 
430
        eq_(
 
431
            listen_two.mock_calls,
 
432
            [call(5, 7), call(10, 5)]
406
433
        )
407
434
 
408
 
class TestPropagate(fixtures.TestBase):
 
435
class PropagateTest(fixtures.TestBase):
409
436
    def setUp(self):
410
437
        class TargetEvents(event.Events):
411
438
            def event_one(self, arg):
420
447
 
421
448
 
422
449
    def test_propagate(self):
423
 
        result = []
424
 
        def listen_one(target, arg):
425
 
            result.append((target, arg))
426
 
 
427
 
        def listen_two(target, arg):
428
 
            result.append((target, arg))
 
450
        listen_one = Mock()
 
451
        listen_two = Mock()
429
452
 
430
453
        t1 = self.Target()
431
454
 
438
461
 
439
462
        t2.dispatch.event_one(t2, 1)
440
463
        t2.dispatch.event_two(t2, 2)
441
 
        eq_(result, [(t2, 1)])
 
464
 
 
465
        eq_(
 
466
            listen_one.mock_calls,
 
467
            [call(t2, 1)]
 
468
        )
 
469
        eq_(
 
470
            listen_two.mock_calls,
 
471
            []
 
472
        )
 
473
 
 
474
class JoinTest(fixtures.TestBase):
 
475
    def setUp(self):
 
476
        class TargetEvents(event.Events):
 
477
            def event_one(self, target, arg):
 
478
                pass
 
479
 
 
480
        class BaseTarget(object):
 
481
            dispatch = event.dispatcher(TargetEvents)
 
482
 
 
483
        class TargetFactory(BaseTarget):
 
484
            def create(self):
 
485
                return TargetElement(self)
 
486
 
 
487
        class TargetElement(BaseTarget):
 
488
            def __init__(self, parent):
 
489
                self.dispatch = self.dispatch._join(parent.dispatch)
 
490
 
 
491
            def run_event(self, arg):
 
492
                list(self.dispatch.event_one)
 
493
                self.dispatch.event_one(self, arg)
 
494
 
 
495
        self.BaseTarget = BaseTarget
 
496
        self.TargetFactory = TargetFactory
 
497
        self.TargetElement = TargetElement
 
498
 
 
499
    def tearDown(self):
 
500
        for cls in (self.TargetElement,
 
501
                self.TargetFactory, self.BaseTarget):
 
502
            if 'dispatch' in cls.__dict__:
 
503
                event._remove_dispatcher(cls.__dict__['dispatch'].events)
 
504
 
 
505
    def test_neither(self):
 
506
        element = self.TargetFactory().create()
 
507
        element.run_event(1)
 
508
        element.run_event(2)
 
509
        element.run_event(3)
 
510
 
 
511
    def test_parent_class_only(self):
 
512
        l1 = Mock()
 
513
 
 
514
        event.listen(self.TargetFactory, "event_one", l1)
 
515
 
 
516
        element = self.TargetFactory().create()
 
517
        element.run_event(1)
 
518
        element.run_event(2)
 
519
        element.run_event(3)
 
520
        eq_(
 
521
            l1.mock_calls,
 
522
            [call(element, 1), call(element, 2), call(element, 3)]
 
523
        )
 
524
 
 
525
    def test_parent_class_child_class(self):
 
526
        l1 = Mock()
 
527
        l2 = Mock()
 
528
 
 
529
        event.listen(self.TargetFactory, "event_one", l1)
 
530
        event.listen(self.TargetElement, "event_one", l2)
 
531
 
 
532
        element = self.TargetFactory().create()
 
533
        element.run_event(1)
 
534
        element.run_event(2)
 
535
        element.run_event(3)
 
536
        eq_(
 
537
            l1.mock_calls,
 
538
            [call(element, 1), call(element, 2), call(element, 3)]
 
539
        )
 
540
        eq_(
 
541
            l2.mock_calls,
 
542
            [call(element, 1), call(element, 2), call(element, 3)]
 
543
        )
 
544
 
 
545
    def test_parent_class_child_instance_apply_after(self):
 
546
        l1 = Mock()
 
547
        l2 = Mock()
 
548
 
 
549
        event.listen(self.TargetFactory, "event_one", l1)
 
550
        element = self.TargetFactory().create()
 
551
 
 
552
        element.run_event(1)
 
553
 
 
554
        event.listen(element, "event_one", l2)
 
555
        element.run_event(2)
 
556
        element.run_event(3)
 
557
 
 
558
        eq_(
 
559
            l1.mock_calls,
 
560
            [call(element, 1), call(element, 2), call(element, 3)]
 
561
        )
 
562
        eq_(
 
563
            l2.mock_calls,
 
564
            [call(element, 2), call(element, 3)]
 
565
        )
 
566
 
 
567
    def test_parent_class_child_instance_apply_before(self):
 
568
        l1 = Mock()
 
569
        l2 = Mock()
 
570
 
 
571
        event.listen(self.TargetFactory, "event_one", l1)
 
572
        element = self.TargetFactory().create()
 
573
 
 
574
        event.listen(element, "event_one", l2)
 
575
 
 
576
        element.run_event(1)
 
577
        element.run_event(2)
 
578
        element.run_event(3)
 
579
 
 
580
        eq_(
 
581
            l1.mock_calls,
 
582
            [call(element, 1), call(element, 2), call(element, 3)]
 
583
        )
 
584
        eq_(
 
585
            l2.mock_calls,
 
586
            [call(element, 1), call(element, 2), call(element, 3)]
 
587
        )
 
588
 
 
589
    def test_parent_instance_child_class_apply_before(self):
 
590
        l1 = Mock()
 
591
        l2 = Mock()
 
592
 
 
593
        event.listen(self.TargetElement, "event_one", l2)
 
594
 
 
595
        factory = self.TargetFactory()
 
596
        event.listen(factory, "event_one", l1)
 
597
 
 
598
        element = factory.create()
 
599
 
 
600
        element.run_event(1)
 
601
        element.run_event(2)
 
602
        element.run_event(3)
 
603
 
 
604
        eq_(
 
605
            l1.mock_calls,
 
606
            [call(element, 1), call(element, 2), call(element, 3)]
 
607
        )
 
608
        eq_(
 
609
            l2.mock_calls,
 
610
            [call(element, 1), call(element, 2), call(element, 3)]
 
611
        )
 
612
 
 
613
 
 
614
    def test_parent_instance_child_class_apply_after(self):
 
615
        l1 = Mock()
 
616
        l2 = Mock()
 
617
 
 
618
        event.listen(self.TargetElement, "event_one", l2)
 
619
 
 
620
        factory = self.TargetFactory()
 
621
        element = factory.create()
 
622
 
 
623
        element.run_event(1)
 
624
 
 
625
        event.listen(factory, "event_one", l1)
 
626
 
 
627
        element.run_event(2)
 
628
        element.run_event(3)
 
629
 
 
630
        # c1 gets no events due to _JoinedListener
 
631
        # fixing the "parent" at construction time.
 
632
        # this can be changed to be "live" at the cost
 
633
        # of performance.
 
634
        eq_(
 
635
            l1.mock_calls, []
 
636
        )
 
637
        eq_(
 
638
            l2.mock_calls,
 
639
            [call(element, 1), call(element, 2), call(element, 3)]
 
640
        )
 
641
 
 
642
    def test_parent_instance_child_instance_apply_before(self):
 
643
        l1 = Mock()
 
644
        l2 = Mock()
 
645
        factory = self.TargetFactory()
 
646
 
 
647
        event.listen(factory, "event_one", l1)
 
648
 
 
649
        element = factory.create()
 
650
        event.listen(element, "event_one", l2)
 
651
 
 
652
        element.run_event(1)
 
653
        element.run_event(2)
 
654
        element.run_event(3)
 
655
 
 
656
        eq_(
 
657
            l1.mock_calls,
 
658
            [call(element, 1), call(element, 2), call(element, 3)]
 
659
        )
 
660
        eq_(
 
661
            l2.mock_calls,
 
662
            [call(element, 1), call(element, 2), call(element, 3)]
 
663
        )
 
664
 
 
665
    def test_parent_events_child_no_events(self):
 
666
        l1 = Mock()
 
667
        factory = self.TargetFactory()
 
668
 
 
669
        event.listen(self.TargetElement, "event_one", l1)
 
670
        element = factory.create()
 
671
 
 
672
        element.run_event(1)
 
673
        element.run_event(2)
 
674
        element.run_event(3)
 
675
 
 
676
        eq_(
 
677
            l1.mock_calls,
 
678
            [call(element, 1), call(element, 2), call(element, 3)]
 
679
        )