36 |
36 |
37 def __call__(self): |
37 def __call__(self): |
38 # new value of entity.data should be the same as before |
38 # new value of entity.data should be the same as before |
39 oldvalue = self._cw.transaction_data['orig_file_value'] |
39 oldvalue = self._cw.transaction_data['orig_file_value'] |
40 assert oldvalue == self.entity.data.getvalue() |
40 assert oldvalue == self.entity.data.getvalue() |
41 |
|
42 |
41 |
43 class StorageTC(CubicWebTC): |
42 class StorageTC(CubicWebTC): |
44 |
43 |
45 def setup_database(self): |
44 def setup_database(self): |
46 self.tempdir = tempfile.mkdtemp() |
45 self.tempdir = tempfile.mkdtemp() |
86 {'f': f1.eid})[0][0] |
85 {'f': f1.eid})[0][0] |
87 self.assertEquals(fspath.getvalue(), expected_filepath) |
86 self.assertEquals(fspath.getvalue(), expected_filepath) |
88 |
87 |
89 def test_bfss_fs_importing_doesnt_touch_path(self): |
88 def test_bfss_fs_importing_doesnt_touch_path(self): |
90 self.session.transaction_data['fs_importing'] = True |
89 self.session.transaction_data['fs_importing'] = True |
91 f1 = self.session.create_entity('File', data=Binary('/the/path'), |
90 filepath = osp.abspath(__file__) |
|
91 f1 = self.session.create_entity('File', data=Binary(filepath), |
92 data_format=u'text/plain', data_name=u'foo') |
92 data_format=u'text/plain', data_name=u'foo') |
93 fspath = self.execute('Any fspath(D) WHERE F eid %(f)s, F data D', |
93 fspath = self.execute('Any fspath(D) WHERE F eid %(f)s, F data D', |
94 {'f': f1.eid})[0][0] |
94 {'f': f1.eid})[0][0] |
95 self.assertEquals(fspath.getvalue(), '/the/path') |
95 self.assertEquals(fspath.getvalue(), filepath) |
96 |
96 |
97 def test_source_storage_transparency(self): |
97 def test_source_storage_transparency(self): |
98 with self.temporary_appobjects(DummyBeforeHook, DummyAfterHook): |
98 with self.temporary_appobjects(DummyBeforeHook, DummyAfterHook): |
99 self.create_file() |
99 self.create_file() |
100 |
100 |
154 ex = self.assertRaises(QueryError, self.execute, |
154 ex = self.assertRaises(QueryError, self.execute, |
155 'Any X,UPPER(D) WHERE X eid %(x)s, X data D', |
155 'Any X,UPPER(D) WHERE X eid %(x)s, X data D', |
156 {'x': f1.eid}, 'x') |
156 {'x': f1.eid}, 'x') |
157 self.assertEquals(str(ex), 'UPPER can not be called on mapped attribute') |
157 self.assertEquals(str(ex), 'UPPER can not be called on mapped attribute') |
158 |
158 |
|
159 |
|
160 def test_bfss_fs_importing_transparency(self): |
|
161 self.session.transaction_data['fs_importing'] = True |
|
162 filepath = osp.abspath(__file__) |
|
163 f1 = self.session.create_entity('File', data=Binary(filepath), |
|
164 data_format=u'text/plain', data_name=u'foo') |
|
165 self.assertEquals(f1.data.getvalue(), file(filepath).read(), |
|
166 'files content differ') |
|
167 |
|
168 |
|
169 def test_bfss_update_with_existing_data(self): |
|
170 # use self.session to use server-side cache |
|
171 f1 = self.session.create_entity('File', data=Binary('some data'), |
|
172 data_format=u'text/plain', data_name=u'foo') |
|
173 # NOTE: do not use set_attributes() which would automatically |
|
174 # update f1's local dict. We want the pure rql version to work |
|
175 self.execute('SET F data %(d)s WHERE F eid %(f)s', |
|
176 {'d': Binary('some other data'), 'f': f1.eid}) |
|
177 self.assertEquals(f1.data.getvalue(), 'some other data') |
|
178 self.commit() |
|
179 f2 = self.entity('Any F WHERE F eid %(f)s, F is File', {'f': f1.eid}) |
|
180 self.assertEquals(f2.data.getvalue(), 'some other data') |
|
181 |
|
182 |
159 if __name__ == '__main__': |
183 if __name__ == '__main__': |
160 unittest_main() |
184 unittest_main() |