[devtools/testlib,fill] use the new connection api (for auto_populate)
Note that this changes API of CubicWebTC's custom_populate and
post_populate methods to take a connection instead of a cursor, which
may affect some cubes.
--- a/devtools/fill.py Wed Jul 02 14:42:24 2014 +0200
+++ b/devtools/fill.py Wed Jul 02 15:11:00 2014 +0200
@@ -1,5 +1,5 @@
# -*- coding: iso-8859-1 -*-
-# copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
#
# This file is part of CubicWeb.
@@ -352,7 +352,7 @@
-def select(constraints, cursor, selectvar='O', objtype=None):
+def select(constraints, cnx, selectvar='O', objtype=None):
"""returns list of eids matching <constraints>
<selectvar> should be either 'O' or 'S' to match schema definitions
@@ -361,7 +361,7 @@
rql = 'Any %s WHERE %s' % (selectvar, constraints)
if objtype:
rql += ', %s is %s' % (selectvar, objtype)
- rset = cursor.execute(rql)
+ rset = cnx.execute(rql)
except Exception:
print "could restrict eid_list with given constraints (%r)" % constraints
return []
@@ -369,7 +369,7 @@
-def make_relations_queries(schema, edict, cursor, ignored_relations=(),
+def make_relations_queries(schema, edict, cnx, ignored_relations=(),
existingrels=None):
"""returns a list of generated RQL queries for relations
:param schema: The instance schema
@@ -379,7 +379,7 @@
:param ignored_relations: list of relations to ignore (i.e. don't try
to generate insert queries for these relations)
"""
- gen = RelationsQueriesGenerator(schema, cursor, existingrels)
+ gen = RelationsQueriesGenerator(schema, cnx, existingrels)
return gen.compute_queries(edict, ignored_relations)
def composite_relation(rschema):
@@ -393,9 +393,9 @@
class RelationsQueriesGenerator(object):
rql_tmpl = 'SET S %s O WHERE S eid %%(subjeid)s, O eid %%(objeid)s'
- def __init__(self, schema, cursor, existing=None):
+ def __init__(self, schema, cnx, existing=None):
self.schema = schema
- self.cursor = cursor
+ self.cnx = cnx
self.existingrels = existing or {}
def compute_queries(self, edict, ignored_relations):
@@ -457,7 +457,7 @@
# restrict object eids if possible
# XXX the attempt to restrict below in completely wrong
# disabling it for now
- objeids = select(restrictions, self.cursor, objtype=obj)
+ objeids = select(restrictions, self.cnx, objtype=obj)
else:
objeids = oedict.get(obj, frozenset())
if subjcard in '?1' or objcard in '?1':
--- a/devtools/testlib.py Wed Jul 02 14:42:24 2014 +0200
+++ b/devtools/testlib.py Wed Jul 02 15:11:00 2014 +0200
@@ -1179,7 +1179,7 @@
# XXX cleanup unprotected_entities & all mess
-def how_many_dict(schema, cursor, how_many, skip):
+def how_many_dict(schema, cnx, how_many, skip):
"""given a schema, compute how many entities by type we need to be able to
satisfy relations cardinality.
@@ -1213,7 +1213,7 @@
# step 1, compute a base number of each entity types: number of already
# existing entities of this type + `how_many`
for etype in unprotected_entities(schema, strict=True):
- howmanydict[str(etype)] = cursor.execute('Any COUNT(X) WHERE X is %s' % etype)[0][0]
+ howmanydict[str(etype)] = cnx.execute('Any COUNT(X) WHERE X is %s' % etype)[0][0]
if etype in unprotected:
howmanydict[str(etype)] += how_many
# step 2, augment nb entity per types to satisfy cardinality constraints,
@@ -1247,10 +1247,10 @@
def to_test_etypes(self):
return unprotected_entities(self.schema, strict=True)
- def custom_populate(self, how_many, cursor):
+ def custom_populate(self, how_many, cnx):
pass
- def post_populate(self, cursor):
+ def post_populate(self, cnx):
pass
@@ -1259,42 +1259,42 @@
"""this method populates the database with `how_many` entities
of each possible type. It also inserts random relations between them
"""
- with self.session.security_enabled(read=False, write=False):
- self._auto_populate(how_many)
+ with self.admin_access.repo_cnx() as cnx:
+ with cnx.security_enabled(read=False, write=False):
+ self._auto_populate(cnx, how_many)
+ cnx.commit()
- def _auto_populate(self, how_many):
- cu = self.cursor()
- self.custom_populate(how_many, cu)
+ def _auto_populate(self, cnx, how_many):
+ self.custom_populate(how_many, cnx)
vreg = self.vreg
- howmanydict = how_many_dict(self.schema, cu, how_many, self.no_auto_populate)
+ howmanydict = how_many_dict(self.schema, cnx, how_many, self.no_auto_populate)
for etype in unprotected_entities(self.schema):
if etype in self.no_auto_populate:
continue
nb = howmanydict.get(etype, how_many)
for rql, args in insert_entity_queries(etype, self.schema, vreg, nb):
- cu.execute(rql, args)
+ cnx.execute(rql, args)
edict = {}
for etype in unprotected_entities(self.schema, strict=True):
- rset = cu.execute('%s X' % etype)
+ rset = cnx.execute('%s X' % etype)
edict[str(etype)] = set(row[0] for row in rset.rows)
existingrels = {}
ignored_relations = SYSTEM_RELATIONS | self.ignored_relations
for rschema in self.schema.relations():
if rschema.final or rschema in ignored_relations:
continue
- rset = cu.execute('DISTINCT Any X,Y WHERE X %s Y' % rschema)
+ rset = cnx.execute('DISTINCT Any X,Y WHERE X %s Y' % rschema)
existingrels.setdefault(rschema.type, set()).update((x, y) for x, y in rset)
- q = make_relations_queries(self.schema, edict, cu, ignored_relations,
+ q = make_relations_queries(self.schema, edict, cnx, ignored_relations,
existingrels=existingrels)
for rql, args in q:
try:
- cu.execute(rql, args)
+ cnx.execute(rql, args)
except ValidationError as ex:
# failed to satisfy some constraint
print 'error in automatic db population', ex
- self.session.commit_state = None # reset uncommitable flag
- self.post_populate(cu)
- self.commit()
+ cnx.commit_state = None # reset uncommitable flag
+ self.post_populate(cnx)
def iter_individual_rsets(self, etypes=None, limit=None):
etypes = etypes or self.to_test_etypes()