Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
cubicweb
cubicweb
Commits
b2530e3e0afb
Commit
c112dc2b
authored
Aug 11, 2009
by
Sylvain Thénault
Browse files
[testlib] #345052 and #344207: major test lib refactoring/cleanup + update usage
parent
58c519e5a31f
Changes
57
Expand all
Hide whitespace changes
Inline
Side-by-side
common/test/unittest_mail.py
View file @
b2530e3e
...
...
@@ -13,7 +13,7 @@ import pwd
from
logilab.common.testlib
import
unittest_main
from
logilab.common.umessage
import
message_from_string
from
cubicweb.devtools.
app
test
import
EnvBased
TC
from
cubicweb.devtools.test
lib
import
CubicWeb
TC
from
cubicweb.common.mail
import
format_mail
...
...
@@ -25,7 +25,7 @@ def getlogin():
return
pwd
.
getpwuid
(
os
.
getuid
())[
0
]
class
EmailTC
(
EnvBased
TC
):
class
EmailTC
(
CubicWeb
TC
):
def
test_format_mail
(
self
):
self
.
set_option
(
'sender-addr'
,
'bim@boum.fr'
)
...
...
common/test/unittest_migration.py
View file @
b2530e3e
...
...
@@ -10,8 +10,6 @@ from os.path import abspath
from
logilab.common.testlib
import
TestCase
,
unittest_main
from
cubicweb.devtools
import
TestServerConfiguration
from
cubicweb.devtools.apptest
import
TestEnvironment
from
cubicweb.cwconfig
import
CubicWebConfiguration
from
cubicweb.common.migration
import
MigrationHelper
,
filter_scripts
from
cubicweb.server.migractions
import
ServerMigrationHelper
...
...
@@ -98,8 +96,8 @@ class BaseCreationTC(TestCase):
config
=
ApptestConfiguration
(
'data'
)
source
=
config
.
sources
()[
'system'
]
self
.
assertEquals
(
source
[
'db-driver'
],
'sqlite'
)
cleanup_sqlite
(
source
[
'db-name'
],
remove
cub
e
=
True
)
init_test_database
(
driver
=
source
[
'db-driver'
],
config
=
config
)
cleanup_sqlite
(
source
[
'db-name'
],
remove
templat
e
=
True
)
init_test_database
(
config
=
config
)
if
__name__
==
'__main__'
:
...
...
common/test/unittest_mixins.py
View file @
b2530e3e
...
...
@@ -6,9 +6,9 @@
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
from
logilab.common.testlib
import
unittest_main
from
cubicweb.devtools.
app
test
import
EnvBased
TC
from
cubicweb.devtools.test
lib
import
CubicWeb
TC
class
WorkfloableMixInTC
(
EnvBased
TC
):
class
WorkfloableMixInTC
(
CubicWeb
TC
):
def
test_wf_state
(
self
):
s
=
self
.
add_entity
(
'State'
,
name
=
u
'activated'
)
self
.
execute
(
'SET X state_of ET WHERE ET name "Bookmark", X eid %(x)s'
,
...
...
devtools/__init__.py
View file @
b2530e3e
...
...
@@ -13,20 +13,54 @@ from datetime import timedelta
from
os.path
import
(
abspath
,
join
,
exists
,
basename
,
dirname
,
normpath
,
split
,
isfile
,
isabs
)
from
cubicweb
import
CW_SOFTWARE_ROOT
,
ConfigurationError
from
cubicweb
import
CW_SOFTWARE_ROOT
,
ConfigurationError
,
schema
,
cwconfig
from
cubicweb.utils
import
strptime
from
cubicweb.toolsutils
import
read_config
from
cubicweb.cwconfig
import
CubicWebConfiguration
,
merge_options
from
cubicweb.server.serverconfig
import
ServerConfiguration
from
cubicweb.etwist.twconfig
import
TwistedConfiguration
cwconfig
.
CubicWebConfiguration
.
cls_adjust_sys_path
()
# db auto-population configuration #############################################
SYSTEM_ENTITIES
=
schema
.
SCHEMA_TYPES
|
set
((
'CWGroup'
,
'CWUser'
,
'CWProperty'
,
'State'
,
'Transition'
,
'TrInfo'
,
))
SYSTEM_RELATIONS
=
schema
.
META_RTYPES
|
set
((
# workflow related
'state_of'
,
'transition_of'
,
'initial_state'
,
'allowed_transition'
,
'destination_state'
,
'in_state'
,
'wf_info_for'
,
'from_state'
,
'to_state'
,
'condition'
,
# cwproperty
'for_user'
,
# schema definition
'specializes'
,
'relation_type'
,
'from_entity'
,
'to_entity'
,
'constrained_by'
,
'cstrtype'
,
'widget'
,
'read_permission'
,
'update_permission'
,
'delete_permission'
,
'add_permission'
,
# permission
'in_group'
,
'require_group'
,
'require_permission'
,
# deducted from other relations
'primary_email'
,
))
# content validation configuration #############################################
# validators are used to validate (XML, DTD, whatever) view's content
# validators availables are :
# 'dtd' : validates XML + declared DTD
# 'xml' : guarantees XML is well formed
# None : do not try to validate anything
# {'vid': validator}
VIEW_VALIDATORS
=
{}
# cubicweb test configuration ##################################################
BASE_URL
=
'http://testing.fr/cubicweb/'
DEFAULT_SOURCES
=
{
'system'
:
{
'adapter'
:
'native'
,
'db-encoding'
:
'UTF-8'
,
#'ISO-8859-1',
'db-user'
:
u
'admin'
,
...
...
@@ -40,13 +74,14 @@ DEFAULT_SOURCES = {'system': {'adapter' : 'native',
},
}
class
TestServerConfiguration
(
ServerConfiguration
):
mode
=
'test'
set_language
=
False
read_instance_schema
=
False
bootstrap_schema
=
False
init_repository
=
True
options
=
merge_options
(
ServerConfiguration
.
options
+
(
options
=
cwconfig
.
merge_options
(
ServerConfiguration
.
options
+
(
(
'anonymous-user'
,
{
'type'
:
'string'
,
'default'
:
None
,
...
...
@@ -66,7 +101,6 @@ class TestServerConfiguration(ServerConfiguration):
def
__init__
(
self
,
appid
,
log_threshold
=
logging
.
CRITICAL
+
10
):
ServerConfiguration
.
__init__
(
self
,
appid
)
self
.
global_set_option
(
'log-file'
,
None
)
self
.
init_log
(
log_threshold
,
force
=
True
)
# need this, usually triggered by cubicweb-ctl
self
.
load_cwctl_plugins
()
...
...
@@ -118,30 +152,11 @@ class TestServerConfiguration(ServerConfiguration):
sources
=
DEFAULT_SOURCES
return
sources
def
load_defaults
(
self
):
super
(
TestServerConfiguration
,
self
).
load_defaults
()
# note: don't call global set option here, OptionManager may not yet be initialized
# add anonymous user
self
.
set_option
(
'anonymous-user'
,
'anon'
)
self
.
set_option
(
'anonymous-password'
,
'anon'
)
# uncomment the line below if you want rql queries to be logged
#self.set_option('query-log-file', '/tmp/test_rql_log.' + `os.getpid()`)
self
.
set_option
(
'sender-name'
,
'cubicweb-test'
)
self
.
set_option
(
'sender-addr'
,
'cubicweb-test@logilab.fr'
)
try
:
send_to
=
'%s@logilab.fr'
%
os
.
getlogin
()
except
OSError
:
send_to
=
'%s@logilab.fr'
%
(
os
.
environ
.
get
(
'USER'
)
or
os
.
environ
.
get
(
'USERNAME'
)
or
os
.
environ
.
get
(
'LOGNAME'
))
self
.
set_option
(
'sender-addr'
,
send_to
)
self
.
set_option
(
'default-dest-addrs'
,
send_to
)
self
.
set_option
(
'base-url'
,
BASE_URL
)
class
BaseApptestConfiguration
(
TestServerConfiguration
,
TwistedConfiguration
):
repo_method
=
'inmemory'
options
=
merge_options
(
TestServerConfiguration
.
options
+
TwistedConfiguration
.
options
)
options
=
cwconfig
.
merge_options
(
TestServerConfiguration
.
options
+
TwistedConfiguration
.
options
)
cubicweb_appobject_path
=
TestServerConfiguration
.
cubicweb_appobject_path
|
TwistedConfiguration
.
cubicweb_appobject_path
cube_appobject_path
=
TestServerConfiguration
.
cube_appobject_path
|
TwistedConfiguration
.
cube_appobject_path
...
...
@@ -163,98 +178,91 @@ class ApptestConfiguration(BaseApptestConfiguration):
BaseApptestConfiguration
.
__init__
(
self
,
appid
,
log_threshold
=
log_threshold
)
self
.
init_repository
=
sourcefile
is
None
self
.
sourcefile
=
sourcefile
import
re
self
.
global_set_option
(
'
embed-allowed'
,
re
.
compile
(
'.*'
)
)
self
.
global_set_option
(
'anonymous-user'
,
'anon'
)
self
.
global_set_option
(
'
anonymous-password'
,
'anon'
)
def
load_configuration
(
self
):
super
(
ApptestConfiguration
,
self
).
load_configuration
()
self
.
global_set_option
(
'anonymous-user'
,
'anon'
)
self
.
global_set_option
(
'anonymous-password'
,
'anon'
)
class
RealDatabaseConfiguration
(
ApptestConfiguration
):
init_repository
=
False
sourcesdef
=
{
'system'
:
{
'adapter'
:
'native'
,
'db-encoding'
:
'UTF-8'
,
#'ISO-8859-1',
'db-user'
:
u
'admin'
,
'db-password'
:
'gingkow'
,
'db-name'
:
'seotest'
,
'db-driver'
:
'postgres'
,
'db-host'
:
None
,
},
'admin'
:
{
'login'
:
u
'admin'
,
'password'
:
u
'gingkow'
,
},
}
def
__init__
(
self
,
appid
,
log_threshold
=
logging
.
CRITICAL
,
sourcefile
=
None
):
ApptestConfiguration
.
__init__
(
self
,
appid
)
self
.
init_repository
=
False
# test database handling #######################################################
def
init_test_database
(
config
=
None
,
configdir
=
'data'
):
"""init a test database for a specific driver"""
from
cubicweb.dbapi
import
in_memory_cnx
config
=
config
or
TestServerConfiguration
(
configdir
)
sources
=
config
.
sources
()
driver
=
sources
[
'system'
][
'db-driver'
]
if
driver
==
'sqlite'
:
init_test_database_sqlite
(
config
)
elif
driver
==
'postgres'
:
init_test_database_postgres
(
config
)
else
:
raise
ValueError
(
'no initialization function for driver %r'
%
driver
)
config
.
_cubes
=
None
# avoid assertion error
repo
,
cnx
=
in_memory_cnx
(
config
,
unicode
(
sources
[
'admin'
][
'login'
]),
sources
[
'admin'
][
'password'
]
or
'xxx'
)
if
driver
==
'sqlite'
:
install_sqlite_patch
(
repo
.
querier
)
return
repo
,
cnx
def
sources
(
self
):
"""
By default, we run tests with the sqlite DB backend.
One may use its own configuration by just creating a
'sources' file in the test directory from wich tests are
launched.
"""
self
.
_sources
=
self
.
sourcesdef
return
self
.
_sources
def
reset_test_database
(
config
):
"""init a test database for a specific driver"""
driver
=
config
.
sources
()[
'system'
][
'db-driver'
]
if
driver
==
'sqlite'
:
reset_test_database_sqlite
(
config
)
else
:
raise
ValueError
(
'no reset function for driver %r'
%
driver
)
def
buildconfig
(
dbuser
,
dbpassword
,
dbname
,
adminuser
,
adminpassword
,
dbhost
=
None
):
"""convenience function that builds a real-db configuration class"""
sourcesdef
=
{
'system'
:
{
'adapter'
:
'native'
,
'db-encoding'
:
'UTF-8'
,
#'ISO-8859-1',
'db-user'
:
dbuser
,
'db-password'
:
dbpassword
,
'db-name'
:
dbname
,
'db-driver'
:
'postgres'
,
'db-host'
:
dbhost
,
},
'admin'
:
{
'login'
:
adminuser
,
'password'
:
adminpassword
,
},
}
return
type
(
'MyRealDBConfig'
,
(
RealDatabaseConfiguration
,),
{
'sourcesdef'
:
sourcesdef
})
### postgres test database handling ############################################
def
loadconfig
(
filename
):
"""convenience function that builds a real-db configuration class
from a file
"""
return
type
(
'MyRealDBConfig'
,
(
RealDatabaseConfiguration
,),
{
'sourcesdef'
:
read_config
(
filename
)})
class
LivetestConfiguration
(
BaseApptestConfiguration
):
init_repository
=
False
def
__init__
(
self
,
cube
=
None
,
sourcefile
=
None
,
pyro_name
=
None
,
log_threshold
=
logging
.
CRITICAL
):
TestServerConfiguration
.
__init__
(
self
,
cube
,
log_threshold
=
log_threshold
)
self
.
appid
=
pyro_name
or
cube
# don't change this, else some symlink problems may arise in some
# environment (e.g. mine (syt) ;o)
# XXX I'm afraid this test will prevent to run test from a production
# environment
self
.
_sources
=
None
# instance cube test
if
cube
is
not
None
:
self
.
apphome
=
self
.
cube_dir
(
cube
)
elif
'web'
in
os
.
getcwd
().
split
(
os
.
sep
):
# web test
self
.
apphome
=
join
(
normpath
(
join
(
dirname
(
__file__
),
'..'
)),
'web'
)
else
:
# cube test
self
.
apphome
=
abspath
(
'..'
)
self
.
sourcefile
=
sourcefile
self
.
global_set_option
(
'realm'
,
''
)
self
.
use_pyro
=
pyro_name
is
not
None
def
init_test_database_postgres
(
config
):
"""initialize a fresh sqlite databse used for testing purpose"""
if
config
.
init_repository
:
from
cubicweb.server
import
init_repository
init_repository
(
config
,
interactive
=
False
,
drop
=
True
)
def
pyro_enabled
(
self
):
if
self
.
use_pyro
:
return
True
else
:
return
False
CubicWebConfiguration
.
cls_adjust_sys_path
()
### sqlite test database handling ##############################################
def
cleanup_sqlite
(
dbfile
,
removetemplate
=
False
):
try
:
os
.
remove
(
dbfile
)
os
.
remove
(
'%s-journal'
%
dbfile
)
except
OSError
:
pass
if
removetemplate
:
try
:
os
.
remove
(
'%s-template'
%
dbfile
)
except
OSError
:
pass
def
reset_test_database_sqlite
(
config
):
import
shutil
dbfile
=
config
.
sources
()[
'system'
][
'db-name'
]
cleanup_sqlite
(
dbfile
)
template
=
'%s-template'
%
dbfile
if
exists
(
template
):
shutil
.
copy
(
template
,
dbfile
)
return
True
return
False
def
init_test_database_sqlite
(
config
):
"""initialize a fresh sqlite databse used for testing purpose"""
# remove database file if it exists
dbfile
=
config
.
sources
()[
'system'
][
'db-name'
]
if
not
reset_test_database_sqlite
(
config
):
# initialize the database
import
shutil
from
cubicweb.server
import
init_repository
init_repository
(
config
,
interactive
=
False
)
dbfile
=
config
.
sources
()[
'system'
][
'db-name'
]
shutil
.
copy
(
dbfile
,
'%s-template'
%
dbfile
)
def
install_sqlite_patch
(
querier
):
"""This patch hotfixes the following sqlite bug :
...
...
@@ -293,57 +301,3 @@ def install_sqlite_patch(querier):
return
new_execute
querier
.
__class__
.
execute
=
wrap_execute
(
querier
.
__class__
.
execute
)
querier
.
__class__
.
_devtools_sqlite_patched
=
True
def
init_test_database
(
driver
=
'sqlite'
,
configdir
=
'data'
,
config
=
None
,
vreg
=
None
):
"""init a test database for a specific driver"""
from
cubicweb.dbapi
import
in_memory_cnx
if
vreg
and
not
config
:
config
=
vreg
.
config
config
=
config
or
TestServerConfiguration
(
configdir
)
source
=
config
.
sources
()
if
driver
==
'sqlite'
:
init_test_database_sqlite
(
config
,
source
)
elif
driver
==
'postgres'
:
init_test_database_postgres
(
config
,
source
)
else
:
raise
ValueError
(
'no initialization function for driver %r'
%
driver
)
config
.
_cubes
=
None
# avoid assertion error
repo
,
cnx
=
in_memory_cnx
(
vreg
or
config
,
unicode
(
source
[
'admin'
][
'login'
]),
source
[
'admin'
][
'password'
]
or
'xxx'
)
if
driver
==
'sqlite'
:
install_sqlite_patch
(
repo
.
querier
)
return
repo
,
cnx
def
init_test_database_postgres
(
config
,
source
,
vreg
=
None
):
"""initialize a fresh sqlite databse used for testing purpose"""
if
config
.
init_repository
:
from
cubicweb.server
import
init_repository
init_repository
(
config
,
interactive
=
False
,
drop
=
True
,
vreg
=
vreg
)
def
cleanup_sqlite
(
dbfile
,
removecube
=
False
):
try
:
os
.
remove
(
dbfile
)
os
.
remove
(
'%s-journal'
%
dbfile
)
except
OSError
:
pass
if
removecube
:
try
:
os
.
remove
(
'%s-template'
%
dbfile
)
except
OSError
:
pass
def
init_test_database_sqlite
(
config
,
source
,
vreg
=
None
):
"""initialize a fresh sqlite databse used for testing purpose"""
import
shutil
# remove database file if it exists (actually I know driver == 'sqlite' :)
dbfile
=
source
[
'system'
][
'db-name'
]
cleanup_sqlite
(
dbfile
)
template
=
'%s-template'
%
dbfile
if
exists
(
template
):
shutil
.
copy
(
template
,
dbfile
)
else
:
# initialize the database
from
cubicweb.server
import
init_repository
init_repository
(
config
,
interactive
=
False
,
vreg
=
vreg
)
shutil
.
copy
(
dbfile
,
template
)
devtools/_apptest.py
deleted
100644 → 0
View file @
58c519e5
"""Hidden internals for the devtools.apptest module
:organization: Logilab
:copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.
:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
__docformat__
=
"restructuredtext en"
import
sys
,
traceback
from
logilab.common.pytest
import
pause_tracing
,
resume_tracing
import
yams.schema
from
cubicweb.dbapi
import
repo_connect
,
ConnectionProperties
,
ProgrammingError
from
cubicweb.cwvreg
import
CubicWebVRegistry
from
cubicweb.web.application
import
CubicWebPublisher
from
cubicweb.web
import
Redirect
from
cubicweb.devtools
import
ApptestConfiguration
,
init_test_database
from
cubicweb.devtools.fake
import
FakeRequest
SYSTEM_ENTITIES
=
(
'CWGroup'
,
'CWUser'
,
'CWAttribute'
,
'CWRelation'
,
'CWConstraint'
,
'CWConstraintType'
,
'CWProperty'
,
'CWEType'
,
'CWRType'
,
'State'
,
'Transition'
,
'TrInfo'
,
'RQLExpression'
,
)
SYSTEM_RELATIONS
=
(
# virtual relation
'identity'
,
# metadata
'is'
,
'is_instance_of'
,
'owned_by'
,
'created_by'
,
'specializes'
,
# workflow related
'state_of'
,
'transition_of'
,
'initial_state'
,
'allowed_transition'
,
'destination_state'
,
'in_state'
,
'wf_info_for'
,
'from_state'
,
'to_state'
,
'condition'
,
# permission
'in_group'
,
'require_group'
,
'require_permission'
,
'read_permission'
,
'update_permission'
,
'delete_permission'
,
'add_permission'
,
# eproperty
'for_user'
,
# schema definition
'relation_type'
,
'from_entity'
,
'to_entity'
,
'constrained_by'
,
'cstrtype'
,
'widget'
,
# deducted from other relations
'primary_email'
,
)
def
unprotected_entities
(
app_schema
,
strict
=
False
):
"""returned a Set of each non final entity type, excluding CWGroup, and CWUser...
"""
if
strict
:
protected_entities
=
yams
.
schema
.
BASE_TYPES
else
:
protected_entities
=
yams
.
schema
.
BASE_TYPES
.
union
(
set
(
SYSTEM_ENTITIES
))
entities
=
set
(
app_schema
.
entities
())
return
entities
-
protected_entities
def
ignore_relations
(
*
relations
):
global
SYSTEM_RELATIONS
SYSTEM_RELATIONS
+=
relations
class
TestEnvironment
(
object
):
"""TestEnvironment defines a context (e.g. a config + a given connection) in
which the tests are executed
"""
def
__init__
(
self
,
appid
,
reporter
=
None
,
verbose
=
False
,
configcls
=
ApptestConfiguration
,
requestcls
=
FakeRequest
):
config
=
configcls
(
appid
)
self
.
requestcls
=
requestcls
self
.
cnx
=
None
config
.
db_perms
=
False
source
=
config
.
sources
()[
'system'
]
if
verbose
:
print
"init test database ..."
self
.
vreg
=
vreg
=
CubicWebVRegistry
(
config
)
self
.
admlogin
=
source
[
'db-user'
]
# restore database <=> init database
self
.
restore_database
()
if
verbose
:
print
"init done"
config
.
repository
=
lambda
x
=
None
:
self
.
repo
self
.
app
=
CubicWebPublisher
(
config
,
vreg
=
vreg
)
self
.
verbose
=
verbose
schema
=
self
.
vreg
.
schema
# else we may run into problems since email address are ususally share in app tests
# XXX should not be necessary anymore
schema
.
rschema
(
'primary_email'
).
set_rproperty
(
'CWUser'
,
'EmailAddress'
,
'composite'
,
False
)
self
.
deletable_entities
=
unprotected_entities
(
schema
)
def
restore_database
(
self
):
"""called by unittests' tearDown to restore the original database
"""
try
:
pause_tracing
()
if
self
.
cnx
:
self
.
cnx
.
close
()
source
=
self
.
vreg
.
config
.
sources
()[
'system'
]
self
.
repo
,
self
.
cnx
=
init_test_database
(
driver
=
source
[
'db-driver'
],
vreg
=
self
.
vreg
)
self
.
_orig_cnx
=
self
.
cnx
resume_tracing
()
except
:
resume_tracing
()
traceback
.
print_exc
()
sys
.
exit
(
1
)
# XXX cnx decoration is usually done by the repository authentication manager,
# necessary in authentication tests
self
.
cnx
.
vreg
=
self
.
vreg
self
.
cnx
.
login
=
source
[
'db-user'
]
self
.
cnx
.
password
=
source
[
'db-password'
]
def
create_user
(
self
,
login
,
groups
=
(
'users'
,),
req
=
None
):
req
=
req
or
self
.
create_request
()
cursor
=
self
.
_orig_cnx
.
cursor
(
req
)
rset
=
cursor
.
execute
(
'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s,'
'X in_state S WHERE S name "activated"'
,
{
'login'
:
unicode
(
login
),
'passwd'
:
login
.
encode
(
'utf8'
)})
user
=
rset
.
get_entity
(
0
,
0
)
cursor
.
execute
(
'SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'
%
','
.
join
(
repr
(
g
)
for
g
in
groups
),
{
'x'
:
user
.
eid
},
'x'
)
user
.
clear_related_cache
(
'in_group'
,
'subject'
)
self
.
_orig_cnx
.
commit
()
return
user
def
login
(
self
,
login
,
password
=
None
):
if
login
==
self
.
admlogin
:
self
.
restore_connection
()
else
:
self
.
cnx
=
repo_connect
(
self
.
repo
,
unicode
(
login
),
password
or
str
(
login
),
ConnectionProperties
(
'inmemory'
))
if
login
==
self
.
vreg
.
config
.
anonymous_user
()[
0
]:
self
.
cnx
.
anonymous_connection
=
True
return
self
.
cnx
def
restore_connection
(
self
):
if
not
self
.
cnx
is
self
.
_orig_cnx
:
try
:
self
.
cnx
.
close
()
except
ProgrammingError
:
pass
# already closed
self
.
cnx
=
self
.
_orig_cnx
############################################################################
def
execute
(
self
,
rql
,
args
=
None
,
eidkey
=
None
,
req
=
None
):
"""executes <rql>, builds a resultset, and returns a couple (rset, req)
where req is a FakeRequest
"""
req
=
req
or
self
.
create_request
(
rql
=
rql
)
return
self
.
cnx
.
cursor
(
req
).
execute
(
unicode
(
rql
),
args
,
eidkey
)
def
create_request
(
self
,
rql
=
None
,
**
kwargs
):
"""executes <rql>, builds a resultset, and returns a
couple (rset, req) where req is a FakeRequest
"""
if
rql
:
kwargs
[
'rql'
]
=
rql
req
=
self
.
requestcls
(
self
.
vreg
,
form
=
kwargs
)
req
.
set_connection
(
self
.
cnx
)
return
req
def
get_rset_and_req
(
self
,
rql
,
optional_args
=
None
,
args
=
None
,
eidkey
=
None
):
"""executes <rql>, builds a resultset, and returns a
couple (rset, req) where req is a FakeRequest
"""
return
(
self
.
execute
(
rql
,
args
,
eidkey
),
self
.
create_request
(
rql
=
rql
,
**
optional_args
or
{}))
class
ExistingTestEnvironment
(
TestEnvironment
):
def
__init__
(
self
,
appid
,
sourcefile
,
verbose
=
False
):
config
=
ApptestConfiguration
(
appid
,
sourcefile
=
sourcefile
)
if
verbose
:
print
"init test database ..."
source
=
config
.
sources
()[
'system'
]
self
.
vreg
=
CubicWebVRegistry
(
config
)
self
.
cnx
=
init_test_database
(
driver
=
source
[
'db-driver'
],
vreg
=
self
.
vreg
)[
1
]
if
verbose
:
print
"init done"
self
.
app
=
CubicWebPublisher
(
config
,
vreg
=
self
.
vreg
)
self
.
verbose
=
verbose
# this is done when the publisher is opening a connection
self
.
cnx
.
vreg
=
self
.
vreg
def
setup
(
self
,
config
=
None
):
"""config is passed by TestSuite but is ignored in this environment"""
cursor
=
self
.
cnx
.
cursor
()
self
.
last_eid
=
cursor
.
execute
(
'Any X WHERE X creation_date D ORDERBY D DESC LIMIT 1'
).
rows
[
0
][
0
]
def
cleanup
(
self
):
"""cancel inserted elements during tests"""
cursor
=
self
.
cnx
.
cursor
()
cursor
.
execute
(
'DELETE Any X WHERE X eid > %(x)s'
,
{
'x'
:
self
.
last_eid
},
eid_key
=
'x'
)
print
"cleaning done"
self
.
cnx
.
commit
()
devtools/apptest.py
deleted
100644 → 0
View file @
58c519e5
"""This module provides misc utilities to test instances
:organization: Logilab