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

"""this module contains base classes and utilities for cubicweb tests"""

from __future__ import with_statement

__docformat__ = "restructuredtext en"

import os

import sys

import re

import urlparse

from os.path import dirname, join, abspath

from urllib import unquote

from math import log

from contextlib import contextmanager

from warnings import warn

import yams.schema

from logilab.common.testlib import TestCase, InnerTest, Tags

from logilab.common.pytest import nocoverage, pause_tracing, resume_tracing

from logilab.common.debugger import Debugger

from logilab.common.umessage import message_from_string

from logilab.common.decorators import cached, classproperty, clear_cache

from logilab.common.deprecation import deprecated, class_deprecated

from logilab.common.shellutils import getlogin

from cubicweb import ValidationError, NoSelectableObject, AuthenticationError

from cubicweb import cwconfig, devtools, web, server

from cubicweb.dbapi import ProgrammingError, DBAPISession, repo_connect

from cubicweb.sobjects import notification

from cubicweb.web import Redirect, application

from cubicweb.server.session import security_enabled

from cubicweb.server.hook import SendMailOp

from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS

from cubicweb.devtools import BASE_URL, fake, htmlparser, DEFAULT_EMPTY_DB_ID

from cubicweb.utils import json

# low-level utilities ##########################################################

class CubicWebDebugger(Debugger):

    """special debugger class providing a 'view' function which saves some

    html into a temporary file and open a web browser to examinate it.

    """

    def do_view(self, arg):

        import webbrowser

        data = self._getval(arg)

        with file('/tmp/toto.html', 'w') as toto:

            toto.write(data)

        webbrowser.open('file:///tmp/toto.html')

def line_context_filter(line_no, center, before=3, after=None):

    """return true if line are in context

    if after is None: after = before

    """

    if after is None:

        after = before

    return center - before <= line_no <= center + after

def unprotected_entities(schema, strict=False):

    """returned a set of each non final entity type, excluding "system" entities

    (eg CWGroup, CWUser...)

    """

    if strict:

        protected_entities = yams.schema.BASE_TYPES

    else:

        protected_entities = yams.schema.BASE_TYPES.union(SYSTEM_ENTITIES)

    return set(schema.entities()) - protected_entities

# email handling, to test emails sent by an application ########################

MAILBOX = []

class Email:

    """you'll get instances of Email into MAILBOX during tests that trigger

    some notification.

    * `msg` is the original message object

    * `recipients` is a list of email address which are the recipients of this

      message

    """

    def __init__(self, recipients, msg):

        self.recipients = recipients

        self.msg = msg

    @property

    def message(self):

        return message_from_string(self.msg)

    @property

    def subject(self):

        return self.message.get('Subject')

    @property

    def content(self):

        return self.message.get_payload(decode=True)

    def __repr__(self):

        return '<Email to %s with subject %s>' % (','.join(self.recipients),

                                                  self.message.get('Subject'))

# the trick to get email into MAILBOX instead of actually sent: monkey patch

# cwconfig.SMTP object

class MockSMTP:

    def __init__(self, server, port):

        pass

    def close(self):

        pass

    def sendmail(self, helo_addr, recipients, msg):

        MAILBOX.append(Email(recipients, msg))

cwconfig.SMTP = MockSMTP

class TestCaseConnectionProxy(object):

    """thin wrapper around `cubicweb.dbapi.Connection` context-manager

    used in CubicWebTC (cf. `cubicweb.devtools.testlib.CubicWebTC.login` method)

    It just proxies to the default connection context manager but

    restores the original connection on exit.

    """

    def __init__(self, testcase, cnx):

        self.testcase = testcase

        self.cnx = cnx

    def __getattr__(self, attrname):

        return getattr(self.cnx, attrname)

    def __enter__(self):

        return self.cnx.__enter__()

    def __exit__(self, exctype, exc, tb):

        try:

            return self.cnx.__exit__(exctype, exc, tb)

        finally:

            self.cnx.close()

            self.testcase.restore_connection()

# base class for cubicweb tests requiring a full cw environments ###############

class CubicWebTC(TestCase):

    """abstract class for test using an apptest environment

    attributes:

    * `vreg`, the vregistry

    * `schema`, self.vreg.schema

    * `config`, cubicweb configuration

    * `cnx`, dbapi connection to the repository using an admin user

    * `session`, server side session associated to `cnx`

    * `app`, the cubicweb publisher (for web testing)

    * `repo`, the repository object

    * `admlogin`, login of the admin user

    * `admpassword`, password of the admin user

    * `shell`, create and use shell environment

    """

    appid = 'data'

    configcls = devtools.ApptestConfiguration

    reset_schema = reset_vreg = False # reset schema / vreg between tests

    tags = TestCase.tags | Tags('cubicweb', 'cw_repo')

    test_db_id = DEFAULT_EMPTY_DB_ID

    _cnxs = set() # establised connection

    _cnx  = None  # current connection

    # Too much complicated stuff. the class doesn't need to bear the repo anymore

    @classmethod

    def set_cnx(cls, cnx):

        cls._cnxs.add(cnx)

        cls._cnx = cnx

    @property

    def cnx(self):

        return self.__class__._cnx

    @classproperty

    def config(cls):

        """return the configuration object

        Configuration is cached on the test class.

        """

        try:

            assert not cls is CubicWebTC, "Don't use CubicWebTC directly to prevent database caching issue"

            return cls.__dict__['_config']

        except KeyError:

            home = abspath(join(dirname(sys.modules[cls.__module__].__file__), cls.appid))

            config = cls._config = cls.configcls(cls.appid, apphome=home)

            config.mode = 'test'

            return config

    @classmethod

    def init_config(cls, config):

        """configuration initialization hooks.

        You may only want to override here the configuraton logic.

        Otherwise, consider to use a different :class:`ApptestConfiguration`

        defined in the `configcls` class attribute"""

        source = config.sources()['system']

        cls.admlogin = unicode(source['db-user'])

        cls.admpassword = source['db-password']

        # uncomment the line below if you want rql queries to be logged

        #config.global_set_option('query-log-file',

        #                         '/tmp/test_rql_log.' + `os.getpid()`)

        config.global_set_option('log-file', None)

        # set default-dest-addrs to a dumb email address to avoid mailbox or

        # mail queue pollution

        config.global_set_option('default-dest-addrs', ['whatever'])

        send_to =  '%s@logilab.fr' % getlogin()

        config.global_set_option('sender-addr', send_to)

        config.global_set_option('default-dest-addrs', send_to)

        config.global_set_option('sender-name', 'cubicweb-test')

        config.global_set_option('sender-addr', 'cubicweb-test@logilab.fr')

        # default_base_url on config class isn't enough for TestServerConfiguration

        config.global_set_option('base-url', config.default_base_url())

        # web resources

        try:

            config.global_set_option('embed-allowed', re.compile('.*'))

        except: # not in server only configuration

            pass

    #XXX this doesn't need to a be classmethod anymore

    @classmethod

    def _init_repo(cls):

        """init the repository and connection to it.

        """

        # setup configuration for test

        cls.init_config(cls.config)

        # get or restore and working db.

        db_handler = devtools.get_test_db_handler(cls.config)

        db_handler.build_db_cache(cls.test_db_id, cls.pre_setup_database)

        cls.repo, cnx = db_handler.get_repo_and_cnx(cls.test_db_id)

        # no direct assignation to cls.cnx anymore.

        # cnx is now an instance property that use a class protected attributes.

        cls.set_cnx(cnx)

        cls.vreg = cls.repo.vreg

        cls._orig_cnx = (cnx, cls.websession)

        cls.config.repository = lambda x=None: cls.repo

    def _close_cnx(self):

        for cnx in list(self._cnxs):

            if not cnx._closed:

                cnx.rollback()

                cnx.close()

            self._cnxs.remove(cnx)

    # global resources accessors ###############################################

    @property

    def schema(self):

        """return the application schema"""

        return self.vreg.schema

    @property

    def session(self):

        """return current server side session (using default manager account)"""

        session = self.repo._sessions[self.cnx.sessionid]

        session.set_pool()

        return session

    @property

    def adminsession(self):

        """return current server side session (using default manager account)"""

        return self.repo._sessions[self._orig_cnx[0].sessionid]

    def shell(self):

        """return a shell session object"""

        from cubicweb.server.migractions import ServerMigrationHelper

        return ServerMigrationHelper(None, repo=self.repo, cnx=self.cnx,

                                     interactive=False,

                                     # hack so it don't try to load fs schema

                                     schema=1)

    def set_option(self, optname, value):

        self.config.global_set_option(optname, value)

    def set_debug(self, debugmode):

        server.set_debug(debugmode)

    def debugged(self, debugmode):

        return server.debugged(debugmode)

    # default test setup and teardown #########################################

    def setUp(self):

        # monkey patch send mail operation so emails are sent synchronously

        self._patch_SendMailOp()

        pause_tracing()

        previous_failure = self.__class__.__dict__.get('_repo_init_failed')

        if previous_failure is not None:

            self.skipTest('repository is not initialised: %r' % previous_failure)

        try:

            self._init_repo()

            self.addCleanup(self._close_cnx)

        except Exception, ex:

            self.__class__._repo_init_failed = ex

            raise

        resume_tracing()

        self.setup_database()

        self.commit()

        MAILBOX[:] = [] # reset mailbox

    def tearDown(self):

        # XXX hack until logilab.common.testlib is fixed

        while self._cleanups:

            cleanup, args, kwargs = self._cleanups.pop(-1)

            cleanup(*args, **kwargs)

    def _patch_SendMailOp(self):

        # monkey patch send mail operation so emails are sent synchronously

        _old_mail_postcommit_event = SendMailOp.postcommit_event

        SendMailOp.postcommit_event = SendMailOp.sendmails

        def reverse_SendMailOp_monkey_patch():

            SendMailOp.postcommit_event = _old_mail_postcommit_event

        self.addCleanup(reverse_SendMailOp_monkey_patch)

    def setup_database(self):

        """add your database setup code by overriding this method"""

    @classmethod

    def pre_setup_database(cls, session, config):

        """add your pre database setup code by overriding this method

        Do not forget to set the cls.test_db_id value to enable caching of the

        result.

        """

    # user / session management ###############################################

    def user(self, req=None):

        """return the application schema"""

        if req is None:

            req = self.request()

            return self.cnx.user(req)

        else:

            return req.user

    def create_user(self, login, groups=('users',), password=None, req=None,

                    commit=True, **kwargs):

        """create and return a new user entity"""

        if password is None:

            password = login.encode('utf8')

        if req is None:

            req = self._orig_cnx[0].request()

        user = req.create_entity('CWUser', login=unicode(login),

                                 upassword=password, **kwargs)

        req.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'

                    % ','.join(repr(str(g)) for g in groups),

                    {'x': user.eid})

        user.cw_clear_relation_cache('in_group', 'subject')

        if commit:

            req.cnx.commit()

        return user

    def login(self, login, **kwargs):

        """return a connection for the given login/password"""

        if login == self.admlogin:

            self.restore_connection()

            # definitly don't want autoclose when used as a context manager

            return self.cnx

        autoclose = kwargs.pop('autoclose', True)

        if not kwargs:

            kwargs['password'] = str(login)

        self.set_cnx(repo_connect(self.repo, unicode(login), **kwargs))

        self.websession = DBAPISession(self.cnx)

        if login == self.vreg.config.anonymous_user()[0]:

            self.cnx.anonymous_connection = True

        if autoclose:

            return TestCaseConnectionProxy(self, self.cnx)