# copyright 2003-2014 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 logilab.common.decorators import monkeypatch

from cubicweb import Binary, NoSelectableObject, ValidationError, AuthenticationError

from cubicweb.schema import RRQLExpression

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

import cubicweb.server.session

from cubicweb.server.session import Connection

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

from cubicweb.web.views.basecontrollers import JSonController, xhtmlize, jsonize

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

import cubicweb.transaction as tx

from cubicweb.server.hook import Hook, Operation

from cubicweb.predicates import is_instance

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',