# 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/>.

"""cubicweb.web.views.basecontrollers unit tests"""

import time

from six import text_type

from six.moves.urllib.parse import urlsplit, urlunsplit, urljoin, parse_qs

import lxml

from logilab.common.testlib import unittest_main

from cubicweb import Binary, NoSelectableObject, ValidationError, transaction as tx

from cubicweb.schema import RRQLExpression

from cubicweb.predicates import is_instance

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.devtools.webtest import CubicWebTestTC

from cubicweb.devtools.httptest import CubicWebServerTC

from cubicweb.utils import json_dumps

from cubicweb.uilib import rql_for_eid

from cubicweb.web import Redirect, RemoteCallFailed, http_headers, formfields as ff

from cubicweb.web.views.autoform import get_pending_inserts, get_pending_deletes

from cubicweb.web.views.ajaxcontroller import ajaxfunc, AjaxFunction

from cubicweb.server.session import Connection

from cubicweb.server.hook import Hook, Operation

class ViewControllerTC(CubicWebTestTC):

    def test_view_ctrl_with_valid_cache_headers(self):

        now = time.time()

        resp = self.webapp.get('/manage')

        self.assertEqual(resp.etag, 'manage/guests')

        self.assertEqual(resp.status_code, 200)

        self.assertGreaterEqual(http_headers.parseDateTime(resp.headers['Last-Modified']), int(now))

        cache_headers = {'if-modified-since': resp.headers['Last-Modified'],

                         'if-none-match': resp.etag}

        resp = self.webapp.get('/manage', headers=cache_headers)

        self.assertEqual(resp.status_code, 304)

        self.assertEqual(len(resp.body), 0)

def req_form(user):

    return {'eid': [str(user.eid)],

            '_cw_entity_fields:%s' % user.eid: '_cw_generic_field',

            '__type:%s' % user.eid: user.__regid__

            }

class EditControllerTC(CubicWebTC):

    def setUp(self):

        CubicWebTC.setUp(self)

        self.assertIn('users', self.schema.eschema('CWGroup').get_groups('read'))

    def tearDown(self):

        CubicWebTC.tearDown(self)

        self.assertIn('users', self.schema.eschema('CWGroup').get_groups('read'))

    def test_noparam_edit(self):

        """check behaviour of this controller without any form parameter

        """

        with self.admin_access.web_request() as req:

            with self.assertRaises(ValidationError) as cm:

                self.ctrl_publish(req)

            self.assertEqual(cm.exception.errors, {None: u'no selected entities'})

    def test_validation_unique(self):

        """test creation of two linked entities

        """

        with self.admin_access.web_request() as req:

            req.form = {'eid': 'X', '__type:X': 'CWUser',

                        '_cw_entity_fields:X': 'login-subject,upassword-subject',

                        'login-subject:X': u'admin',

                        'upassword-subject:X': u'toto',

                        'upassword-subject-confirm:X': u'toto',

                    }

            with self.assertRaises(ValidationError) as cm:

                self.ctrl_publish(req)

            cm.exception.translate(text_type)

            expected = {

                '': u'some relations violate a unicity constraint',

                'login': u'login is part of violated unicity constraint',

            }

            self.assertEqual(cm.exception.errors, expected)

    def test_simultaneous_edition_only_one_commit(self):

        """ Allow two simultaneous edit view of the same entity as long as only one commits

        """

        with self.admin_access.web_request() as req:

            e = req.create_entity('BlogEntry', title=u'cubicweb.org', content=u"hop")

            expected_path = e.rest_path()

            req.cnx.commit()

            form = self.vreg['views'].select('edition', req, rset=e.as_rset(), row=0)

            html_form = lxml.html.fromstring(form.render(w=None, action='edit')).forms[0]

        with self.admin_access.web_request() as req2:

            form2 = self.vreg['views'].select('edition', req, rset=e.as_rset(), row=0)

        with self.admin_access.web_request(**dict(html_form.form_values())) as req:

            path, args = self.expect_redirect_handle_request(req, path='edit')

            self.assertEqual(path, expected_path)

    def test_simultaneous_edition_refuse_second_commit(self):

        """ Disallow committing changes to an entity edited in between """

        with self.admin_access.web_request() as req:

            e = req.create_entity('BlogEntry', title=u'cubicweb.org', content=u"hop")

            eid = e.eid

            req.cnx.commit()

            form = self.vreg['views'].select('edition', req, rset=e.as_rset(), row=0)

            html_form = lxml.html.fromstring(form.render(w=None, action='edit')).forms[0]

        with self.admin_access.web_request() as req2:

            e = req2.entity_from_eid(eid)

            e.cw_set(content = u"hip")

            req2.cnx.commit()

        form_field_name = "content-subject:%d" % eid

        form_values = dict(html_form.form_values())

        assert form_field_name in form_values

        form_values[form_field_name] = u'yep'

        with self.admin_access.web_request(**form_values) as req:

            with self.assertRaises(ValidationError) as cm:

                self.ctrl_publish(req)

            reported_eid, dict_info = cm.exception.args

            self.assertEqual(reported_eid, eid)

            self.assertIn(None, dict_info)

            self.assertIn("has changed since you started to edit it.", dict_info[None])

    def test_user_editing_itself(self):

        """checking that a manager user can edit itself

        """

        with self.admin_access.web_request() as req:

            user = req.user

            groupeids = [eid for eid, in req.execute('CWGroup G WHERE G name '

                                                     'in ("managers", "users")')]

            groups = [text_type(eid) for eid in groupeids]

            eid = text_type(user.eid)

            req.form = {

                'eid': eid, '__type:'+eid: 'CWUser',

                '_cw_entity_fields:'+eid: 'login-subject,firstname-subject,surname-subject,in_group-subject',

                'login-subject:'+eid:     text_type(user.login),

                'surname-subject:'+eid: u'Th\xe9nault',

                'firstname-subject:'+eid:   u'Sylvain',

                'in_group-subject:'+eid:  groups,

                }

            self.expect_redirect_handle_request(req, 'edit')

            e = req.execute('Any X WHERE X eid %(x)s',

                            {'x': user.eid}).get_entity(0, 0)

            self.assertEqual(e.firstname, u'Sylvain')

            self.assertEqual(e.surname, u'Th\xe9nault')

            self.assertEqual(e.login, user.login)

            self.assertEqual([g.eid for g in e.in_group], groupeids)

    def test_user_can_change_its_password(self):

        with self.admin_access.repo_cnx() as cnx:

            self.create_user(cnx, u'user')

            cnx.commit()

        with self.new_access(u'user').web_request() as req:

            eid = text_type(req.user.eid)

            req.form = {

                'eid': eid, '__maineid' : eid,

                '__type:'+eid: 'CWUser',

                '_cw_entity_fields:'+eid: 'upassword-subject',

                'upassword-subject:'+eid: 'tournicoton',

                'upassword-subject-confirm:'+eid: 'tournicoton',

                }

            path, params = self.expect_redirect_handle_request(req, 'edit')

            req.cnx.commit() # commit to check we don't get late validation error for instance

            self.assertEqual(path, 'cwuser/user')

            self.assertNotIn('vid', params)

    def test_user_editing_itself_no_relation(self):

        """checking we can edit an entity without specifying some required

        relations (meaning no changes)

        """

        with self.admin_access.web_request() as req:

            user = req.user

            groupeids = [g.eid for g in user.in_group]

            eid = text_type(user.eid)

            req.form = {

                'eid':       eid,

                '__type:'+eid:    'CWUser',

                '_cw_entity_fields:'+eid: 'login-subject,firstname-subject,surname-subject',

                'login-subject:'+eid:     text_type(user.login),

                'firstname-subject:'+eid: u'Th\xe9nault',

                'surname-subject:'+eid:   u'Sylvain',

                }

            self.expect_redirect_handle_request(req, 'edit')

            e = req.execute('Any X WHERE X eid %(x)s',

                            {'x': user.eid}).get_entity(0, 0)

            self.assertEqual(e.login, user.login)

            self.assertEqual(e.firstname, u'Th\xe9nault')

            self.assertEqual(e.surname, u'Sylvain')

            self.assertEqual([g.eid for g in e.in_group], groupeids)

            self.assertEqual(e.cw_adapt_to('IWorkflowable').state, 'activated')

    def test_create_multiple_linked(self):

        with self.admin_access.web_request() as req:

            gueid = req.execute('CWGroup G WHERE G name "users"')[0][0]

            req.form = {'eid': ['X', 'Y'], '__maineid' : 'X',

                        '__type:X': 'CWUser',

                        '_cw_entity_fields:X': 'login-subject,upassword-subject,surname-subject,in_group-subject',

                        'login-subject:X': u'adim',

                        'upassword-subject:X': u'toto', 'upassword-subject-confirm:X': u'toto',

                        'surname-subject:X': u'Di Mascio',

                        'in_group-subject:X': text_type(gueid),

                        '__type:Y': 'EmailAddress',

                        '_cw_entity_fields:Y': 'address-subject,use_email-object',

                        'address-subject:Y': u'dima@logilab.fr',

                        'use_email-object:Y': 'X',

                        }

            path, _params = self.expect_redirect_handle_request(req, 'edit')

            # should be redirected on the created person

            self.assertEqual(path, 'cwuser/adim')

            e = req.execute('Any P WHERE P surname "Di Mascio"').get_entity(0, 0)

            self.assertEqual(e.surname, 'Di Mascio')

            email = e.use_email[0]

            self.assertEqual(email.address, 'dima@logilab.fr')

    def test_create_mandatory_inlined(self):

        with self.admin_access.web_request() as req:

            req.form = {'eid': ['X', 'Y'], '__maineid' : 'X',

                        '__type:X': 'Salesterm',

                        '_cw_entity_fields:X': '',

                        '__type:Y': 'File',

                        '_cw_entity_fields:Y': 'data-subject,described_by_test-object',

                        'data-subject:Y': (u'coucou.txt', Binary(b'coucou')),

                        'described_by_test-object:Y': 'X',

                        }

            path, _params = self.expect_redirect_handle_request(req, 'edit')

            self.assertTrue(path.startswith('salesterm/'), path)

            eid = path.split('/')[1]

            salesterm = req.entity_from_eid(eid)

            # The NOT NULL constraint of mandatory relation implies that the File

            # must be created before the Salesterm, otherwise Salesterm insertion

            # will fail.

            # NOTE: sqlite does have NOT NULL constraint, unlike Postgres so the

            # insertion does not fail and we have to check dumbly that File is

            # created before.

            self.assertGreater(salesterm.eid, salesterm.described_by_test[0].eid)

    def test_create_mandatory_inlined2(self):

        with self.admin_access.web_request() as req:

            req.form = {'eid': ['X', 'Y'], '__maineid' : 'X',

                        '__type:X': 'Salesterm',

                        '_cw_entity_fields:X': 'described_by_test-subject',

                        'described_by_test-subject:X': 'Y',

                        '__type:Y': 'File',

                        '_cw_entity_fields:Y': 'data-subject',

                        'data-subject:Y': (u'coucou.txt', Binary(b'coucou')),

                        }

            path, _params = self.expect_redirect_handle_request(req, 'edit')

            self.assertTrue(path.startswith('salesterm/'), path)

            eid = path.split('/')[1]

            salesterm = req.entity_from_eid(eid)

            # The NOT NULL constraint of mandatory relation implies that the File

            # must be created before the Salesterm, otherwise Salesterm insertion

            # will fail.

            # NOTE: sqlite does have NOT NULL constraint, unlike Postgres so the

            # insertion does not fail and we have to check dumbly that File is

            # created before.

            self.assertGreater(salesterm.eid, salesterm.described_by_test[0].eid)

    def test_edit_mandatory_inlined3_object(self):

        # non regression test for #3120495. Without the fix, leads to

        # "unhashable type: 'list'" error

        with self.admin_access.web_request() as req:

            cwrelation = text_type(req.execute('CWEType X WHERE X name "CWSource"')[0][0])

            req.form = {'eid': [cwrelation], '__maineid' : cwrelation,

                        '__type:'+cwrelation: 'CWEType',

                        '_cw_entity_fields:'+cwrelation: 'to_entity-object',

                        'to_entity-object:'+cwrelation: [9999, 9998],

                        }

            with req.cnx.deny_all_hooks_but():

                path, _params = self.expect_redirect_handle_request(req, 'edit')

            self.assertTrue(path.startswith('cwetype/CWSource'), path)

    def test_edit_multiple_linked(self):

        with self.admin_access.web_request() as req:

            peid = text_type(self.create_user(req, u'adim').eid)

            req.form = {'eid': [peid, 'Y'], '__maineid': peid,

                        '__type:'+peid: u'CWUser',

                        '_cw_entity_fields:'+peid: u'surname-subject',

                        'surname-subject:'+peid: u'Di Masci',

                        '__type:Y': u'EmailAddress',

                        '_cw_entity_fields:Y': u'address-subject,use_email-object',

                        'address-subject:Y': u'dima@logilab.fr',

                        'use_email-object:Y': peid,

                        }

            path, _params = self.expect_redirect_handle_request(req, 'edit')

            # should be redirected on the created person

            self.assertEqual(path, 'cwuser/adim')

            e = req.execute('Any P WHERE P surname "Di Masci"').get_entity(0, 0)

            email = e.use_email[0]

            self.assertEqual(email.address, 'dima@logilab.fr')

        # with self.admin_access.web_request() as req:

            emaileid = text_type(email.eid)

            req.form = {'eid': [peid, emaileid],

                        '__type:'+peid: u'CWUser',

                        '_cw_entity_fields:'+peid: u'surname-subject',

                        'surname-subject:'+peid: u'Di Masci',

                        '__type:'+emaileid: u'EmailAddress',

                        '_cw_entity_fields:'+emaileid: u'address-subject,use_email-object',

                        'address-subject:'+emaileid: u'adim@logilab.fr',

                        'use_email-object:'+emaileid: peid,

                        }

            self.expect_redirect_handle_request(req, 'edit')

            email.cw_clear_all_caches()

            self.assertEqual(email.address, 'adim@logilab.fr')

    def test_password_confirm(self):

        """test creation of two linked entities

        """

        with self.admin_access.web_request() as req:

            user = req.user

            req.form = {'eid': 'X',

                        '__cloned_eid:X': text_type(user.eid), '__type:X': 'CWUser',

                        '_cw_entity_fields:X': 'login-subject,upassword-subject',

                        'login-subject:X': u'toto',

                        'upassword-subject:X': u'toto',

                        }

            with self.assertRaises(ValidationError) as cm:

                self.ctrl_publish(req)

            self.assertEqual({'upassword-subject': u'password and confirmation don\'t match'},