Drop python2 support
This mostly consists in removing the dependency on "six" and updating
the code to use only Python3 idioms.
Notice that we previously used TemporaryDirectory from
cubicweb.devtools.testlib for compatibility with Python2. We now
directly import it from tempfile.
# copyright 2003-2016 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
#
# This file is part of CubicWeb.
#
# CubicWeb is free software: you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# CubicWeb is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with CubicWeb. If not, see <http://www.gnu.org/licenses/>.
"""unit tests for cubicweb.web.application"""
import base64
import http.client
from http.cookies import SimpleCookie
from logilab.common.testlib import TestCase, unittest_main
from logilab.common.decorators import clear_cache
from cubicweb import view
from cubicweb.devtools.testlib import CubicWebTC, real_error_handling
from cubicweb.devtools.fake import FakeRequest
from cubicweb.web import LogOut, Redirect, INTERNAL_FIELD_VALUE
from cubicweb.web.views.basecontrollers import ViewController
from cubicweb.web.application import anonymized_request
class FakeMapping:
"""emulates a mapping module"""
def __init__(self):
self.ENTITIES_MAP = {}
self.ATTRIBUTES_MAP = {}
self.RELATIONS_MAP = {}
class MockCursor:
def __init__(self):
self.executed = []
def execute(self, rql, args=None, build_descr=False):
args = args or {}
self.executed.append(rql % args)
class FakeController(ViewController):
def __init__(self, form=None):
self._cw = FakeRequest()
self._cw.form = form or {}
self._cursor = MockCursor()
self._cw.execute = self._cursor.execute
def new_cursor(self):
self._cursor = MockCursor()
self._cw.execute = self._cursor.execute
def set_form(self, form):
self._cw.form = form
class RequestBaseTC(TestCase):
def setUp(self):
self._cw = FakeRequest()
def test_list_arg(self):
"""tests the list_arg() function"""
list_arg = self._cw.list_form_param
self.assertEqual(list_arg('arg3', {}), [])
d = {'arg1': "value1",
'arg2': ('foo', INTERNAL_FIELD_VALUE,),
'arg3': ['bar']}
self.assertEqual(list_arg('arg1', d, True), ['value1'])
self.assertEqual(d, {'arg2': ('foo', INTERNAL_FIELD_VALUE), 'arg3': ['bar']})
self.assertEqual(list_arg('arg2', d, True), ['foo'])
self.assertEqual({'arg3': ['bar']}, d)
self.assertEqual(list_arg('arg3', d), ['bar'])
self.assertEqual({'arg3': ['bar']}, d)
def test_from_controller(self):
self._cw.vreg['controllers'] = {'view': 1, 'login': 1}
self.assertEqual(self._cw.from_controller(), 'view')
req = FakeRequest(url='project?vid=list')
req.vreg['controllers'] = {'view': 1, 'login': 1}
# this assertion is just to make sure that relative_path can be
# correctly computed as it is used in from_controller()
self.assertEqual(req.relative_path(False), 'project')
self.assertEqual(req.from_controller(), 'view')
# test on a valid non-view controller
req = FakeRequest(url='login?x=1&y=2')
req.vreg['controllers'] = {'view': 1, 'login': 1}
self.assertEqual(req.relative_path(False), 'login')
self.assertEqual(req.from_controller(), 'login')
class UtilsTC(TestCase):
"""test suite for misc application utilities"""
def setUp(self):
self.ctrl = FakeController()
def test_execute_linkto(self):
"""tests the execute_linkto() function"""
self.assertEqual(self.ctrl.execute_linkto(), None)
self.assertEqual(self.ctrl._cursor.executed,
[])
self.ctrl.set_form({'__linkto': 'works_for:12_13_14:object',
'eid': 8})
self.ctrl.execute_linkto()
self.assertEqual(self.ctrl._cursor.executed,
['SET Y works_for X WHERE X eid 8, Y eid %s' % i
for i in (12, 13, 14)])
self.ctrl.new_cursor()
self.ctrl.set_form({'__linkto': 'works_for:12_13_14:subject',
'eid': 8})
self.ctrl.execute_linkto()
self.assertEqual(self.ctrl._cursor.executed,
['SET X works_for Y WHERE X eid 8, Y eid %s' % i
for i in (12, 13, 14)])
self.ctrl.new_cursor()
self.ctrl._cw.form = {'__linkto': 'works_for:12_13_14:object'}
self.ctrl.execute_linkto(eid=8)
self.assertEqual(self.ctrl._cursor.executed,
['SET Y works_for X WHERE X eid 8, Y eid %s' % i
for i in (12, 13, 14)])
self.ctrl.new_cursor()
self.ctrl.set_form({'__linkto': 'works_for:12_13_14:subject'})
self.ctrl.execute_linkto(eid=8)
self.assertEqual(self.ctrl._cursor.executed,
['SET X works_for Y WHERE X eid 8, Y eid %s' % i
for i in (12, 13, 14)])
class ApplicationTC(CubicWebTC):
@classmethod
def setUpClass(cls):
super(ApplicationTC, cls).setUpClass()
cls.config.global_set_option('allow-email-login', True)
def test_cnx_user_groups_sync(self):
with self.admin_access.client_cnx() as cnx:
user = cnx.user
self.assertEqual(user.groups, set(('managers',)))
cnx.execute('SET X in_group G WHERE X eid %s, G name "guests"' % user.eid)
user = cnx.user
self.assertEqual(user.groups, set(('managers',)))
cnx.commit()
user = cnx.user
self.assertEqual(user.groups, set(('managers', 'guests')))
# cleanup
cnx.execute('DELETE X in_group G WHERE X eid %s, G name "guests"' % user.eid)
cnx.commit()
def test_publish_validation_error(self):
with self.admin_access.web_request() as req:
user = req.user
eid = str(user.eid)
req.form = {
'eid': eid,
'__type:' + eid: 'CWUser',
'_cw_entity_fields:' + eid: 'login-subject',
'login-subject:' + eid: '', # ERROR: no login specified
# just a sample, missing some necessary information for real life
'__errorurl': 'view?vid=edition...'
}
path, params = self.expect_redirect_handle_request(req, 'edit')
forminfo = req.session.data['view?vid=edition...']
eidmap = forminfo['eidmap']
self.assertEqual(eidmap, {})
values = forminfo['values']
self.assertEqual(values['login-subject:' + eid], '')
self.assertEqual(values['eid'], eid)
error = forminfo['error']
self.assertEqual(error.entity, user.eid)
self.assertEqual(error.errors['login-subject'], 'required field')
def test_validation_error_dont_loose_subentity_data_ctrl(self):
"""test creation of two linked entities
error occurs on the web controller
"""
with self.admin_access.web_request() as req:
# set Y before X to ensure both entities are edited, not only X
req.form = {'eid': ['Y', 'X'], '__maineid': 'X',
'__type:X': 'CWUser', '_cw_entity_fields:X': 'login-subject',
# missing required field
'login-subject:X': u'',
# but email address is set
'__type:Y': 'EmailAddress', '_cw_entity_fields:Y': 'address-subject',
'address-subject:Y': u'bougloup@logilab.fr',
'use_email-object:Y': 'X',
# necessary to get validation error handling
'__errorurl': 'view?vid=edition...',
}
path, params = self.expect_redirect_handle_request(req, 'edit')
forminfo = req.session.data['view?vid=edition...']
self.assertEqual(set(forminfo['eidmap']), set('XY'))
self.assertEqual(forminfo['eidmap']['X'], None)
self.assertIsInstance(forminfo['eidmap']['Y'], int)
self.assertEqual(forminfo['error'].entity, 'X')
self.assertEqual(forminfo['error'].errors,
{'login-subject': 'required field'})
self.assertEqual(forminfo['values'], req.form)
def test_handle_request_with_lang_fromurl(self):
"""No language negociation, get language from URL."""
self.config.global_set_option('language-mode', 'url-prefix')
req = self.init_authentication('http')
self.assertEqual(req.url(), 'http://testing.fr/cubicweb/login')
self.assertEqual(req.lang, 'en')
self.app.handle_request(req)
newreq = self.requestcls(req.vreg, url='fr/toto')
self.assertEqual(newreq.lang, 'en')
self.assertEqual(newreq.url(), 'http://testing.fr/cubicweb/fr/toto')
self.app.handle_request(newreq)
self.assertEqual(newreq.lang, 'fr')
self.assertEqual(newreq.url(), 'http://testing.fr/cubicweb/fr/toto')
# unknown language
newreq = self.requestcls(req.vreg, url='unknown-lang/cwuser')
result = self.app.handle_request(newreq)
self.assertEqual(newreq.lang, 'en')
self.assertEqual(newreq.url(), 'http://testing.fr/cubicweb/unknown-lang/cwuser')
self.assertIn('this resource does not exist',
result.decode('ascii', errors='ignore'))
# no prefix
newreq = self.requestcls(req.vreg, url='cwuser')
result = self.app.handle_request(newreq)
self.assertEqual(newreq.lang, 'en')
self.assertEqual(newreq.url(), 'http://testing.fr/cubicweb/cwuser')
self.assertNotIn('this resource does not exist',
result.decode('ascii', errors='ignore'))
def test_handle_request_with_lang_negotiated(self):
"""Language negociated, normal case."""
self.config.global_set_option('language-mode', 'http-negotiation')
orig_translations = self.config.translations.copy()
self.config.translations = {
'fr': (str, lambda x, y: str(y)),
'en': (str, lambda x, y: str(y))}
try:
headers = {'Accept-Language': 'fr'}
with self.admin_access.web_request(headers=headers) as req:
self.app.handle_request(req)
self.assertEqual(req.lang, 'fr')
finally:
self.config.translations = orig_translations
def test_handle_request_with_lang_negotiated_prefix_in_url(self):
"""Language negociated, unexpected language prefix in URL."""
self.config.global_set_option('language-mode', 'http-negotiation')
with self.admin_access.web_request(url='fr/toto') as req:
result = self.app.handle_request(req)
self.assertIn('this resource does not exist', # NotFound.
result.decode('ascii', errors='ignore'))
def test_handle_request_no_lang_negotiation_fixed_language(self):
"""No language negociation, "ui.language" fixed."""
self.config.global_set_option('language-mode', '')
vreg = self.app.vreg
self.assertEqual(vreg.property_value('ui.language'), 'en')
props = []
try:
with self.admin_access.cnx() as cnx:
props.append(cnx.create_entity('CWProperty', value=u'de',
pkey=u'ui.language').eid)
cnx.commit()
self.assertEqual(vreg.property_value('ui.language'), 'de')
headers = {'Accept-Language': 'fr'} # should not have any effect.
with self.admin_access.web_request(headers=headers) as req:
self.app.handle_request(req)
# user has no "ui.language" property, getting site's default.
self.assertEqual(req.lang, 'de')
# XXX The following should work, but nasty handling of session and
# request user make it fail...
# with self.admin_access.cnx() as cnx:
# props.append(cnx.create_entity('CWProperty', value=u'es',
# pkey=u'ui.language',
# for_user=cnx.user).eid)
# cnx.commit()
# with self.admin_access.web_request(headers=headers) as req:
# result = self.app.handle_request(req)
# self.assertEqual(req.lang, 'es')
finally:
with self.admin_access.cnx() as cnx:
for peid in props:
cnx.entity_from_eid(peid).cw_delete()
cnx.commit()
def test_validation_error_dont_loose_subentity_data_repo(self):
"""test creation of two linked entities
error occurs on the repository
"""
with self.admin_access.web_request() as req:
# set Y before X to ensure both entities are edited, not only X
req.form = {
'eid': ['Y', 'X'], '__maineid': 'X',
'__type:X': 'CWUser', '_cw_entity_fields:X': 'login-subject,upassword-subject',
# already existent user
'login-subject:X': u'admin',
'upassword-subject:X': u'admin', 'upassword-subject-confirm:X': u'admin',
'__type:Y': 'EmailAddress', '_cw_entity_fields:Y': 'address-subject',
'address-subject:Y': u'bougloup@logilab.fr',
'use_email-object:Y': 'X',
# necessary to get validation error handling
'__errorurl': 'view?vid=edition...',
}
path, params = self.expect_redirect_handle_request(req, 'edit')
forminfo = req.session.data['view?vid=edition...']
self.assertEqual(set(forminfo['eidmap']), set('XY'))
self.assertIsInstance(forminfo['eidmap']['X'], int)
self.assertIsInstance(forminfo['eidmap']['Y'], int)
self.assertEqual(forminfo['error'].entity, forminfo['eidmap']['X'])
expected_errors = {
'': u'some relations violate a unicity constraint',
'login': u'login is part of violated unicity constraint',
}
self.assertEqual(forminfo['error'].errors, expected_errors)
self.assertEqual(forminfo['values'], req.form)
def _edit_parent(self, dir_eid, parent_eid, role='subject',
etype='Directory', **kwargs):
parent_eid = parent_eid or '__cubicweb_internal_field__'
with self.admin_access.web_request() as req:
req.form = {
'eid': str(dir_eid),
'__maineid': str(dir_eid),
'__type:%s' % dir_eid: etype,
'parent-%s:%s' % (role, dir_eid): parent_eid,
}
req.form.update(kwargs)
req.form['_cw_entity_fields:%s' % dir_eid] = ','.join(
['parent-%s' % role]
+ [key.split(':')[0]
for key in kwargs.keys()
if not key.startswith('_')])
self.expect_redirect_handle_request(req)
def _edit_in_version(self, ticket_eid, version_eid, **kwargs):
version_eid = version_eid or '__cubicweb_internal_field__'
with self.admin_access.web_request() as req:
req.form = {
'eid': str(ticket_eid),
'__maineid': str(ticket_eid),
'__type:%s' % ticket_eid: 'Ticket',
'in_version-subject:%s' % ticket_eid: version_eid,
}
req.form.update(kwargs)
req.form['_cw_entity_fields:%s' % ticket_eid] = ','.join(
['in_version-subject']
+ [key.split(':')[0]
for key in kwargs.keys()
if not key.startswith('_')])
self.expect_redirect_handle_request(req)
def test_create_and_link_directories(self):
with self.admin_access.web_request() as req:
req.form = {
'eid': (u'A', u'B'),
'__maineid': u'A',
'__type:A': 'Directory',
'__type:B': 'Directory',
'parent-subject:B': u'A',
'name-subject:A': u'topd',
'name-subject:B': u'subd',
'_cw_entity_fields:A': 'name-subject',
'_cw_entity_fields:B': 'parent-subject,name-subject',
}
self.expect_redirect_handle_request(req)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', name=u'topd'))
self.assertTrue(cnx.find('Directory', name=u'subd'))
self.assertEqual(1, cnx.execute(
'Directory SUBD WHERE SUBD parent TOPD,'
' SUBD name "subd", TOPD name "topd"').rowcount)
def test_create_subentity(self):
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
cnx.commit()
with self.admin_access.web_request() as req:
req.form = {
'eid': (str(topd.eid), u'B'),
'__maineid': str(topd.eid),
'__type:%s' % topd.eid: 'Directory',
'__type:B': 'Directory',
'parent-object:%s' % topd.eid: u'B',
'name-subject:B': u'subd',
'_cw_entity_fields:%s' % topd.eid: 'parent-object',
'_cw_entity_fields:B': 'name-subject',
}
self.expect_redirect_handle_request(req)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', name=u'topd'))
self.assertTrue(cnx.find('Directory', name=u'subd'))
self.assertEqual(1, cnx.execute(
'Directory SUBD WHERE SUBD parent TOPD,'
' SUBD name "subd", TOPD name "topd"').rowcount)
def test_subject_subentity_removal(self):
"""Editcontroller: detaching a composite relation removes the subentity
(edit from the subject side)
"""
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
sub1 = cnx.create_entity('Directory', name=u'sub1', parent=topd)
sub2 = cnx.create_entity('Directory', name=u'sub2', parent=topd)
cnx.commit()
attrs = {'name-subject:%s' % sub1.eid: ''}
self._edit_parent(sub1.eid, parent_eid=None, **attrs)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=topd.eid))
self.assertFalse(cnx.find('Directory', eid=sub1.eid))
self.assertTrue(cnx.find('Directory', eid=sub2.eid))
def test_object_subentity_removal(self):
"""Editcontroller: detaching a composite relation removes the subentity
(edit from the object side)
"""
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
sub1 = cnx.create_entity('Directory', name=u'sub1', parent=topd)
sub2 = cnx.create_entity('Directory', name=u'sub2', parent=topd)
cnx.commit()
self._edit_parent(topd.eid, parent_eid=sub1.eid, role='object')
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=topd.eid))
self.assertTrue(cnx.find('Directory', eid=sub1.eid))
self.assertFalse(cnx.find('Directory', eid=sub2.eid))
def test_reparent_subentity(self):
"Editcontroller: re-parenting a subentity does not remove it"
with self.admin_access.repo_cnx() as cnx:
top1 = cnx.create_entity('Directory', name=u'top1')
top2 = cnx.create_entity('Directory', name=u'top2')
subd = cnx.create_entity('Directory', name=u'subd', parent=top1)
cnx.commit()
self._edit_parent(subd.eid, parent_eid=top2.eid)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=top1.eid))
self.assertTrue(cnx.find('Directory', eid=top2.eid))
self.assertTrue(cnx.find('Directory', eid=subd.eid))
self.assertEqual(
cnx.find('Directory', eid=subd.eid).one().parent[0], top2)
def test_reparent_subentity_inlined(self):
"""Editcontroller: re-parenting a subentity does not remove it
(inlined case)"""
with self.admin_access.repo_cnx() as cnx:
version1 = cnx.create_entity('Version', name=u'version1')
version2 = cnx.create_entity('Version', name=u'version2')
ticket = cnx.create_entity('Ticket', title=u'ticket',
in_version=version1)
cnx.commit()
self._edit_in_version(ticket.eid, version_eid=version2.eid)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Version', eid=version1.eid))
self.assertTrue(cnx.find('Version', eid=version2.eid))
self.assertTrue(cnx.find('Ticket', eid=ticket.eid))
self.assertEqual(
cnx.find('Ticket', eid=ticket.eid).one().in_version[0], version2)
def test_subject_mixed_composite_subentity_removal_1(self):
"""Editcontroller: detaching several subentities respects each rdef's
compositeness - Remove non composite
"""
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
fs = cnx.create_entity('Filesystem', name=u'/tmp')
subd = cnx.create_entity('Directory', name=u'subd',
parent=(topd, fs))
cnx.commit()
self._edit_parent(subd.eid, parent_eid=topd.eid)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=topd.eid))
self.assertTrue(cnx.find('Directory', eid=subd.eid))
self.assertTrue(cnx.find('Filesystem', eid=fs.eid))
self.assertEqual(cnx.find('Directory', eid=subd.eid).one().parent,
(topd,))
def test_subject_mixed_composite_subentity_removal_2(self):
"""Editcontroller: detaching several subentities respects each rdef's
compositeness - Remove composite
"""
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
fs = cnx.create_entity('Filesystem', name=u'/tmp')
subd = cnx.create_entity('Directory', name=u'subd',
parent=(topd, fs))
cnx.commit()
self._edit_parent(subd.eid, parent_eid=fs.eid)
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=topd.eid))
self.assertFalse(cnx.find('Directory', eid=subd.eid))
self.assertTrue(cnx.find('Filesystem', eid=fs.eid))
def test_object_mixed_composite_subentity_removal_1(self):
"""Editcontroller: detaching several subentities respects each rdef's
compositeness - Remove non composite
"""
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
fs = cnx.create_entity('Filesystem', name=u'/tmp')
subd = cnx.create_entity('Directory', name=u'subd',
parent=(topd, fs))
cnx.commit()
self._edit_parent(fs.eid, parent_eid=None, role='object',
etype='Filesystem')
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=topd.eid))
self.assertTrue(cnx.find('Directory', eid=subd.eid))
self.assertTrue(cnx.find('Filesystem', eid=fs.eid))
self.assertEqual(cnx.find('Directory', eid=subd.eid).one().parent,
(topd,))
def test_object_mixed_composite_subentity_removal_2(self):
"""Editcontroller: detaching several subentities respects each rdef's
compositeness - Remove composite
"""
with self.admin_access.repo_cnx() as cnx:
topd = cnx.create_entity('Directory', name=u'topd')
fs = cnx.create_entity('Filesystem', name=u'/tmp')
subd = cnx.create_entity('Directory', name=u'subd',
parent=(topd, fs))
cnx.commit()
self._edit_parent(topd.eid, parent_eid=None, role='object')
with self.admin_access.repo_cnx() as cnx:
self.assertTrue(cnx.find('Directory', eid=topd.eid))
self.assertFalse(cnx.find('Directory', eid=subd.eid))
self.assertTrue(cnx.find('Filesystem', eid=fs.eid))
def test_delete_mandatory_composite(self):
with self.admin_access.repo_cnx() as cnx:
perm = cnx.create_entity('DirectoryPermission')
mydir = cnx.create_entity('Directory', name=u'dir',
has_permission=perm)
cnx.commit()
with self.admin_access.web_request() as req:
dir_eid = str(mydir.eid)
perm_eid = str(perm.eid)
req.form = {
'eid': [dir_eid, perm_eid],
'__maineid': dir_eid,
'__type:%s' % dir_eid: 'Directory',
'__type:%s' % perm_eid: 'DirectoryPermission',
'_cw_entity_fields:%s' % dir_eid: '',
'_cw_entity_fields:%s' % perm_eid: 'has_permission-object',
'has_permission-object:%s' % perm_eid: '',
}
path, _params = self.expect_redirect_handle_request(req, 'edit')
self.assertTrue(req.find('Directory', eid=mydir.eid))
self.assertFalse(req.find('DirectoryPermission', eid=perm.eid))
def test_ajax_view_raise_arbitrary_error(self):
class ErrorAjaxView(view.View):
__regid__ = 'test.ajax.error'
def call(self):
raise Exception('whatever')
with self.temporary_appobjects(ErrorAjaxView):
with real_error_handling(self.app) as app:
with self.admin_access.web_request(vid='test.ajax.error', url='') as req:
req.ajax_request = True
app.handle_request(req)
self.assertEqual(http.client.INTERNAL_SERVER_ERROR,
req.status_out)
def _test_cleaned(self, kwargs, injected, cleaned):
with self.admin_access.web_request(**kwargs) as req:
page = self.app_handle_request(req)
self.assertNotIn(injected.encode('ascii'), page)
self.assertIn(cleaned.encode('ascii'), page)
def test_nonregr_script_kiddies(self):
"""test against current script injection"""
injected = '<i>toto</i>'
cleaned = 'toto'
for kwargs in ({'vid': injected}, {'vtitle': injected}):
with self.subTest(**kwargs):
self._test_cleaned(kwargs, injected, cleaned)
def test_site_wide_eproperties_sync(self):
# XXX work in all-in-one configuration but not in twisted for instance
# in which case we need a kindof repo -> http server notification
# protocol
vreg = self.app.vreg
# default value
self.assertEqual(vreg.property_value('ui.language'), 'en')
with self.admin_access.client_cnx() as cnx:
cnx.execute('INSERT CWProperty X: X value "fr", X pkey "ui.language"')
self.assertEqual(vreg.property_value('ui.language'), 'en')
cnx.commit()
self.assertEqual(vreg.property_value('ui.language'), 'fr')
cnx.execute('SET X value "de" WHERE X pkey "ui.language"')
self.assertEqual(vreg.property_value('ui.language'), 'fr')
cnx.commit()
self.assertEqual(vreg.property_value('ui.language'), 'de')
cnx.execute('DELETE CWProperty X WHERE X pkey "ui.language"')
self.assertEqual(vreg.property_value('ui.language'), 'de')
cnx.commit()
self.assertEqual(vreg.property_value('ui.language'), 'en')
# authentication tests ####################################################
def test_http_auth_no_anon(self):
req = self.init_authentication('http')
self.assertAuthFailure(req)
self.app.handle_request(req)
self.assertEqual(401, req.status_out)
clear_cache(req, 'get_authorization')
authstr = base64.encodestring(('%s:%s' % (self.admlogin, self.admpassword)).encode('ascii'))
req.set_request_header('Authorization', 'basic %s' % authstr.decode('ascii'))
self.assertAuthSuccess(req)
req._url = 'logout'
self.assertRaises(LogOut, self.app_handle_request, req)
self.assertEqual(len(self.open_sessions), 0)
def test_cookie_auth_no_anon(self):
req = self.init_authentication('cookie')
self.assertAuthFailure(req)
try:
form = self.app.handle_request(req)
except Redirect:
self.fail('anonymous user should get login form')
clear_cache(req, 'get_authorization')
self.assertIn(b'__login', form)
self.assertIn(b'__password', form)
self.assertFalse(req.cnx) # Mock cnx are False
req.form['__login'] = self.admlogin
req.form['__password'] = self.admpassword
self.assertAuthSuccess(req)
req._url = 'logout'
self.assertRaises(LogOut, self.app_handle_request, req)
self.assertEqual(len(self.open_sessions), 0)
def test_login_by_email(self):
with self.admin_access.client_cnx() as cnx:
login = cnx.user.login
address = login + u'@localhost'
cnx.execute('INSERT EmailAddress X: X address %(address)s, U primary_email X '
'WHERE U login %(login)s', {'address': address, 'login': login})
cnx.commit()
req = self.init_authentication('cookie')
self.set_option('allow-email-login', True)
req.form['__login'] = address
req.form['__password'] = self.admpassword
self.assertAuthSuccess(req)
req._url = 'logout'
self.assertRaises(LogOut, self.app_handle_request, req)
self.assertEqual(len(self.open_sessions), 0)
def _reset_cookie(self, req):
# preparing the suite of the test
# set session id in cookie
cookie = SimpleCookie()
sessioncookie = self.app.session_handler.session_cookie(req)
cookie[sessioncookie] = req.session.sessionid
req.set_request_header('Cookie', cookie[sessioncookie].OutputString(),
raw=True)
clear_cache(req, 'get_authorization')
def _test_auth_anon(self, req):
asession = self.app.get_session(req)
# important otherwise _reset_cookie will not use the right session
cnx = asession.new_cnx()
with cnx:
req.set_cnx(cnx)
self.assertEqual(len(self.open_sessions), 1)
self.assertEqual(asession.user.login, 'anon')
self.assertTrue(asession.anonymous_session)
self._reset_cookie(req)
def _test_anon_auth_fail(self, req):
self.assertEqual(1, len(self.open_sessions))
session = self.app.get_session(req)
cnx = session.new_cnx()
with cnx:
# important otherwise _reset_cookie will not use the right session
req.set_cnx(cnx)
self.assertEqual(req.message, 'authentication failure')
self.assertEqual(req.session.anonymous_session, True)
self.assertEqual(1, len(self.open_sessions))
self._reset_cookie(req)
def test_http_auth_anon_allowed(self):
req = self.init_authentication('http', 'anon')
self._test_auth_anon(req)
authstr = base64.encodestring(b'toto:pouet')
req.set_request_header('Authorization', 'basic %s' % authstr.decode('ascii'))
self._test_anon_auth_fail(req)
authstr = base64.encodestring(('%s:%s' % (self.admlogin, self.admpassword)).encode('ascii'))
req.set_request_header('Authorization', 'basic %s' % authstr.decode('ascii'))
self.assertAuthSuccess(req)
req._url = 'logout'
self.assertRaises(LogOut, self.app_handle_request, req)
self.assertEqual(len(self.open_sessions), 0)
def test_cookie_auth_anon_allowed(self):
req = self.init_authentication('cookie', 'anon')
self._test_auth_anon(req)
req.form['__login'] = 'toto'
req.form['__password'] = 'pouet'
self._test_anon_auth_fail(req)
req.form['__login'] = self.admlogin
req.form['__password'] = self.admpassword
self.assertAuthSuccess(req)
req._url = 'logout'
self.assertRaises(LogOut, self.app_handle_request, req)
self.assertEqual(0, len(self.open_sessions))
def test_anonymized_request(self):
with self.admin_access.web_request() as req:
self.assertEqual(self.admlogin, req.session.user.login)
# admin should see anon + admin
self.assertEqual(2, len(list(req.find('CWUser'))))
with anonymized_request(req):
self.assertEqual('anon', req.session.user.login)
# anon should only see anon user
self.assertEqual(1, len(list(req.find('CWUser'))))
self.assertEqual(self.admlogin, req.session.user.login)
self.assertEqual(2, len(list(req.find('CWUser'))))
def test_non_regr_optional_first_var(self):
with self.admin_access.web_request() as req:
# expect a rset with None in [0][0]
req.form['rql'] = 'rql:Any OV1, X WHERE X custom_workflow OV1?'
self.app_handle_request(req)
if __name__ == '__main__':
unittest_main()