1 """Fake objects to ease testing of cubicweb without a fully working environment

     2 

     3 :organization: Logilab

     4 :copyright: 2001-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

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

     6 """

     7 __docformat__ = "restructuredtext en"

     8 

     9 from logilab.common.testlib import mock_object as Mock

    10 from logilab.common.adbh import get_adv_func_helper

    11 

    12 from indexer import get_indexer

    13 

    14 from cubicweb import RequestSessionMixIn

    15 from cubicweb.web.request import CubicWebRequestBase

    16 from cubicweb.devtools import BASE_URL, BaseApptestConfiguration

    17 

    18 

    19 class FakeConfig(dict, BaseApptestConfiguration):

    20     translations = {}

    21     apphome = None

    22     def __init__(self, appid='data', apphome=None, cubes=()):

    23         self.appid = appid

    24         self.apphome = apphome

    25         self._cubes = cubes

    26         self['auth-mode'] = 'cookie'

    27         self['uid'] = None 

    28         self['base-url'] = BASE_URL

    29         self['rql-cache-size'] = 100

    30        

    31     def cubes(self, expand=False):

    32         return self._cubes

    33     

    34     def sources(self):

    35         return {}

    36 

    37 class FakeVReg(object):

    38     def __init__(self, schema=None, config=None):

    39         self.schema = schema

    40         self.config = config or FakeConfig()

    41         self.properties = {'ui.encoding': 'UTF8',

    42                            'ui.language': 'en',

    43                            }

    44         

    45     def property_value(self, key):

    46         return self.properties[key]

    47 

    48     _registries = {

    49         'controllers' : [Mock(id='view'), Mock(id='login'),

    50                          Mock(id='logout'), Mock(id='edit')],

    51         'views' : [Mock(id='primary'), Mock(id='secondary'),

    52                          Mock(id='oneline'), Mock(id='list')],

    53         }

    54     

    55     def registry_objects(self, name, oid=None):

    56         return self._registries[name]

    57     

    58     def etype_class(self, etype):

    59         class Entity(dict):

    60             e_schema = self.schema[etype]

    61             def __init__(self, session, eid, row=0, col=0):

    62                 self.req = session

    63                 self.eid = eid

    64                 self.row, self.col = row, col

    65             def set_eid(self, eid):

    66                 self.eid = self['eid'] = eid

    67         return Entity

    68 

    69 

    70 class FakeRequest(CubicWebRequestBase):

    71     """test implementation of an cubicweb request object"""

    72 

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

    74         if not (args or 'vreg' in kwargs):

    75             kwargs['vreg'] = FakeVReg()

    76         kwargs['https'] = False

    77         self._url = kwargs.pop('url', 'view?rql=Blop&vid=blop')

    78         super(FakeRequest, self).__init__(*args, **kwargs)

    79         self._session_data = {}

    80         self._headers = {}

    81 

    82     def header_accept_language(self):

    83         """returns an ordered list of preferred languages"""

    84         return ('en',)

    85 

    86     def header_if_modified_since(self):

    87         return None

    88 

    89     def base_url(self):

    90         """return the root url of the application"""

    91         return BASE_URL

    92 

    93     def relative_path(self, includeparams=True):

    94         """return the normalized path of the request (ie at least relative

    95         to the application's root, but some other normalization may be needed

    96         so that the returned path may be used to compare to generated urls

    97         """

    98         if self._url.startswith(BASE_URL):

    99             url = self._url[len(BASE_URL):]

   100         else:

   101             url = self._url

   102         if includeparams:

   103             return url

   104         return url.split('?', 1)[0]

   105 

   106     def set_content_type(self, content_type, filename=None, encoding=None):

   107         """set output content type for this request. An optional filename

   108         may be given

   109         """

   110         pass

   111 

   112     def set_header(self, header, value):

   113         """set an output HTTP header"""

   114         pass

   115     

   116     def add_header(self, header, value):

   117         """set an output HTTP header"""

   118         pass

   119     

   120     def remove_header(self, header):

   121         """remove an output HTTP header"""

   122         pass

   123     

   124     def get_header(self, header, default=None):

   125         """return the value associated with the given input header,

   126         raise KeyError if the header is not set

   127         """

   128         return self._headers.get(header, default)

   129 

   130     def set_cookie(self, cookie, key, maxage=300):

   131         """set / update a cookie key

   132 

   133         by default, cookie will be available for the next 5 minutes

   134         """

   135         pass

   136 

   137     def remove_cookie(self, cookie, key):

   138         """remove a cookie by expiring it"""

   139         pass

   140 

   141     def validate_cache(self):

   142         pass

   143 

   144     # session compatibility (in some test are using this class to test server

   145     # side views...)

   146     def actual_session(self):

   147         """return the original parent session if any, else self"""

   148         return self

   149 

   150     def unsafe_execute(self, *args, **kwargs):

   151         """return the original parent session if any, else self"""

   152         kwargs.pop('propagate', None)

   153         return self.execute(*args, **kwargs)

   154 

   155 

   156 class FakeUser(object):

   157     login = 'toto'

   158     eid = 0

   159     def in_groups(self, groups):

   160         return True

   161 

   162 

   163 class FakeSession(RequestSessionMixIn):

   164     def __init__(self, repo=None, user=None):

   165         self.repo = repo

   166         self.vreg = getattr(self.repo, 'vreg', FakeVReg())

   167         self.pool = FakePool()

   168         self.user = user or FakeUser()

   169         self.is_internal_session = False

   170         self.is_super_session = self.user.eid == -1

   171         self._query_data = {}

   172         

   173     def execute(self, *args):

   174         pass

   175     def commit(self, *args):

   176         self._query_data.clear()

   177     def close(self, *args):

   178         pass

   179     def system_sql(self, sql, args=None):

   180         pass

   181 

   182     def decorate_rset(self, rset, propagate=False):

   183         rset.vreg = self.vreg

   184         rset.req = self

   185         return rset

   186 

   187     def set_entity_cache(self, entity):

   188         pass

   189     

   190 class FakeRepo(object):

   191     querier = None

   192     def __init__(self, schema, vreg=None, config=None):

   193         self.extids = {}

   194         self.eids = {}

   195         self._count = 0

   196         self.schema = schema

   197         self.vreg = vreg or FakeVReg()

   198         self.config = config or FakeConfig()

   199 

   200     def internal_session(self):

   201         return FakeSession(self)

   202     

   203     def extid2eid(self, source, extid, etype, session, insert=True):

   204         try:

   205             return self.extids[extid]

   206         except KeyError:

   207             if not insert:

   208                 return None

   209             self._count += 1

   210             eid = self._count

   211             entity = source.before_entity_insertion(session, extid, etype, eid)

   212             self.extids[extid] = eid

   213             self.eids[eid] = extid

   214             source.after_entity_insertion(session, extid, entity)

   215             return eid

   216         

   217     def eid2extid(self, source, eid, session=None):

   218         return self.eids[eid]

   219 

   220 

   221 class FakeSource(object):

   222     dbhelper = get_adv_func_helper('sqlite')

   223     indexer = get_indexer('sqlite', 'UTF8')

   224     dbhelper.fti_uid_attr = indexer.uid_attr

   225     dbhelper.fti_table = indexer.table

   226     dbhelper.fti_restriction_sql = indexer.restriction_sql

   227     dbhelper.fti_need_distinct_query = indexer.need_distinct

   228     def __init__(self, uri):

   229         self.uri = uri

   230 

   231         

   232 class FakePool(object):

   233     def source(self, uri):

   234         return FakeSource(uri)

   235 

   236 # commented until proven to be useful

   237 ## from logging import getLogger

   238 ## from cubicweb import set_log_methods

   239 ## for cls in (FakeConfig, FakeVReg, FakeRequest, FakeSession, FakeRepo,

   240 ##             FakeSource, FakePool):

   241 ##     set_log_methods(cls, getLogger('fake'))