1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
import pytest
import sqlalchemy as sa
from sqlalchemy_utils.observer import observes
@pytest.mark.usefixtures('postgresql_dsn')
class TestObservesForColumn:
@pytest.fixture
def Product(self, Base):
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
price = sa.Column(sa.Integer)
@observes('price')
def product_price_observer(self, price):
self.price = price * 2
return Product
@pytest.fixture
def init_models(self, Product):
pass
def test_simple_insert(self, session, Product):
product = Product(price=100)
session.add(product)
session.flush()
assert product.price == 200
@pytest.mark.usefixtures('postgresql_dsn')
class TestObservesForColumnWithoutActualChanges:
@pytest.fixture
def Product(self, Base):
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
price = sa.Column(sa.Integer)
@observes('price')
def product_price_observer(self, price):
raise Exception('Trying to change price')
return Product
@pytest.fixture
def init_models(self, Product):
pass
def test_only_notifies_observer_on_actual_changes(self, session, Product):
product = Product()
session.add(product)
session.flush()
with pytest.raises(Exception) as e:
product.price = 500
session.commit()
assert str(e.value) == 'Trying to change price'
@pytest.mark.usefixtures('postgresql_dsn')
class TestObservesForMultipleColumns:
@pytest.fixture
def Order(self, Base):
class Order(Base):
__tablename__ = 'order'
id = sa.Column(sa.Integer, primary_key=True)
unit_price = sa.Column(sa.Integer)
amount = sa.Column(sa.Integer)
total_price = sa.Column(sa.Integer)
@observes('amount', 'unit_price')
def total_price_observer(self, amount, unit_price):
self.total_price = amount * unit_price
return Order
@pytest.fixture
def init_models(self, Order):
pass
def test_only_notifies_observer_on_actual_changes(self, session, Order):
order = Order()
order.amount = 2
order.unit_price = 10
session.add(order)
session.flush()
order.amount = 1
session.flush()
assert order.total_price == 10
order.unit_price = 100
session.flush()
assert order.total_price == 100
@pytest.mark.usefixtures('postgresql_dsn')
class TestObservesForMultipleColumnsFiresOnlyOnce:
@pytest.fixture
def Order(self, Base):
class Order(Base):
__tablename__ = 'order'
id = sa.Column(sa.Integer, primary_key=True)
unit_price = sa.Column(sa.Integer)
amount = sa.Column(sa.Integer)
@observes('amount', 'unit_price')
def total_price_observer(self, amount, unit_price):
self.call_count = self.call_count + 1
return Order
@pytest.fixture
def init_models(self, Order):
pass
def test_only_notifies_observer_on_actual_changes(self, session, Order):
order = Order()
order.amount = 2
order.unit_price = 10
order.call_count = 0
session.add(order)
session.flush()
assert order.call_count == 1
order.amount = 1
order.unit_price = 100
session.flush()
assert order.call_count == 2
|
