+
−"""Fake objects to ease testing of cubicweb without a fully working environment
+
−
+
−:organization: Logilab
+
−:copyright: 2001-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+
−:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
+
−"""
+
−__docformat__ = "restructuredtext en"
+
−
+
−from logilab.common.testlib import mock_object as Mock
+
−from logilab.common.adbh import get_adv_func_helper
+
−
+
−from indexer import get_indexer
+
−
+
−from cubicweb import RequestSessionMixIn
+
−from cubicweb.web.request import CubicWebRequestBase
+
−from cubicweb.devtools import BASE_URL, BaseApptestConfiguration
+
−
+
−
+
−class FakeConfig(dict, BaseApptestConfiguration):
+
−    translations = {}
+
−    apphome = None
+
−    def __init__(self, appid='data', apphome=None, cubes=()):
+
−        self.appid = appid
+
−        self.apphome = apphome
+
−        self._cubes = cubes
+
−        self['auth-mode'] = 'cookie'
+
−        self['uid'] = None 
+
−        self['base-url'] = BASE_URL
+
−        self['rql-cache-size'] = 100
+
−       
+
−    def cubes(self, expand=False):
+
−        return self._cubes
+
−    
+
−    def sources(self):
+
−        return {}
+
−
+
−class FakeVReg(object):
+
−    def __init__(self, schema=None, config=None):
+
−        self.schema = schema
+
−        self.config = config or FakeConfig()
+
−        self.properties = {'ui.encoding': 'UTF8',
+
−                           'ui.language': 'en',
+
−                           }
+
−        
+
−    def property_value(self, key):
+
−        return self.properties[key]
+
−
+
−    _registries = {
+
−        'controllers' : [Mock(id='view'), Mock(id='login'),
+
−                         Mock(id='logout'), Mock(id='edit')],
+
−        'views' : [Mock(id='primary'), Mock(id='secondary'),
+
−                         Mock(id='oneline'), Mock(id='list')],
+
−        }
+
−    
+
−    def registry_objects(self, name, oid=None):
+
−        return self._registries[name]
+
−    
+
−    def etype_class(self, etype):
+
−        class Entity(dict):
+
−            e_schema = self.schema[etype]
+
−            def __init__(self, session, eid, row=0, col=0):
+
−                self.req = session
+
−                self.eid = eid
+
−                self.row, self.col = row, col
+
−            def set_eid(self, eid):
+
−                self.eid = self['eid'] = eid
+
−        return Entity
+
−
+
−
+
−class FakeRequest(CubicWebRequestBase):
+
−    """test implementation of an cubicweb request object"""
+
−
+
−    def __init__(self, *args, **kwargs):
+
−        if not (args or 'vreg' in kwargs):
+
−            kwargs['vreg'] = FakeVReg()
+
−        kwargs['https'] = False
+
−        self._url = kwargs.pop('url', 'view?rql=Blop&vid=blop')
+
−        super(FakeRequest, self).__init__(*args, **kwargs)
+
−        self._session_data = {}
+
−        self._headers = {}
+
−
+
−    def header_accept_language(self):
+
−        """returns an ordered list of preferred languages"""
+
−        return ('en',)
+
−
+
−    def header_if_modified_since(self):
+
−        return None
+
−
+
−    def base_url(self):
+
−        """return the root url of the application"""
+
−        return BASE_URL
+
−
+
−    def relative_path(self, includeparams=True):
+
−        """return the normalized path of the request (ie at least relative
+
−        to the application's root, but some other normalization may be needed
+
−        so that the returned path may be used to compare to generated urls
+
−        """
+
−        if self._url.startswith(BASE_URL):
+
−            url = self._url[len(BASE_URL):]
+
−        else:
+
−            url = self._url
+
−        if includeparams:
+
−            return url
+
−        return url.split('?', 1)[0]
+
−
+
−    def set_content_type(self, content_type, filename=None, encoding=None):
+
−        """set output content type for this request. An optional filename
+
−        may be given
+
−        """
+
−        pass
+
−
+
−    def set_header(self, header, value):
+
−        """set an output HTTP header"""
+
−        pass
+
−    
+
−    def add_header(self, header, value):
+
−        """set an output HTTP header"""
+
−        pass
+
−    
+
−    def remove_header(self, header):
+
−        """remove an output HTTP header"""
+
−        pass
+
−    
+
−    def get_header(self, header, default=None):
+
−        """return the value associated with the given input header,
+
−        raise KeyError if the header is not set
+
−        """
+
−        return self._headers.get(header, default)
+
−
+
−    def set_cookie(self, cookie, key, maxage=300):
+
−        """set / update a cookie key
+
−
+
−        by default, cookie will be available for the next 5 minutes
+
−        """
+
−        pass
+
−
+
−    def remove_cookie(self, cookie, key):
+
−        """remove a cookie by expiring it"""
+
−        pass
+
−
+
−    def validate_cache(self):
+
−        pass
+
−
+
−    # session compatibility (in some test are using this class to test server
+
−    # side views...)
+
−    def actual_session(self):
+
−        """return the original parent session if any, else self"""
+
−        return self
+
−
+
−    def unsafe_execute(self, *args, **kwargs):
+
−        """return the original parent session if any, else self"""
+
−        kwargs.pop('propagate', None)
+
−        return self.execute(*args, **kwargs)
+
−
+
−
+
−class FakeUser(object):
+
−    login = 'toto'
+
−    eid = 0
+
−    def in_groups(self, groups):
+
−        return True
+
−
+
−
+
−class FakeSession(RequestSessionMixIn):
+
−    def __init__(self, repo=None, user=None):
+
−        self.repo = repo
+
−        self.vreg = getattr(self.repo, 'vreg', FakeVReg())
+
−        self.pool = FakePool()
+
−        self.user = user or FakeUser()
+
−        self.is_internal_session = False
+
−        self.is_super_session = self.user.eid == -1
+
−        self._query_data = {}
+
−        
+
−    def execute(self, *args):
+
−        pass
+
−    def commit(self, *args):
+
−        self._query_data.clear()
+
−    def close(self, *args):
+
−        pass
+
−    def system_sql(self, sql, args=None):
+
−        pass
+
−
+
−    def decorate_rset(self, rset, propagate=False):
+
−        rset.vreg = self.vreg
+
−        rset.req = self
+
−        return rset
+
−
+
−    def set_entity_cache(self, entity):
+
−        pass
+
−    
+
−class FakeRepo(object):
+
−    querier = None
+
−    def __init__(self, schema, vreg=None, config=None):
+
−        self.extids = {}
+
−        self.eids = {}
+
−        self._count = 0
+
−        self.schema = schema
+
−        self.vreg = vreg or FakeVReg()
+
−        self.config = config or FakeConfig()
+
−
+
−    def internal_session(self):
+
−        return FakeSession(self)
+
−    
+
−    def extid2eid(self, source, extid, etype, session, insert=True):
+
−        try:
+
−            return self.extids[extid]
+
−        except KeyError:
+
−            if not insert:
+
−                return None
+
−            self._count += 1
+
−            eid = self._count
+
−            entity = source.before_entity_insertion(session, extid, etype, eid)
+
−            self.extids[extid] = eid
+
−            self.eids[eid] = extid
+
−            source.after_entity_insertion(session, extid, entity)
+
−            return eid
+
−        
+
−    def eid2extid(self, source, eid, session=None):
+
−        return self.eids[eid]
+
−
+
−
+
−class FakeSource(object):
+
−    dbhelper = get_adv_func_helper('sqlite')
+
−    indexer = get_indexer('sqlite', 'UTF8')
+
−    dbhelper.fti_uid_attr = indexer.uid_attr
+
−    dbhelper.fti_table = indexer.table
+
−    dbhelper.fti_restriction_sql = indexer.restriction_sql
+
−    dbhelper.fti_need_distinct_query = indexer.need_distinct
+
−    def __init__(self, uri):
+
−        self.uri = uri
+
−
+
−        
+
−class FakePool(object):
+
−    def source(self, uri):
+
−        return FakeSource(uri)
+
−
+
−# commented until proven to be useful
+
−## from logging import getLogger
+
−## from cubicweb import set_log_methods
+
−## for cls in (FakeConfig, FakeVReg, FakeRequest, FakeSession, FakeRepo,
+
−##             FakeSource, FakePool):
+
−##     set_log_methods(cls, getLogger('fake'))