# copyright 2003-2010 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/>.

from datetime import datetime, timedelta

from cubicweb.devtools import TestServerConfiguration, init_test_database

from cubicweb.devtools.repotest import do_monkey_patch, undo_monkey_patch

class ExternalSource1Configuration(TestServerConfiguration):

    sourcefile = 'sources_extern'

class ExternalSource2Configuration(TestServerConfiguration):

    sourcefile = 'sources_multi'

MTIME = datetime.now() - timedelta(0, 10)

EXTERN_SOURCE_CFG = u'''

pyro-ns-id = extern

cubicweb-user = admin

cubicweb-password = gingkow

mapping-file = extern_mapping.py

base-url=http://extern.org/

'''

# hi-jacking

from cubicweb.server.sources.pyrorql import PyroRQLSource

from cubicweb.dbapi import Connection

PyroRQLSource_get_connection = PyroRQLSource.get_connection

Connection_close = Connection.close

class TwoSourcesTC(CubicWebTC):

    """Main repo -> extern-multi -> extern

                  \-------------/

    """

    @classmethod

    def setUp(self):

        CubicWebTC.setUp(self)

        do_monkey_patch()

    def tearDown(self):

        for source in self.repo.sources[1:]:

            self.repo.remove_source(source.uri)

        CubicWebTC.tearDown(self)

        undo_monkey_patch()

                            ('extern-multi', '''

pyro-ns-id = extern-multi

cubicweb-user = admin

cubicweb-password = gingkow

mapping-file = extern_mapping.py

''')]:

                                         type=u'pyrorql',

        # trigger discovery

        # add some entities

        self.ic1 = self.sexecute('INSERT Card X: X title "C1: An internal card", X wikiid "aaai"')[0][0]

        self.ic2 = self.sexecute('INSERT Card X: X title "C2: Ze internal card", X wikiid "zzzi"')[0][0]

    def test_eid_comp(self):

        rset = self.sexecute('Card X WHERE X eid > 1')

        self.assertEqual(len(rset), 4)

        rset = self.sexecute('Any X,T WHERE X title T, X eid > 1')

        self.assertEqual(len(rset), 4)

    def test_metainformation(self):

        rset = self.sexecute('Card X ORDERBY T WHERE X title T')

        # 2 added to the system source, 2 added to the external source

        self.assertEqual(len(rset), 4)

        # since they are orderd by eid, we know the 3 first one is coming from the system source

        # and the others from external source

        self.assertEqual(rset.get_entity(0, 0).cw_metainformation(),

                          {'source': {'type': 'native', 'uri': 'system'},

                           'type': u'Card', 'extid': None})

        externent = rset.get_entity(3, 0)

        metainf = externent.cw_metainformation()

        self.assertEqual(metainf['source'], {'type': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'})

        self.assertEqual(metainf['type'], 'Card')

        self.assert_(metainf['extid'])

        etype = self.sexecute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s',

                             {'x': externent.eid})[0][0]

        self.assertEqual(etype, 'Card')

    def test_order_limit_offset(self):

        rsetbase = self.sexecute('Any W,X ORDERBY W,X WHERE X wikiid W')

        self.assertEqual(len(rsetbase), 4)

        self.assertEqual(sorted(rsetbase.rows), rsetbase.rows)

        rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W')

        self.assertEqual(rset.rows, rsetbase.rows[2:4])

    def test_has_text(self):

        self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before

        self.failUnless(self.sexecute('Any X WHERE X has_text "affref"'))

        self.failUnless(self.sexecute('Affaire X WHERE X has_text "affref"'))

        self.failUnless(self.sexecute('Any X ORDERBY FTIRANK(X) WHERE X has_text "affref"'))

        self.failUnless(self.sexecute('Affaire X ORDERBY FTIRANK(X) WHERE X has_text "affref"'))

    def test_anon_has_text(self):

        self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before

        self.sexecute('INSERT Affaire X: X ref "no readable card"')[0][0]

        aff1 = self.sexecute('INSERT Affaire X: X ref "card"')[0][0]

        # grant read access

        self.sexecute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1})

        self.commit()

        cnx = self.login('anon')

        cu = cnx.cursor()

        rset = cu.execute('Any X WHERE X has_text "card"')

        # 5: 4 card + 1 readable affaire

        self.assertEqual(len(rset), 5, zip(rset.rows, rset.description))

        rset = cu.execute('Any X ORDERBY FTIRANK(X) WHERE X has_text "card"')

        self.assertEqual(len(rset), 5, zip(rset.rows, rset.description))

        Connection_close(cnx.cnx) # cnx is a TestCaseConnectionProxy

    def test_synchronization(self):

        assert cu.execute('Any X WHERE X eid %(x)s', {'x': self.aff1})

        cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': self.aff1})

        aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0]

        try:

            # force sync

            self.repo.sources_by_uri['extern'].synchronize(MTIME)

            self.failUnless(self.sexecute('Any X WHERE X has_text "blah"'))

            self.failUnless(self.sexecute('Any X WHERE X has_text "affreux"'))

            cu.execute('DELETE Affaire X WHERE X eid %(x)s', {'x': aff2})

            self.repo.sources_by_uri['extern'].synchronize(MTIME)

            rset = self.sexecute('Any X WHERE X has_text "affreux"')

            self.failIf(rset)

        finally:

            # restore state

            cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': self.aff1})

    def test_simplifiable_var(self):

        affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0]

        rset = self.sexecute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB',

                            {'x': affeid})

        self.assertEqual(len(rset), 1)

        self.assertEqual(rset[0][1], "pitetre")

    def test_simplifiable_var_2(self):

        affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0]

        rset = self.sexecute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"',

                             {'x': affeid, 'u': self.session.user.eid})

        self.assertEqual(len(rset), 1)

    def test_sort_func(self):

        self.sexecute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF')

    def test_sort_func_ambigous(self):

        self.sexecute('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF')

    def test_in_eid(self):

        iec1 = self.repo.extid2eid(self.repo.sources_by_uri['extern'], str(self.ec1),

                                   'Card', self.session)

        rset = self.sexecute('Any X WHERE X eid IN (%s, %s)' % (iec1, self.ic1))

        self.assertEqual(sorted(r[0] for r in rset.rows), sorted([iec1, self.ic1]))

    def test_greater_eid(self):

        rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1))

        self.assertEqual(len(rset.rows), 2) # self.ic1 and self.ic2

        ec2 = cu.execute('INSERT Card X: X title "glup"')[0][0]

        # 'X eid > something' should not trigger discovery

        rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1))

        self.assertEqual(len(rset.rows), 2)

        # trigger discovery using another query

        crset = self.sexecute('Card X WHERE X title "glup"')

        self.assertEqual(len(crset.rows), 1)

        rset = self.sexecute('Any X WHERE X eid > %s' % (self.ic1 - 1))

        self.assertEqual(len(rset.rows), 3)

        rset = self.sexecute('Any MAX(X)')

        self.assertEqual(len(rset.rows), 1)

        self.assertEqual(rset.rows[0][0], crset[0][0])

    def test_attr_unification_1(self):

        n1 = self.sexecute('INSERT Note X: X type "AFFREF"')[0][0]

        n2 = self.sexecute('INSERT Note X: X type "AFFREU"')[0][0]

        rset = self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X type T, Y ref T')

        self.assertEqual(len(rset), 1, rset.rows)

    def test_attr_unification_2(self):

        ec2 = cu.execute('INSERT Card X: X title "AFFREF"')[0][0]

        try:

            c1 = self.sexecute('INSERT Card C: C title "AFFREF"')[0][0]

            rset = self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T')

            self.assertEqual(len(rset), 2, rset.rows)

        finally:

            cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2})

    def test_attr_unification_neq_1(self):

        # XXX complete

        self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D')

    def test_attr_unification_neq_2(self):

        # XXX complete

        self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X creation_date D, Y creation_date > D')

    def test_union(self):

        afeids = self.sexecute('Affaire X')

        ueids = self.sexecute('CWUser X')

        rset = self.sexecute('(Any X WHERE X is Affaire) UNION (Any X WHERE X is CWUser)')

        self.assertEqual(sorted(r[0] for r in rset.rows),

                          sorted(r[0] for r in afeids + ueids))

    def test_subquery1(self):

        rsetbase = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)')

        self.assertEqual(len(rsetbase), 4)

        self.assertEqual(sorted(rsetbase.rows), rsetbase.rows)

        rset = self.sexecute('Any W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)')

        self.assertEqual(rset.rows, rsetbase.rows[2:4])

        rset = self.sexecute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X WHERE X wikiid W)')

        self.assertEqual(rset.rows, rsetbase.rows[2:4])

        rset = self.sexecute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W)')

        self.assertEqual(rset.rows, rsetbase.rows[2:4])

    def test_subquery2(self):

        affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0]

        rset = self.sexecute('Any X,AA,AB WITH X,AA,AB BEING (Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB)',

                            {'x': affeid})

        self.assertEqual(len(rset), 1)

        self.assertEqual(rset[0][1], "pitetre")

    def test_not_relation(self):

        states = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN'))

        userstate = self.session.user.in_state[0]

        states.remove((userstate.eid, userstate.name))

        notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s',

                                                       {'x': self.session.user.eid}))

        self.assertSetEqual(notstates, states)

        aff1 = self.sexecute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0]

        aff1stateeid, aff1statename = self.sexecute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1})[0]

        self.assertEqual(aff1statename, 'pitetre')

        states.add((userstate.eid, userstate.name))

        states.remove((aff1stateeid, aff1statename))

        notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s',

                                                       {'x': aff1}))

        self.assertSetEqual(notstates, states)

    def test_absolute_url_base_url(self):

        ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0]

        lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0)

        self.assertEqual(lc.absolute_url(), 'http://extern.org/card/eid/%s' % ceid)

        cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid})

    def test_absolute_url_no_base_url(self):

        ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0]

        lc = self.sexecute('Card X WHERE X title "without wikiid to get eid based url"').get_entity(0, 0)

        self.assertEqual(lc.absolute_url(), 'http://testing.fr/cubicweb/card/eid/%s' % lc.eid)

        cu.execute('DELETE Card X WHERE X eid %(x)s', {'x':ceid})