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

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

import os

import sys

import threading

import time

from copy import deepcopy

from mx.DateTime import DateTimeType, now

from logilab.common.testlib import TestCase, unittest_main

from cubicweb.devtools.apptest import RepositoryBasedTC

from cubicweb.devtools.repotest import tuplify

from yams.constraints import UniqueConstraint

from cubicweb import BadConnectionId, RepositoryError, ValidationError, UnknownEid, AuthenticationError

from cubicweb.schema import CubicWebSchema, RQLConstraint

from cubicweb.dbapi import connect, repo_connect

from cubicweb.server import repository 

# start name server anyway, process will fail if already running

os.system('pyro-ns >/dev/null 2>/dev/null &')

class RepositoryTC(RepositoryBasedTC):

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

    and relation

    """

#     def setUp(self):

#         pass

#     def tearDown(self):

#         self.repo.config.db_perms = True

#         cnxid = self.repo.connect(*self.default_user_password())

#         for etype in ('Affaire', 'Note', 'Societe', 'Personne'):

#             self.repo.execute(cnxid, 'DELETE %s X' % etype)

#             self.repo.commit(cnxid)

#         self.repo.close(cnxid)

    def test_fill_schema(self):

        self.repo.schema = CubicWebSchema(self.repo.config.appid)

        self.repo.config._cubes = None # avoid assertion error

        self.repo.fill_schema()

        pool = self.repo._get_pool()

        try:

            sqlcursor = pool['system']

            sqlcursor.execute('SELECT name FROM EEType WHERE final is NULL')

            self.assertEquals(sqlcursor.fetchall(), [])

            sqlcursor.execute('SELECT name FROM EEType WHERE final=%(final)s ORDER BY name', {'final': 'TRUE'})

            self.assertEquals(sqlcursor.fetchall(), [(u'Boolean',), (u'Bytes',),

                                                     (u'Date',), (u'Datetime',),

                                                     (u'Decimal',),(u'Float',),

                                                     (u'Int',),

                                                     (u'Interval',), (u'Password',),

                                                     (u'String',), (u'Time',)])

        finally:

            self.repo._free_pool(pool)

    def test_schema_has_owner(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        self.failIf(repo.execute(cnxid, 'EEType X WHERE NOT X owned_by U'))

        self.failIf(repo.execute(cnxid, 'ERType X WHERE NOT X owned_by U'))

        self.failIf(repo.execute(cnxid, 'EFRDef X WHERE NOT X owned_by U'))

        self.failIf(repo.execute(cnxid, 'ENFRDef X WHERE NOT X owned_by U'))

        self.failIf(repo.execute(cnxid, 'EConstraint X WHERE NOT X owned_by U'))

        self.failIf(repo.execute(cnxid, 'EConstraintType X WHERE NOT X owned_by U'))

    def test_connect(self):

        login, passwd = self.default_user_password()

        self.assert_(self.repo.connect(login, passwd))

        self.assertRaises(AuthenticationError,

                          self.repo.connect, login, 'nimportnawak')

        self.assertRaises(AuthenticationError,

                          self.repo.connect, login, None)

        self.assertRaises(AuthenticationError,

                          self.repo.connect, None, None)

    def test_execute(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        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.default_user_password())

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

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

        repo.commit(cnxid)

        repo.close(cnxid)

        self.assert_(repo.connect(u"barnab�", u"h�h�h�".encode('UTF8')))

    def test_invalid_entity_rollback(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        repo.execute(cnxid, 'INSERT EUser X: X login %(login)s, X upassword %(passwd)s, X in_state S WHERE S name "activated"',

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

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

        rset = repo.execute(cnxid, 'EUser X WHERE X login "tutetute"')

        self.assertEquals(rset.rowcount, 0)

    def test_close(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        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.default_user_password())

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

        cnxid2 = repo.connect(*self.default_user_password())

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

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

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

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

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

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

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

        self.assertEquals(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.default_user_password())

        self.assertEquals(repo.check_session(cnxid), None)

        repo.close(cnxid)

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

    def test_transaction_base(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        # check db state

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

        self.assertEquals(result.rowcount, 0)

        # rollback entity insertion

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

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

        self.assertEquals(result.rowcount, 1)

        repo.rollback(cnxid)

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

        self.assertEquals(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.assertEquals(result.rowcount, 1)

    def test_transaction_base2(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        # 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.assertEquals(result.rowcount, 1)

        repo.rollback(cnxid)

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

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

    def test_transaction_base3(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

        # rollback state change which trigger TrInfo insertion

        ueid = repo._get_session(cnxid).user.eid

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

        self.assertEquals(len(rset), 1)

        repo.execute(cnxid, 'SET X in_state S WHERE X eid %(x)s, S name "deactivated"',

                     {'x': ueid}, 'x')

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

        self.assertEquals(len(rset), 2)

        repo.rollback(cnxid)

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

        self.assertEquals(len(rset), 1)

    def test_transaction_interleaved(self):

        self.skip('implement me')

    def test_initial_schema(self):

        schema = self.repo.schema

        # check order of attributes is respected

        self.assertListEquals([r.type for r in schema.eschema('EFRDef').ordered_relations()

                               if not r.type in ('eid', 'is', 'is_instance_of', 'identity', 

                                                 'creation_date', 'modification_date',

                                                 'owned_by', 'created_by')],

                              ['relation_type', 'from_entity', 'to_entity', 'constrained_by',

                               'cardinality', 'ordernum', 

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

                               'defaultval', 'description_format', 'description'])

        self.assertEquals(schema.eschema('EEType').main_attribute(), 'name')

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

        constraints = schema.rschema('name').rproperty('EEType', 'String', 'constraints')

        self.assertEquals(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.assertEquals(sizeconstraint.min, None)

        self.assertEquals(sizeconstraint.max, 64)

        constraints = schema.rschema('relation_type').rproperty('EFRDef', 'ERType', 'constraints')

        self.assertEquals(len(constraints), 1)

        cstr = constraints[0]

        self.assert_(isinstance(cstr, RQLConstraint))

        self.assertEquals(cstr.restriction, 'O final TRUE')

        ownedby = schema.rschema('owned_by')

        self.assertEquals(ownedby.objects('EEType'), ('EUser',))

    def test_pyro(self):

        import Pyro

        Pyro.config.PYRO_MULTITHREADED = 0

        lock = threading.Lock()

        # the client part has to be in the thread due to sqlite limitations

        t = threading.Thread(target=self._pyro_client, args=(lock,))

        try:

            daemon = self.repo.pyro_register()

            t.start()

            # connection

            daemon.handleRequests(1.0)

            daemon.handleRequests(1.0)

            daemon.handleRequests(1.0)

            # get schema

            daemon.handleRequests(1.0)

            # execute

            daemon.handleRequests(1.0)

            t.join()

        finally:

            repository.pyro_unregister(self.repo.config)

    def _pyro_client(self, lock):

        cnx = connect(self.repo.config.appid, u'admin', 'gingkow')

        # check we can get the schema

        schema = cnx.get_schema()

        self.assertEquals(schema.__hashmode__, None)

        rset = cnx.cursor().execute('Any U,G WHERE U in_group G')

    def test_internal_api(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

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

        self.assertEquals(repo.type_and_source_from_eid(1, session), ('EGroup', 'system', None))

        self.assertEquals(repo.type_from_eid(1, session), 'EGroup')

        self.assertEquals(repo.source_from_eid(1, session).uri, 'system')

        self.assertEquals(repo.eid2extid(repo.system_source, 1, session), None)

        class dummysource: uri = 'toto'

        self.assertRaises(UnknownEid, repo.eid2extid, dummysource, 1, session)

    def test_public_api(self):

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

        self.assertEquals(self.repo.source_defs(), {'system': {'adapter': 'native', 'uri': 'system'}})

        # .properties() return a result set

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

    def test_session_api(self):

        repo = self.repo

        cnxid = repo.connect(*self.default_user_password())

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

        self.assertEquals(repo.describe(cnxid, 1), (u'EGroup', u'system', None))

        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

        cnxid = repo.connect(*self.default_user_password())

        self.assertEquals(repo.get_shared_data(cnxid, 'data'), None)

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

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

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

        repo.get_shared_data(cnxid, 'whatever', pop=True)

        self.assertEquals(repo.get_shared_data(cnxid, 'data'), None)

        repo.close(cnxid)

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

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

class DataHelpersTC(RepositoryBasedTC):

    def setUp(self):

        """ called before each test from this class """

        cnxid = self.repo.connect(*self.default_user_password())

        self.session = self.repo._sessions[cnxid]

        self.session.set_pool()

    def tearDown(self):

        self.session.rollback()

    def test_create_eid(self):

        self.assert_(self.repo.system_source.create_eid(self.session))

    def test_source_from_eid(self):

        self.assertEquals(self.repo.source_from_eid(1, self.session),

                          self.repo.sources_by_uri['system'])

    def test_source_from_eid_raise(self):

        self.assertRaises(UnknownEid, self.repo.source_from_eid, -2, self.session)

    def test_type_from_eid(self):

        self.assertEquals(self.repo.type_from_eid(1, self.session), 'EGroup')

    def test_type_from_eid_raise(self):

        self.assertRaises(UnknownEid, self.repo.type_from_eid, -2, self.session)

    def test_add_delete_info(self):

        entity = self.repo.vreg.etype_class('Personne')(self.session, None, None)

        entity.eid = -1

        entity.complete = lambda x: None

        self.repo.add_info(self.session, entity, self.repo.sources_by_uri['system'])

        cursor = self.session.pool['system']

        cursor.execute('SELECT * FROM entities WHERE eid = -1')

        data = cursor.fetchall()

        self.assertIsInstance(data[0][3], DateTimeType)

        data[0] = list(data[0])

        data[0][3] = None

        self.assertEquals(tuplify(data), [(-1, 'Personne', 'system', None, None)])

        self.repo.delete_info(self.session, -1)

        #self.repo.commit()

        cursor.execute('SELECT * FROM entities WHERE eid = -1')

        data = cursor.fetchall()

        self.assertEquals(data, [])

class FTITC(RepositoryBasedTC):

    def test_reindex_and_modified_since(self):

        cursor = self.session.pool['system']

        eidp = self.execute('INSERT Personne X: X nom "toto", X prenom "tutu"')[0][0]

        self.commit()

        ts = now()

        self.assertEquals(len(self.execute('Personne X WHERE X has_text "tutu"')), 1)

        cursor.execute('SELECT mtime, eid FROM entities WHERE eid = %s' % eidp)

        omtime = cursor.fetchone()[0]

        # our sqlite datetime adapter is ignore seconds fraction, so we have to