# 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

from six import text_type

from six.moves import http_client

from six.moves.http_cookies import SimpleCookie

from logilab.common.testlib import TestCase, unittest_main

from logilab.common.decorators import clear_cache, classproperty

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

from cubicweb import repoapi

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):

    @classproperty

    def config(cls):

        try:

            return cls.__dict__['_config']

        except KeyError:

            config = super(ApplicationTC, cls).config

            config.global_set_option('allow-email-login', True)

            return config

    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 = text_type(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_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'])

            self.assertEqual(forminfo['error'].errors,

                             {'login-subject': u'the value "admin" is already used, use another one'})

            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': unicode(dir_eid),

                '__maineid': unicode(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': unicode(ticket_eid),

                '__maineid': unicode(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': (unicode(topd.eid), u'B'),

                '__maineid': unicode(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')