# -*- coding: iso-8859-1 -*-

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

"""unit tests for module cubicweb.server.repository"""

import threading

import time

import logging

from yams.constraints import UniqueConstraint

from yams import register_base_type, unregister_base_type

from logilab.database import get_db_helper

from cubicweb import (BadConnectionId, ValidationError,

                      UnknownEid, AuthenticationError, Unauthorized, QueryError)

from cubicweb.predicates import is_instance

from cubicweb.schema import RQLConstraint

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.devtools.repotest import tuplify

from cubicweb.server import hook

from cubicweb.server.sqlutils import SQL_PREFIX

from cubicweb.server.hook import Hook

from cubicweb.server.sources import native

from cubicweb.server.session import SessionClosedError

class RepositoryTC(CubicWebTC):

    """ singleton providing access to a persistent storage for entities

    and relation

    """

    def test_unique_together_constraint(self):

        with self.admin_access.repo_cnx() as cnx:

            cnx.execute('INSERT Societe S: S nom "Logilab", S type "SSLL", S cp "75013"')

            with self.assertRaises(ValidationError) as wraperr:

                cnx.execute('INSERT Societe S: S nom "Logilab", S type "SSLL", S cp "75013"')

            self.assertEqual(

                {'cp': u'%(KEY-rtype)s is part of violated unicity constraint',

                 'nom': u'%(KEY-rtype)s is part of violated unicity constraint',

                 'type': u'%(KEY-rtype)s is part of violated unicity constraint',

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

                wraperr.exception.args[1])

    def test_unique_together_schema(self):

        person = self.repo.schema.eschema('Personne')

        self.assertEqual(len(person._unique_together), 1)

        self.assertItemsEqual(person._unique_together[0],

                                           ('nom', 'prenom', 'inline2'))

    def test_all_entities_have_owner(self):

        with self.admin_access.repo_cnx() as cnx:

            self.assertFalse(cnx.execute('Any X WHERE NOT X owned_by U'))

    def test_all_entities_have_is(self):

        with self.admin_access.repo_cnx() as cnx:

            self.assertFalse(cnx.execute('Any X WHERE NOT X is ET'))

    def test_all_entities_have_cw_source(self):

        with self.admin_access.repo_cnx() as cnx:

            self.assertFalse(cnx.execute('Any X WHERE NOT X cw_source S'))

    def test_connect(self):

        cnxid = self.repo.connect(self.admlogin, password=self.admpassword)

        self.assert_(cnxid)

        self.repo.close(cnxid)

        self.assertRaises(AuthenticationError,

                          self.repo.connect, self.admlogin, password='nimportnawak')

        self.assertRaises(AuthenticationError,

                          self.repo.connect, self.admlogin, password='')

        self.assertRaises(AuthenticationError,

                          self.repo.connect, self.admlogin, password=None)

        self.assertRaises(AuthenticationError,

                          self.repo.connect, None, password=None)

        self.assertRaises(AuthenticationError,

                          self.repo.connect, self.admlogin)

        self.assertRaises(AuthenticationError,

                          self.repo.connect, None)

    def test_execute(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        repo.execute(cnxid, 'Any X')

        repo.execute(cnxid, 'Any X where X is Personne')

        repo.execute(cnxid, 'Any X where X is Personne, X nom ~= "to"')

        repo.execute(cnxid, 'Any X WHERE X has_text %(text)s', {'text': u'\xe7a'})

        repo.close(cnxid)

    def test_login_upassword_accent(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_group G WHERE G name "users"',

                     {'login': u"barnab�", 'passwd': u"h�h�h�".encode('UTF8')})

        repo.commit(cnxid)

        repo.close(cnxid)

        cnxid = repo.connect(u"barnab�", password=u"h�h�h�".encode('UTF8'))

        self.assert_(cnxid)

        repo.close(cnxid)

    def test_rollback_on_commit_error(self):

        cnxid = self.repo.connect(self.admlogin, password=self.admpassword)

        self.repo.execute(cnxid,

                          'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s',

                          {'login': u"tutetute", 'passwd': 'tutetute'})

        self.assertRaises(ValidationError, self.repo.commit, cnxid)

        self.assertFalse(self.repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"'))

        self.repo.close(cnxid)

    def test_rollback_on_execute_validation_error(self):

        class ValidationErrorAfterHook(Hook):

            __regid__ = 'valerror-after-hook'

            __select__ = Hook.__select__ & is_instance('CWGroup')

            events = ('after_update_entity',)

            def __call__(self):

                raise ValidationError(self.entity.eid, {})

        with self.admin_access.repo_cnx() as cnx:

            with self.temporary_appobjects(ValidationErrorAfterHook):

                self.assertRaises(ValidationError,

                                  cnx.execute, 'SET X name "toto" WHERE X is CWGroup, X name "guests"')

                self.assertTrue(cnx.execute('Any X WHERE X is CWGroup, X name "toto"'))

                with self.assertRaises(QueryError) as cm:

                    cnx.commit()

                self.assertEqual(str(cm.exception), 'transaction must be rolled back')

                cnx.rollback()

                self.assertFalse(cnx.execute('Any X WHERE X is CWGroup, X name "toto"'))

    def test_rollback_on_execute_unauthorized(self):

        class UnauthorizedAfterHook(Hook):

            __regid__ = 'unauthorized-after-hook'

            __select__ = Hook.__select__ & is_instance('CWGroup')

            events = ('after_update_entity',)

            def __call__(self):

                raise Unauthorized()

        with self.admin_access.repo_cnx() as cnx:

            with self.temporary_appobjects(UnauthorizedAfterHook):

                self.assertRaises(Unauthorized,

                                  cnx.execute, 'SET X name "toto" WHERE X is CWGroup, X name "guests"')

                self.assertTrue(cnx.execute('Any X WHERE X is CWGroup, X name "toto"'))

                with self.assertRaises(QueryError) as cm:

                    cnx.commit()

                self.assertEqual(str(cm.exception), 'transaction must be rolled back')

                cnx.rollback()

                self.assertFalse(cnx.execute('Any X WHERE X is CWGroup, X name "toto"'))

    def test_close(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        self.assert_(cnxid)

        repo.close(cnxid)

        self.assertRaises(BadConnectionId, repo.execute, cnxid, 'Any X')

    def test_invalid_cnxid(self):

        self.assertRaises(BadConnectionId, self.repo.execute, 0, 'Any X')

        self.assertRaises(BadConnectionId, self.repo.close, None)

    def test_shared_data(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        repo.set_shared_data(cnxid, 'data', 4)

        cnxid2 = repo.connect(self.admlogin, password=self.admpassword)

        self.assertEqual(repo.get_shared_data(cnxid, 'data'), 4)

        self.assertEqual(repo.get_shared_data(cnxid2, 'data'), None)

        repo.set_shared_data(cnxid2, 'data', 5)

        self.assertEqual(repo.get_shared_data(cnxid, 'data'), 4)

        self.assertEqual(repo.get_shared_data(cnxid2, 'data'), 5)

        repo.get_shared_data(cnxid2, 'data', pop=True)

        self.assertEqual(repo.get_shared_data(cnxid, 'data'), 4)

        self.assertEqual(repo.get_shared_data(cnxid2, 'data'), None)

        repo.close(cnxid)

        repo.close(cnxid2)

        self.assertRaises(BadConnectionId, repo.get_shared_data, cnxid, 'data')

        self.assertRaises(BadConnectionId, repo.get_shared_data, cnxid2, 'data')

        self.assertRaises(BadConnectionId, repo.set_shared_data, cnxid, 'data', 1)

        self.assertRaises(BadConnectionId, repo.set_shared_data, cnxid2, 'data', 1)

    def test_check_session(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        self.assertIsInstance(repo.check_session(cnxid), float)

        repo.close(cnxid)

        self.assertRaises(BadConnectionId, repo.check_session, cnxid)

    def test_transaction_base(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        # check db state

        result = repo.execute(cnxid, 'Personne X')

        self.assertEqual(result.rowcount, 0)

        # rollback entity insertion

        repo.execute(cnxid, "INSERT Personne X: X nom 'bidule'")

        result = repo.execute(cnxid, 'Personne X')

        self.assertEqual(result.rowcount, 1)

        repo.rollback(cnxid)

        result = repo.execute(cnxid, 'Personne X')

        self.assertEqual(result.rowcount, 0, result.rows)

        # commit

        repo.execute(cnxid, "INSERT Personne X: X nom 'bidule'")

        repo.commit(cnxid)

        result = repo.execute(cnxid, 'Personne X')

        self.assertEqual(result.rowcount, 1)

        repo.close(cnxid)

    def test_transaction_base2(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        # rollback relation insertion

        repo.execute(cnxid, "SET U in_group G WHERE U login 'admin', G name 'guests'")

        result = repo.execute(cnxid, "Any U WHERE U in_group G, U login 'admin', G name 'guests'")

        self.assertEqual(result.rowcount, 1)

        repo.rollback(cnxid)

        result = repo.execute(cnxid, "Any U WHERE U in_group G, U login 'admin', G name 'guests'")

        self.assertEqual(result.rowcount, 0, result.rows)

        repo.close(cnxid)

    def test_transaction_base3(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        # rollback state change which trigger TrInfo insertion

        session = repo._get_session(cnxid)

        user = session.user

        user.cw_adapt_to('IWorkflowable').fire_transition('deactivate')

        rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': user.eid})

        self.assertEqual(len(rset), 1)

        repo.rollback(cnxid)

        rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': user.eid})

        self.assertEqual(len(rset), 0)

        repo.close(cnxid)

    def test_close_kill_processing_request(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        repo.execute(cnxid, 'INSERT CWUser X: X login "toto", X upassword "tutu", X in_group G WHERE G name "users"')

        repo.commit(cnxid)

        lock = threading.Lock()

        lock.acquire()

        # close has to be in the thread due to sqlite limitations

        def close_in_a_few_moment():

            lock.acquire()

            repo.close(cnxid)

        t = threading.Thread(target=close_in_a_few_moment)

        t.start()

        def run_transaction():

            lock.release()

            repo.execute(cnxid, 'DELETE CWUser X WHERE X login "toto"')

            repo.commit(cnxid)

        try:

            with self.assertRaises(SessionClosedError) as cm:

                run_transaction()

            self.assertEqual(str(cm.exception), 'try to access connections set on a closed session %s' % cnxid)

        finally:

            t.join()

    def test_initial_schema(self):

        schema = self.repo.schema

        # check order of attributes is respected

        notin = set(('eid', 'is', 'is_instance_of', 'identity',

                     'creation_date', 'modification_date', 'cwuri',

                     'owned_by', 'created_by', 'cw_source',

                     'update_permission', 'read_permission',

                     'add_permission', 'in_basket'))

        self.assertListEqual(['relation_type',

                              'from_entity', 'to_entity',

                              'constrained_by',

                              'cardinality', 'ordernum', 'formula',

                              'indexed', 'fulltextindexed', 'internationalizable',

                              'defaultval', 'extra_props',

                              'description', 'description_format'],

                             [r.type

                              for r in schema.eschema('CWAttribute').ordered_relations()

                              if r.type not in notin])

        self.assertEqual(schema.eschema('CWEType').main_attribute(), 'name')

        self.assertEqual(schema.eschema('State').main_attribute(), 'name')

        constraints = schema.rschema('name').rdef('CWEType', 'String').constraints

        self.assertEqual(len(constraints), 2)

        for cstr in constraints[:]:

            if isinstance(cstr, UniqueConstraint):

                constraints.remove(cstr)

                break

        else:

            self.fail('unique constraint not found')

        sizeconstraint = constraints[0]

        self.assertEqual(sizeconstraint.min, None)

        self.assertEqual(sizeconstraint.max, 64)

        constraints = schema.rschema('relation_type').rdef('CWAttribute', 'CWRType').constraints

        self.assertEqual(len(constraints), 1)

        cstr = constraints[0]

        self.assert_(isinstance(cstr, RQLConstraint))

        self.assertEqual(cstr.expression, 'O final TRUE')

        ownedby = schema.rschema('owned_by')

        self.assertEqual(ownedby.objects('CWEType'), ('CWUser',))

    def test_internal_api(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        session = repo._get_session(cnxid, setcnxset=True)

        self.assertEqual(repo.type_and_source_from_eid(2, session),

                         ('CWGroup', None, 'system'))

        self.assertEqual(repo.type_from_eid(2, session), 'CWGroup')

        repo.close(cnxid)

    def test_public_api(self):

        self.assertEqual(self.repo.get_schema(), self.repo.schema)

        self.assertEqual(self.repo.source_defs(), {'system': {'type': 'native',

                                                              'uri': 'system',

                                                              'use-cwuri-as-url': False}

                                                  })

        # .properties() return a result set

        self.assertEqual(self.repo.properties().rql, 'Any K,V WHERE P is CWProperty,P pkey K, P value V, NOT P for_user U')

    def test_session_api(self):

        repo = self.repo

        cnxid = repo.connect(self.admlogin, password=self.admpassword)

        self.assertEqual(repo.user_info(cnxid), (6, 'admin', set([u'managers']), {}))

        self.assertEqual({'type': u'CWGroup', 'extid': None, 'source': 'system'},

                         repo.entity_metas(cnxid, 2))

        self.assertEqual(repo.describe(cnxid, 2), (u'CWGroup', 'system', None, 'system'))

        repo.close(cnxid)

        self.assertRaises(BadConnectionId, repo.user_info, cnxid)

        self.assertRaises(BadConnectionId, repo.describe, cnxid, 1)

    def test_shared_data_api(self):

        repo = self.repo