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

:organization: Logilab

:copyright: 2001-2010 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.

:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr

:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses

"""

__docformat__ = "restructuredtext en"

import os

import sys

import re

from urllib import unquote

from math import log

import simplejson

import yams.schema

from logilab.common.testlib import TestCase, InnerTest

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

from cubicweb import ValidationError, NoSelectableObject, AuthenticationError

from cubicweb import cwconfig, devtools, web, server

from cubicweb.dbapi import repo_connect, ConnectionProperties, ProgrammingError

from cubicweb.sobjects import notification

from cubicweb.web import Redirect, application

from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS

from cubicweb.devtools import fake, htmlparser

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

        file('/tmp/toto.html', 'w').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

def refresh_repo(repo, resetschema=False, resetvreg=False):

    devtools.reset_test_database(repo.config)

    for pool in repo.pools:

        pool.reconnect()

    repo._type_source_cache = {}

    repo._extid_cache = {}

    repo.querier._rql_cache = {}

    for source in repo.sources:

        source.reset_caches()

    if resetschema:

        repo.set_schema(repo.config.load_schema(), resetvreg=resetvreg)

# 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

# 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

    """

    appid = 'data'

    configcls = devtools.ApptestConfiguration

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

    @classproperty

    def config(cls):

        """return the configuration object. Configuration is cached on the test

        class.

        """

        try:

            return cls.__dict__['_config']

        except KeyError:

            config = cls._config = cls.configcls(cls.appid)

            config.mode = 'test'

            return config

    @classmethod

    def init_config(cls, config):

        """configuration initialization hooks. You may want to override this."""

        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'])

        try:

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

            # AttributeError since getlogin not available under all platforms

        except (OSError, AttributeError):

            send_to =  '%s@logilab.fr' % (os.environ.get('USER')

                                          or os.environ.get('USERNAME')

                                          or os.environ.get('LOGNAME'))

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

        # web resources

        config.global_set_option('base-url', devtools.BASE_URL)

        try:

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

        except: # not in server only configuration

            pass

    @classmethod

    def _init_repo(cls):

        """init the repository and connection to it.

        Repository and connection are cached on the test class. Once

        initialized, we simply reset connections and repository caches.

        """

        if not 'repo' in cls.__dict__:

            cls._build_repo()

        else:

            cls.cnx.rollback()

            cls._refresh_repo()

    @classmethod

    def _build_repo(cls):

        cls.repo, cls.cnx = devtools.init_test_database(config=cls.config)

        cls.init_config(cls.config)

        cls.vreg = cls.repo.vreg

        cls._orig_cnx = cls.cnx

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

        # necessary for authentication tests

        cls.cnx.login = cls.admlogin

        cls.cnx.authinfo = {'password': cls.admpassword}

    @classmethod

    def _refresh_repo(cls):

        refresh_repo(cls.repo, cls.reset_schema, cls.reset_vreg)

    # 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)"""

        return self.repo._sessions[self.cnx.sessionid]

    @property

    def adminsession(self):

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

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

    def set_option(self, optname, value):

        self.config.global_set_option(optname, value)

    def set_debug(self, debugmode):

        server.set_debug(debugmode)

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

    def setUp(self):

        pause_tracing()

        self._init_repo()

        resume_tracing()

        self.setup_database()

        self.commit()

        MAILBOX[:] = [] # reset mailbox

    def setup_database(self):

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

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

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

        if password is None:

            password = login.encode('utf8')

        cursor = self._orig_cnx.cursor(req or self.request())

        rset = cursor.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s',

                              {'login': unicode(login), 'passwd': password})

        user = rset.get_entity(0, 0)

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

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

                       {'x': user.eid}, 'x')

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

        if commit:

            self._orig_cnx.commit()

        return user

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

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

        if login == self.admlogin:

            self.restore_connection()

        else:

            if not kwargs:

                kwargs['password'] = str(login)

            self.cnx = repo_connect(self.repo, unicode(login),

                                    cnxprops=ConnectionProperties('inmemory'),

                                    **kwargs)

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

            self.cnx.anonymous_connection = True

        return self.cnx

    def restore_connection(self):

        if not self.cnx is self._orig_cnx:

            try:

                self.cnx.close()

            except ProgrammingError:

                pass # already closed

        self.cnx = self._orig_cnx

    # db api ##################################################################

    @nocoverage

    def cursor(self, req=None):

        return self.cnx.cursor(req or self.request())

    @nocoverage

    def execute(self, rql, args=None, eidkey=None, req=None):

        """executes <rql>, builds a resultset, and returns a couple (rset, req)

        where req is a FakeRequest

        """

        req = req or self.request(rql=rql)

        return self.cnx.cursor(req).execute(unicode(rql), args, eidkey)

    @nocoverage

    def commit(self):

        self.cnx.commit()

    @nocoverage

    def rollback(self):

        try:

            self.cnx.rollback()

        except ProgrammingError:

            pass

    # # server side db api #######################################################

    def sexecute(self, rql, args=None, eid_key=None):

        self.session.set_pool()

        return self.session.execute(rql, args, eid_key)

    # other utilities #########################################################

    def entity(self, rql, args=None, eidkey=None, req=None):

        return self.execute(rql, args, eidkey, req=req).get_entity(0, 0)

    # vregistry inspection utilities ###########################################

    def pviews(self, req, rset):

        return sorted((a.__regid__, a.__class__)

                      for a in self.vreg['views'].possible_views(req, rset=rset))

    def pactions(self, req, rset,

                 skipcategories=('addrelated', 'siteactions', 'useractions', 'footer')):

        return [(a.__regid__, a.__class__)

                for a in self.vreg['actions'].poss_visible_objects(req, rset=rset)

                if a.category not in skipcategories]

    def pactions_by_cats(self, req, rset, categories=('addrelated',)):

        return [(a.__regid__, a.__class__)

                for a in self.vreg['actions'].poss_visible_objects(req, rset=rset)

                if a.category in categories]

    def pactionsdict(self, req, rset,

                     skipcategories=('addrelated', 'siteactions', 'useractions', 'footer')):

        res = {}

        for a in self.vreg['actions'].poss_visible_objects(req, rset=rset):

            if a.category not in skipcategories:

                res.setdefault(a.category, []).append(a.__class__)

        return res

    def action_submenu(self, req, rset, id):

        return self._test_action(self.vreg['actions'].select(id, req, rset=rset))

    def _test_action(self, action):

        class fake_menu(list):

            @property

            def items(self):

                return self

        class fake_box(object):

            def mk_action(self, label, url, **kwargs):

                return (label, url)

            def box_action(self, action, **kwargs):

                return (action.title, action.url())

        submenu = fake_menu()

        action.fill_menu(fake_box(), submenu)