-
Notifications
You must be signed in to change notification settings - Fork 4
/
model_bridge_middleware_test.py
223 lines (173 loc) · 7.84 KB
/
model_bridge_middleware_test.py
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
import tests
import unittest
import pyperry
from pyperry.middlewares import ModelBridge
from pyperry.field import Field
from pyperry import caching
from tests.fixtures.test_adapter import TestAdapter
from tests.fixtures.test_adapter import SuccessAdapter
from tests.fixtures.test_adapter import FailureAdapter
from tests.fixtures.association_models import Test
class ModelBridgeBaseTestCase(unittest.TestCase):
pass
class ModelBridgeMiddlewareTestCase(ModelBridgeBaseTestCase):
def test_params(self):
"""should take an argument for next item in the call stack"""
mb = ModelBridge('foo')
self.assertEqual(mb.next, 'foo')
def test_optional_params(self):
"""should take optional configuration options"""
mb = ModelBridge('foo', { 'bar': 'baz' })
self.assertEqual(mb.options, { 'bar': 'baz' })
class ModelBridgeReadTestCase(ModelBridgeBaseTestCase):
def setUp(self):
class FakeAdapter(object):
def __init__(self):
self.called = False
self.args = None
self.return_value = [{ 'id': 1 }]
def __call__(self, **kwargs):
self.called = True
self.args = kwargs
return self.return_value
self.adapter = FakeAdapter()
self.bridge = ModelBridge(self.adapter, {})
self.relation = Test.scoped()
self.stack_opts = { 'relation': self.relation, 'mode': 'read' }
def test_call_stack(self):
"""should always call the next middleware/adapter in the stack"""
self.bridge(**self.stack_opts)
self.assertTrue(self.adapter.called)
def test_args_passing(self):
"""should pass kwargs to next middleware/adapter in the stack"""
self.bridge(**self.stack_opts)
self.assertTrue('relation' in self.adapter.args)
def test_init_records(self):
"""should create instances of the records returned from call"""
result = self.bridge(**self.stack_opts)
self.assertEqual(result[0].__class__, self.relation.klass)
def test_no_relation(self):
"""should do nothing to result if no relation is given"""
result = self.bridge(mode='read')
self.assertEqual(result, [{ 'id': 1 }])
def test_records_have_none_types(self):
"""should exclude none type records from results"""
self.adapter.return_value = [None, {'id': 1}, None]
result = self.bridge(**self.stack_opts)
self.assertEqual(len(result), 1)
self.assertEqual(result[0].fields, {'id': 1})
def test_new_record_false(self):
"""returned records should have their new_record attr set to false"""
result = self.bridge(**self.stack_opts)
self.assertEqual(result[0].new_record, False)
class BridgeTest(Test):
id = Field()
reader = TestAdapter()
class ModelBridgeWriteTestCase(ModelBridgeBaseTestCase):
def setUp(self):
self.model_class = BridgeTest
self.model_class.reader = TestAdapter()
self.model = self.model_class({})
self.model.reader.data = { 'id': 42 }
self.options = { 'model': self.model, 'mode': 'write' }
def tearDown(self):
try:
self.model.reader.reset_calls()
except:
pass
def test_resets_caching(self):
self.called_foo = False
def foo():
self.called_foo = True
caching.register(foo)
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(self.called_foo, True)
class WriteExistingRecordsTestCase(ModelBridgeWriteTestCase):
def test_saved_on_success(self):
"""should set model's saved attribute to true on success"""
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(self.model.saved, True)
def test_saved_on_fail(self):
"""should set model's saved attribute to false on failure"""
ModelBridge(FailureAdapter())(**self.options)
self.assertEqual(self.model.saved, False)
def test_reload_on_success(self):
"""should reload the model on success"""
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(len(TestAdapter.calls), 1)
def test_reload_on_fail(self):
"""should not reload the model on failure"""
ModelBridge(FailureAdapter())(**self.options)
self.assertEqual(len(TestAdapter.calls), 0)
def test_errors_on_fail(self):
"""should add error messages to the model on failure"""
adapter = FailureAdapter()
adapter.response.parsed = { 'errors': { 'base': 'record invalid' } }
ModelBridge(adapter)(**self.options)
self.assertEqual(self.model.errors, { 'base': 'record invalid' })
def test_always_errors_on_fail(self):
"""should add error messages to the model even if the response has no
errors"""
adapter = FailureAdapter()
adapter.response.parsed = None
ModelBridge(adapter)(**self.options)
self.assertEqual(self.model.errors, { 'base': 'record not saved' })
def test_skip_reload(self):
"""should not reload the model if no read adapter is available"""
del self.model_class.reader
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(len(TestAdapter.calls), 0)
class WriteNewRecordsTestCase(ModelBridgeWriteTestCase):
def setUp(self):
super(WriteNewRecordsTestCase, self).setUp()
self.model.new_record = True
def test_set_key(self):
"""should set the model's primary_key attribute on success"""
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(self.model.id, 42)
def test_new_record_success(self):
"""should set the model's new_record attribute to false on success"""
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(self.model.new_record, False)
def test_new_record_fail(self):
"""should keep the model's new_record attribute as true on failure"""
ModelBridge(FailureAdapter())(**self.options)
self.assertEqual(self.model.new_record, True)
def test_exception_when_no_key(self):
"""should thrown an exception if the response does not have a value for
the primary_key attribute"""
adapter = SuccessAdapter()
adapter.response.parsed = None
bridge = ModelBridge(adapter)
self.assertRaises(KeyError, bridge, **self.options)
def test_no_exception_without_read(self):
"""should not throw an exception if no read adapter is configured"""
del self.model_class.reader
adapter = SuccessAdapter()
adapter.response.parsed = None
ModelBridge(adapter)(**self.options)
class ModelBridgeDeleteTestCase(ModelBridgeWriteTestCase):
def setUp(self):
super(ModelBridgeDeleteTestCase, self).setUp()
self.options['mode'] = 'delete'
def test_freeze(self):
"""should freeze the model on success"""
ModelBridge(SuccessAdapter())(**self.options)
self.assertEqual(self.model.frozen(), True)
def test_not_freeze_on_fail(self):
"""should not freeze the model on failure"""
ModelBridge(FailureAdapter())(**self.options)
self.assertEqual(self.model.frozen(), False)
def test_add_errors(self):
"""should add error messages to the model on failure"""
adapter = FailureAdapter()
adapter.response.parsed = { 'errors': { 'base': 'record not found' } }
ModelBridge(adapter)(**self.options)
self.assertEqual(self.model.errors, { 'base': 'record not found' })
def test_add_default_errors(self):
"""should add a default error message to the model on failure even when
the response has no error messages"""
adapter = FailureAdapter()
adapter.response.parsed = None
ModelBridge(adapter)(**self.options)
self.assertEqual(self.model.errors, { 'base': 'record not deleted' })