# 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

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 = 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_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': (text_type, lambda x, y: text_type(y)),

            'en': (text_type, lambda x, y: text_type(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': text_type(dir_eid),

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