30 from logilab.common.testlib import TestCase, unittest_main |
30 from logilab.common.testlib import TestCase, unittest_main |
31 |
31 |
32 from yams.constraints import UniqueConstraint |
32 from yams.constraints import UniqueConstraint |
33 |
33 |
34 from cubicweb import (BadConnectionId, RepositoryError, ValidationError, |
34 from cubicweb import (BadConnectionId, RepositoryError, ValidationError, |
35 UnknownEid, AuthenticationError) |
35 UnknownEid, AuthenticationError, Unauthorized) |
36 from cubicweb.selectors import is_instance |
36 from cubicweb.selectors import is_instance |
37 from cubicweb.schema import CubicWebSchema, RQLConstraint |
37 from cubicweb.schema import CubicWebSchema, RQLConstraint |
38 from cubicweb.dbapi import connect, multiple_connections_unfix |
38 from cubicweb.dbapi import connect, multiple_connections_unfix |
39 from cubicweb.devtools.testlib import CubicWebTC |
39 from cubicweb.devtools.testlib import CubicWebTC |
40 from cubicweb.devtools.repotest import tuplify |
40 from cubicweb.devtools.repotest import tuplify |
134 {'login': u"barnabé", 'passwd': u"héhéhé".encode('UTF8')}) |
134 {'login': u"barnabé", 'passwd': u"héhéhé".encode('UTF8')}) |
135 repo.commit(cnxid) |
135 repo.commit(cnxid) |
136 repo.close(cnxid) |
136 repo.close(cnxid) |
137 self.assert_(repo.connect(u"barnabé", password=u"héhéhé".encode('UTF8'))) |
137 self.assert_(repo.connect(u"barnabé", password=u"héhéhé".encode('UTF8'))) |
138 |
138 |
139 def test_invalid_entity_rollback(self): |
139 def test_rollback_on_commit_error(self): |
140 cnxid = self.repo.connect(self.admlogin, password=self.admpassword) |
140 cnxid = self.repo.connect(self.admlogin, password=self.admpassword) |
141 # no group |
|
142 self.repo.execute(cnxid, |
141 self.repo.execute(cnxid, |
143 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s', |
142 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s', |
144 {'login': u"tutetute", 'passwd': 'tutetute'}) |
143 {'login': u"tutetute", 'passwd': 'tutetute'}) |
145 self.assertRaises(ValidationError, self.repo.commit, cnxid) |
144 self.assertRaises(ValidationError, self.repo.commit, cnxid) |
146 self.failIf(self.repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"')) |
145 self.failIf(self.repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"')) |
|
146 |
|
147 def test_rollback_on_execute_validation_error(self): |
|
148 class ValidationErrorAfterHook(Hook): |
|
149 __regid__ = 'valerror-after-hook' |
|
150 __select__ = Hook.__select__ & is_instance('CWGroup') |
|
151 events = ('after_update_entity',) |
|
152 def __call__(self): |
|
153 raise ValidationError(self.entity.eid, {}) |
|
154 with self.temporary_appobjects(ValidationErrorAfterHook): |
|
155 self.assertRaises(ValidationError, |
|
156 self.execute, 'SET X name "toto" WHERE X is CWGroup, X name "guests"') |
|
157 self.failIf(self.execute('Any X WHERE X is CWGroup, X name "toto"')) |
|
158 |
|
159 def test_rollback_on_execute_unauthorized(self): |
|
160 class UnauthorizedAfterHook(Hook): |
|
161 __regid__ = 'valerror-after-hook' |
|
162 __select__ = Hook.__select__ & is_instance('CWGroup') |
|
163 events = ('after_update_entity',) |
|
164 def __call__(self): |
|
165 raise Unauthorized() |
|
166 with self.temporary_appobjects(UnauthorizedAfterHook): |
|
167 self.assertRaises(Unauthorized, |
|
168 self.execute, 'SET X name "toto" WHERE X is CWGroup, X name "guests"') |
|
169 self.failIf(self.execute('Any X WHERE X is CWGroup, X name "toto"')) |
|
170 |
147 |
171 |
148 def test_close(self): |
172 def test_close(self): |
149 repo = self.repo |
173 repo = self.repo |
150 cnxid = repo.connect(self.admlogin, password=self.admpassword) |
174 cnxid = repo.connect(self.admlogin, password=self.admpassword) |
151 self.assert_(cnxid) |
175 self.assert_(cnxid) |