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

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

from __future__ import print_function

__docformat__ = "restructuredtext en"

import sys

import re

from os.path import dirname, join, abspath

from math import log

from contextlib import contextmanager

from warnings import warn

from itertools import chain

from six import string_types

from six.moves import range

from six.moves.urllib.parse import urlparse, parse_qs, unquote as urlunquote

import yams.schema

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

from logilab.common.pytest import nocoverage, pause_trace

from logilab.common.debugger import Debugger

from logilab.common.umessage import message_from_string

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

from logilab.common.deprecation import deprecated, class_deprecated

from logilab.common.shellutils import getlogin

from cubicweb import (ValidationError, NoSelectableObject, AuthenticationError,

                      ProgrammingError, BadConnectionId)

from cubicweb import cwconfig, devtools, web, server, repoapi

from cubicweb.utils import json

from cubicweb.sobjects import notification

from cubicweb.web import Redirect, application, eid_param

from cubicweb.server.hook import SendMailOp

from cubicweb.server.session import Session

from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS

from cubicweb.devtools import 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 open('/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

class JsonValidator(object):

    def parse_string(self, data):

        return json.loads(data)

@contextmanager

def real_error_handling(app):

    """By default, CubicWebTC `app` attribute (ie the publisher) is monkey

    patched so that unexpected error are raised rather than going through the

    `error_handler` method.

    By using this context manager you disable this monkey-patching temporarily.

    Hence when publishihng a request no error will be raised, you'll get

    req.status_out set to an HTTP error status code and the generated page will

    usually hold a traceback as HTML.

    >>> with real_error_handling(app):

    >>>     page = app.handle_request(req)

    """

    # remove the monkey patched error handler

    fake_error_handler = app.error_handler

    del app.error_handler

    # return the app

    yield app

    # restore

    app.error_handler = fake_error_handler

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

MAILBOX = []

class Email(object):

    """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, fromaddr, recipients, msg):

        self.fromaddr = fromaddr

        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, fromaddr, recipients, msg):

        MAILBOX.append(Email(fromaddr, recipients, msg))

cwconfig.SMTP = MockSMTP

# Repoaccess utility ###############################################3###########

class RepoAccess(object):

    """An helper to easily create object to access the repo as a specific user

    Each RepoAccess have it own session.

    A repo access can create three type of object:

    .. automethod:: cubicweb.testlib.RepoAccess.cnx

    .. automethod:: cubicweb.testlib.RepoAccess.web_request

    The RepoAccess need to be closed to destroy the associated Session.

    TestCase usually take care of this aspect for the user.

    .. automethod:: cubicweb.testlib.RepoAccess.close

    """

    def __init__(self, repo, login, requestcls):

        self._repo = repo

        self._login = login

        self.requestcls = requestcls

        self._session = self._unsafe_connect(login)

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

        """ a completely unsafe connect method for the tests """

        # use an internal connection

        with self._repo.internal_cnx() as cnx:

            # try to get a user object

            user = cnx.find('CWUser', login=login).one()

            user.groups

            user.properties

            user.login

            session = Session(user, self._repo)

            self._repo._sessions[session.sessionid] = session

            user._cw = user.cw_rset.req = session

        with session.new_cnx() as cnx:

            self._repo.hm.call_hooks('session_open', cnx)

            # commit connection at this point in case write operation has been

            # done during `session_open` hooks

            cnx.commit()

        return session

    @contextmanager

    def cnx(self):

        """Context manager returning a server side connection for the user"""

        with self._session.new_cnx() as cnx:

            yield cnx

    # aliases for bw compat

    client_cnx = repo_cnx = cnx

    @contextmanager

    def web_request(self, url=None, headers={}, method='GET', **kwargs):

        """Context manager returning a web request pre-linked to a client cnx

        To commit and rollback use::

            req.cnx.commit()

            req.cnx.rolback()

        """

        req = self.requestcls(self._repo.vreg, url=url, headers=headers,

                              method=method, form=kwargs)

        with self._session.new_cnx() as cnx:

            if 'ecache' in cnx.transaction_data:

                del cnx.transaction_data['ecache']

            req.set_cnx(cnx)

            yield req

    def close(self):

        """Close the session associated to the RepoAccess"""

        if self._session is not None:

            self._repo.close(self._session.sessionid)

        self._session = None

    @contextmanager

    def shell(self):

        from cubicweb.server.migractions import ServerMigrationHelper

        with self._session.new_cnx() as cnx:

            mih = ServerMigrationHelper(None, repo=self._repo, cnx=cnx,

                                        interactive=False,

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

                                        schema=1)

            yield mih

            cnx.commit()

# 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`, repoapi 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

    requestcls = fake.FakeRequest

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

    test_db_id = DEFAULT_EMPTY_DB_ID

    # anonymous is logged by default in cubicweb test cases

    anonymous_allowed = True

    def __init__(self, *args, **kwargs):

        self._admin_session = None

        self.repo = None

        self._open_access = set()

        super(CubicWebTC, self).__init__(*args, **kwargs)

    # repository connection handling ###########################################

    def new_access(self, login):

        """provide a new RepoAccess object for a given user

        The access is automatically closed at the end of the test."""

        login = unicode(login)

        access = RepoAccess(self.repo, login, self.requestcls)

        self._open_access.add(access)

        return access

    def _close_access(self):

        while self._open_access:

            try:

                self._open_access.pop().close()

            except BadConnectionId:

                continue # already closed

    @property

    def session(self):

        """return admin session"""

        return self._admin_session

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

    def _init_repo(self):

        """init the repository and connection to it.

        """

        # get or restore and working db.

        db_handler = devtools.get_test_db_handler(self.config, self.init_config)

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

        db_handler.restore_database(self.test_db_id)

        self.repo = db_handler.get_repo(startup=True)

        # get an admin session (without actual login)

        login = unicode(db_handler.config.default_admin_config['login'])

        self.admin_access = self.new_access(login)

        self._admin_session = self.admin_access._session

    # config management ########################################################

    @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 # XXX could be turned into a regular method

    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.

        This method will be called by the database handler once the config has

        been properly bootstrapped.

        """

        admincfg = config.default_admin_config

        cls.admlogin = unicode(admincfg['login'])

        cls.admpassword = admincfg['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 Exception: # not in server only configuration

            pass

        config.set_anonymous_allowed(cls.anonymous_allowed)

    @property

    def vreg(self):

        return self.repo.vreg

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

    @property

    def schema(self):

        """return the application schema"""

        return self.vreg.schema

    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):