Source code for sqlobject.tests.test_events

from __future__ import print_function

from sqlobject import IntCol, SQLObject, StringCol, events
from sqlobject.inheritance import InheritableSQLObject
from sqlobject.tests.dbtest import setupClass


class EventTester(SQLObject):
    name = StringCol()


def make_watcher():
    log = []

    def watch(*args):
        log.append(args)

    watch.log = log
    return watch


def make_listen(signal, cls=None):
    if cls is None:
        cls = EventTester
    watcher = make_watcher()
    events.listen(watcher, cls, signal)
    return watcher


def test_create():
    watcher = make_listen(events.ClassCreateSignal)

    class EventTesterSub1(EventTester):
        pass

    class EventTesterSub2(EventTesterSub1):
        pass

    assert len(watcher.log) == 2
    assert len(watcher.log[0]) == 5
    assert watcher.log[0][0] == 'EventTesterSub1'
    assert watcher.log[0][1] == (EventTester,)
    assert isinstance(watcher.log[0][2], dict)
    assert isinstance(watcher.log[0][3], list)


def test_row_create():
    setupClass(EventTester)
    watcher = make_listen(events.RowCreateSignal)
    row1 = EventTester(name='foo')
    row2 = EventTester(name='bar')
    assert len(watcher.log) == 2
    assert watcher.log == [
        (row1, {'name': 'foo'}, []),
        (row2, {'name': 'bar'}, [])]


def test_row_destroy():
    setupClass(EventTester)
    watcher = make_listen(events.RowDestroySignal)
    f = EventTester(name='foo')
    assert not watcher.log
    f.destroySelf()
    assert watcher.log == [(f, [])]


def test_row_destroyed():
    setupClass(EventTester)
    watcher = make_listen(events.RowDestroyedSignal)
    f = EventTester(name='foo')
    assert not watcher.log
    f.destroySelf()
    assert watcher.log == [(f, [])]


def test_row_update():
    setupClass(EventTester)
    watcher = make_listen(events.RowUpdateSignal)
    f = EventTester(name='bar')
    assert not watcher.log
    f.name = 'bar2'
    f.set(name='bar3')
    assert watcher.log == [
        (f, {'name': 'bar2'}),
        (f, {'name': 'bar3'})]


def test_row_updated():
    setupClass(EventTester)
    watcher = make_listen(events.RowUpdatedSignal)
    f = EventTester(name='bar')
    assert not watcher.log
    f.name = 'bar2'
    f.set(name='bar3')
    assert watcher.log == [(f, []), (f, [])]


def test_add_column():
    setupClass(EventTester)
    watcher = make_listen(events.AddColumnSignal)
    events.summarize_events_by_sender()

    class NewEventTester(EventTester):
        name2 = StringCol()

    expect = (
        NewEventTester, None,
        'name2', NewEventTester.sqlmeta.columnDefinitions['name2'],
        False, [])
    print(zip(watcher.log[1], expect))
    assert watcher.log[1] == expect


class InheritableEventTestA(InheritableSQLObject):
    a = IntCol()


class InheritableEventTestB(InheritableEventTestA):
    b = IntCol()


class InheritableEventTestC(InheritableEventTestB):
    c = IntCol()


def _query(instance):
    row = InheritableEventTestA.get(instance.id)
    assert isinstance(row, InheritableEventTestC)
    assert row.c == 3


def _signal(instance, kwargs, postfuncs):
    postfuncs.append(_query)


def test_inheritance_row_created():
    setupClass([InheritableEventTestA, InheritableEventTestB,
                InheritableEventTestC])

    events.listen(_signal, InheritableEventTestA, events.RowCreatedSignal)

    InheritableEventTestC(a=1, b=2, c=3)