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
|
import pytest
import sqlalchemy as sa
from sqlalchemy_utils.observer import observes
@pytest.fixture
def Catalog(Base):
class Catalog(Base):
__tablename__ = 'catalog'
id = sa.Column(sa.Integer, primary_key=True)
product_price = sa.Column(sa.Integer)
@observes('category.sub_category.product')
def product_observer(self, product):
self.product_price = product.price if product else None
category = sa.orm.relationship(
'Category',
uselist=False,
backref='catalog'
)
return Catalog
@pytest.fixture
def Category(Base):
class Category(Base):
__tablename__ = 'category'
id = sa.Column(sa.Integer, primary_key=True)
catalog_id = sa.Column(sa.Integer, sa.ForeignKey('catalog.id'))
sub_category = sa.orm.relationship(
'SubCategory',
uselist=False,
backref='category'
)
return Category
@pytest.fixture
def SubCategory(Base):
class SubCategory(Base):
__tablename__ = 'sub_category'
id = sa.Column(sa.Integer, primary_key=True)
category_id = sa.Column(sa.Integer, sa.ForeignKey('category.id'))
product = sa.orm.relationship(
'Product',
uselist=False,
backref='sub_category'
)
return SubCategory
@pytest.fixture
def Product(Base):
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
price = sa.Column(sa.Integer)
sub_category_id = sa.Column(
sa.Integer, sa.ForeignKey('sub_category.id')
)
return Product
@pytest.fixture
def init_models(Catalog, Category, SubCategory, Product):
pass
@pytest.fixture
def catalog(session, Catalog, Category, SubCategory, Product):
sub_category = SubCategory(product=Product(price=123))
category = Category(sub_category=sub_category)
catalog = Catalog(category=category)
session.add(catalog)
session.flush()
return catalog
@pytest.mark.usefixtures('postgresql_dsn')
class TestObservesForOneToOneToOneToOne:
def test_simple_insert(self, catalog):
assert catalog.product_price == 123
def test_replace_leaf_object(self, catalog, session, Product):
product = Product(price=44)
catalog.category.sub_category.product = product
session.flush()
assert catalog.product_price == 44
def test_delete_leaf_object(self, catalog, session):
session.delete(catalog.category.sub_category.product)
session.flush()
assert catalog.product_price is None
|
