1 from six import text_type

     2 

     3 from cubicweb.pyramid.test import PyramidCWTest

     4 from cubicweb.pyramid import tools

     5 

     6 

     7 def set_language(request):

     8     lang = request.POST.get('lang', None)

     9     cnx = request.cw_cnx

    10     if lang is None:

    11         cnx.execute('DELETE CWProperty X WHERE X for_user U, U eid %(u)s',

    12                     {'u': cnx.user.eid})

    13     else:

    14         cnx.user.set_property(u'ui.language', text_type(lang))

    15     cnx.commit()

    16 

    17     request.response.text = text_type(cnx.user.properties.get('ui.language', ''))

    18     return request.response

    19 

    20 

    21 def add_remove_group(request):

    22     add_remove = request.POST['add_remove']

    23     cnx = request.cw_cnx

    24     if add_remove == 'add':

    25         cnx.execute('SET U in_group G WHERE G name "users", U eid %(u)s',

    26                     {'u': cnx.user.eid})

    27     else:

    28         cnx.execute('DELETE U in_group G WHERE G name "users", U eid %(u)s',

    29                     {'u': cnx.user.eid})

    30     cnx.commit()

    31 

    32     request.response.text = text_type(','.join(sorted(cnx.user.groups)))

    33     return request.response

    34 

    35 

    36 class SessionSyncHoooksTC(PyramidCWTest):

    37 

    38     def includeme(self, config):

    39         for view in (set_language, add_remove_group):

    40             config.add_route(view.__name__, '/' + view.__name__)

    41             config.add_view(view, route_name=view.__name__)

    42 

    43     def setUp(self):

    44         super(SessionSyncHoooksTC, self).setUp()

    45         with self.admin_access.repo_cnx() as cnx:

    46             self.admin_eid = cnx.user.eid

    47 

    48     def test_sync_props(self):

    49         # initialize a pyramid session using admin credentials

    50         res = self.webapp.post('/login', {

    51             '__login': self.admlogin, '__password': self.admpassword})

    52         self.assertEqual(res.status_int, 303)

    53         # new property

    54         res = self.webapp.post('/set_language', {'lang': 'fr'})

    55         self.assertEqual(res.text, 'fr')

    56         # updated property

    57         res = self.webapp.post('/set_language', {'lang': 'en'})

    58         self.assertEqual(res.text, 'en')

    59         # removed property

    60         res = self.webapp.post('/set_language')

    61         self.assertEqual(res.text, '')

    62 

    63     def test_sync_groups(self):

    64         # initialize a pyramid session using admin credentials

    65         res = self.webapp.post('/login', {

    66             '__login': self.admlogin, '__password': self.admpassword})

    67         self.assertEqual(res.status_int, 303)

    68         # XXX how to get pyramid request using this session?

    69         res = self.webapp.post('/add_remove_group', {'add_remove': 'add'})

    70         self.assertEqual(res.text, 'managers,users')

    71         res = self.webapp.post('/add_remove_group', {'add_remove': 'remove'})

    72         self.assertEqual(res.text, 'managers')

    73 

    74 

    75 if __name__ == '__main__':

    76     from unittest import main

    77     main()