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
|
import sqlalchemy as sa
from flexmock import flexmock
from sqlalchemy_utils import proxy_dict, ProxyDict
from tests import TestCase
class TestProxyDict(TestCase):
def create_models(self):
class Article(self.Base):
__tablename__ = 'article'
id = sa.Column(sa.Integer, autoincrement=True, primary_key=True)
description = sa.Column(sa.UnicodeText)
_translations = sa.orm.relationship(
'ArticleTranslation',
lazy='dynamic',
cascade='all, delete-orphan',
passive_deletes=True,
backref=sa.orm.backref('parent'),
)
@property
def translations(self):
return proxy_dict(
self,
'_translations',
ArticleTranslation.locale
)
class ArticleTranslation(self.Base):
__tablename__ = 'article_translation'
id = sa.Column(
sa.Integer,
sa.ForeignKey(Article.id),
autoincrement=True,
primary_key=True
)
locale = sa.Column(sa.String(10), primary_key=True)
name = sa.Column(sa.UnicodeText)
self.Article = Article
self.ArticleTranslation = ArticleTranslation
def test_access_key_for_pending_parent(self):
article = self.Article()
self.session.add(article)
assert article.translations['en']
def test_access_key_for_transient_parent(self):
article = self.Article()
assert article.translations['en']
def test_cache(self):
article = self.Article()
(
flexmock(ProxyDict)
.should_receive('fetch')
.once()
)
self.session.add(article)
self.session.commit()
article.translations['en']
article.translations['en']
def test_set_updates_cache(self):
article = self.Article()
(
flexmock(ProxyDict)
.should_receive('fetch')
.once()
)
self.session.add(article)
self.session.commit()
article.translations['en']
article.translations['en'] = self.ArticleTranslation(
locale='en',
name=u'something'
)
article.translations['en']
def test_contains_efficiency(self):
article = self.Article()
self.session.add(article)
self.session.commit()
article.id
query_count = self.connection.query_count
'en' in article.translations
'en' in article.translations
'en' in article.translations
assert self.connection.query_count == query_count + 1
def test_getitem_with_none_value_in_cache(self):
article = self.Article()
self.session.add(article)
self.session.commit()
article.id
'en' in article.translations
assert article.translations['en']
def test_contains(self):
article = self.Article()
assert 'en' not in article.translations
# does not auto-append new translation
assert 'en' not in article.translations
def test_committing_session_empties_proxy_dict_cache(self):
article = self.Article()
(
flexmock(ProxyDict)
.should_receive('fetch')
.twice()
)
self.session.add(article)
self.session.commit()
article.translations['en']
self.session.commit()
article.translations['en']
|
