1 """Hidden internals for the devtools.apptest module |
|
2 |
|
3 :organization: Logilab |
|
4 :copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2. |
|
5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
|
6 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses |
|
7 """ |
|
8 __docformat__ = "restructuredtext en" |
|
9 |
|
10 import sys, traceback |
|
11 |
|
12 from logilab.common.pytest import pause_tracing, resume_tracing |
|
13 |
|
14 import yams.schema |
|
15 |
|
16 from cubicweb.dbapi import repo_connect, ConnectionProperties, ProgrammingError |
|
17 from cubicweb.cwvreg import CubicWebVRegistry |
|
18 |
|
19 from cubicweb.web.application import CubicWebPublisher |
|
20 from cubicweb.web import Redirect |
|
21 |
|
22 from cubicweb.devtools import ApptestConfiguration, init_test_database |
|
23 from cubicweb.devtools.fake import FakeRequest |
|
24 |
|
25 SYSTEM_ENTITIES = ('CWGroup', 'CWUser', |
|
26 'CWAttribute', 'CWRelation', |
|
27 'CWConstraint', 'CWConstraintType', 'CWProperty', |
|
28 'CWEType', 'CWRType', |
|
29 'State', 'Transition', 'TrInfo', |
|
30 'RQLExpression', |
|
31 ) |
|
32 SYSTEM_RELATIONS = ( |
|
33 # virtual relation |
|
34 'identity', |
|
35 # metadata |
|
36 'is', 'is_instance_of', 'owned_by', 'created_by', 'specializes', |
|
37 # workflow related |
|
38 'state_of', 'transition_of', 'initial_state', 'allowed_transition', |
|
39 'destination_state', 'in_state', 'wf_info_for', 'from_state', 'to_state', |
|
40 'condition', |
|
41 # permission |
|
42 'in_group', 'require_group', 'require_permission', |
|
43 'read_permission', 'update_permission', 'delete_permission', 'add_permission', |
|
44 # eproperty |
|
45 'for_user', |
|
46 # schema definition |
|
47 'relation_type', 'from_entity', 'to_entity', |
|
48 'constrained_by', 'cstrtype', 'widget', |
|
49 # deducted from other relations |
|
50 'primary_email', |
|
51 ) |
|
52 |
|
53 def unprotected_entities(app_schema, strict=False): |
|
54 """returned a Set of each non final entity type, excluding CWGroup, and CWUser... |
|
55 """ |
|
56 if strict: |
|
57 protected_entities = yams.schema.BASE_TYPES |
|
58 else: |
|
59 protected_entities = yams.schema.BASE_TYPES.union(set(SYSTEM_ENTITIES)) |
|
60 entities = set(app_schema.entities()) |
|
61 return entities - protected_entities |
|
62 |
|
63 |
|
64 def ignore_relations(*relations): |
|
65 global SYSTEM_RELATIONS |
|
66 SYSTEM_RELATIONS += relations |
|
67 |
|
68 |
|
69 class TestEnvironment(object): |
|
70 """TestEnvironment defines a context (e.g. a config + a given connection) in |
|
71 which the tests are executed |
|
72 """ |
|
73 |
|
74 def __init__(self, appid, reporter=None, verbose=False, |
|
75 configcls=ApptestConfiguration, requestcls=FakeRequest): |
|
76 config = configcls(appid) |
|
77 self.requestcls = requestcls |
|
78 self.cnx = None |
|
79 config.db_perms = False |
|
80 source = config.sources()['system'] |
|
81 if verbose: |
|
82 print "init test database ..." |
|
83 self.vreg = vreg = CubicWebVRegistry(config) |
|
84 self.admlogin = source['db-user'] |
|
85 # restore database <=> init database |
|
86 self.restore_database() |
|
87 if verbose: |
|
88 print "init done" |
|
89 config.repository = lambda x=None: self.repo |
|
90 self.app = CubicWebPublisher(config, vreg=vreg) |
|
91 self.verbose = verbose |
|
92 schema = self.vreg.schema |
|
93 # else we may run into problems since email address are ususally share in app tests |
|
94 # XXX should not be necessary anymore |
|
95 schema.rschema('primary_email').set_rproperty('CWUser', 'EmailAddress', 'composite', False) |
|
96 self.deletable_entities = unprotected_entities(schema) |
|
97 |
|
98 def restore_database(self): |
|
99 """called by unittests' tearDown to restore the original database |
|
100 """ |
|
101 try: |
|
102 pause_tracing() |
|
103 if self.cnx: |
|
104 self.cnx.close() |
|
105 source = self.vreg.config.sources()['system'] |
|
106 self.repo, self.cnx = init_test_database(driver=source['db-driver'], |
|
107 vreg=self.vreg) |
|
108 self._orig_cnx = self.cnx |
|
109 resume_tracing() |
|
110 except: |
|
111 resume_tracing() |
|
112 traceback.print_exc() |
|
113 sys.exit(1) |
|
114 # XXX cnx decoration is usually done by the repository authentication manager, |
|
115 # necessary in authentication tests |
|
116 self.cnx.vreg = self.vreg |
|
117 self.cnx.login = source['db-user'] |
|
118 self.cnx.password = source['db-password'] |
|
119 |
|
120 |
|
121 def create_user(self, login, groups=('users',), req=None): |
|
122 req = req or self.create_request() |
|
123 cursor = self._orig_cnx.cursor(req) |
|
124 rset = cursor.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s,' |
|
125 'X in_state S WHERE S name "activated"', |
|
126 {'login': unicode(login), 'passwd': login.encode('utf8')}) |
|
127 user = rset.get_entity(0, 0) |
|
128 cursor.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)' |
|
129 % ','.join(repr(g) for g in groups), |
|
130 {'x': user.eid}, 'x') |
|
131 user.clear_related_cache('in_group', 'subject') |
|
132 self._orig_cnx.commit() |
|
133 return user |
|
134 |
|
135 def login(self, login, password=None): |
|
136 if login == self.admlogin: |
|
137 self.restore_connection() |
|
138 else: |
|
139 self.cnx = repo_connect(self.repo, unicode(login), |
|
140 password or str(login), |
|
141 ConnectionProperties('inmemory')) |
|
142 if login == self.vreg.config.anonymous_user()[0]: |
|
143 self.cnx.anonymous_connection = True |
|
144 return self.cnx |
|
145 |
|
146 def restore_connection(self): |
|
147 if not self.cnx is self._orig_cnx: |
|
148 try: |
|
149 self.cnx.close() |
|
150 except ProgrammingError: |
|
151 pass # already closed |
|
152 self.cnx = self._orig_cnx |
|
153 |
|
154 ############################################################################ |
|
155 |
|
156 def execute(self, rql, args=None, eidkey=None, req=None): |
|
157 """executes <rql>, builds a resultset, and returns a couple (rset, req) |
|
158 where req is a FakeRequest |
|
159 """ |
|
160 req = req or self.create_request(rql=rql) |
|
161 return self.cnx.cursor(req).execute(unicode(rql), args, eidkey) |
|
162 |
|
163 def create_request(self, rql=None, **kwargs): |
|
164 """executes <rql>, builds a resultset, and returns a |
|
165 couple (rset, req) where req is a FakeRequest |
|
166 """ |
|
167 if rql: |
|
168 kwargs['rql'] = rql |
|
169 req = self.requestcls(self.vreg, form=kwargs) |
|
170 req.set_connection(self.cnx) |
|
171 return req |
|
172 |
|
173 def get_rset_and_req(self, rql, optional_args=None, args=None, eidkey=None): |
|
174 """executes <rql>, builds a resultset, and returns a |
|
175 couple (rset, req) where req is a FakeRequest |
|
176 """ |
|
177 return (self.execute(rql, args, eidkey), |
|
178 self.create_request(rql=rql, **optional_args or {})) |
|
179 |
|
180 |
|
181 class ExistingTestEnvironment(TestEnvironment): |
|
182 |
|
183 def __init__(self, appid, sourcefile, verbose=False): |
|
184 config = ApptestConfiguration(appid, sourcefile=sourcefile) |
|
185 if verbose: |
|
186 print "init test database ..." |
|
187 source = config.sources()['system'] |
|
188 self.vreg = CubicWebVRegistry(config) |
|
189 self.cnx = init_test_database(driver=source['db-driver'], |
|
190 vreg=self.vreg)[1] |
|
191 if verbose: |
|
192 print "init done" |
|
193 self.app = CubicWebPublisher(config, vreg=self.vreg) |
|
194 self.verbose = verbose |
|
195 # this is done when the publisher is opening a connection |
|
196 self.cnx.vreg = self.vreg |
|
197 |
|
198 def setup(self, config=None): |
|
199 """config is passed by TestSuite but is ignored in this environment""" |
|
200 cursor = self.cnx.cursor() |
|
201 self.last_eid = cursor.execute('Any X WHERE X creation_date D ORDERBY D DESC LIMIT 1').rows[0][0] |
|
202 |
|
203 def cleanup(self): |
|
204 """cancel inserted elements during tests""" |
|
205 cursor = self.cnx.cursor() |
|
206 cursor.execute('DELETE Any X WHERE X eid > %(x)s', {'x' : self.last_eid}, eid_key='x') |
|
207 print "cleaning done" |
|
208 self.cnx.commit() |
|