1 # copyright 2003-2015 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

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

     3 #

     4 # This file is part of CubicWeb.

     5 #

     6 # CubicWeb is free software: you can redistribute it and/or modify it under the

     7 # terms of the GNU Lesser General Public License as published by the Free

     8 # Software Foundation, either version 2.1 of the License, or (at your option)

     9 # any later version.

    10 #

    11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT

    12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

    13 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

    14 # details.

    15 #

    16 # You should have received a copy of the GNU Lesser General Public License along

    17 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

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

    19 """

    20 

    21 __docformat__ = "restructuredtext en"

    22 

    23 from contextlib import contextmanager

    24 

    25 from six import string_types

    26 

    27 from logilab.database import get_db_helper

    28 

    29 from cubicweb.req import RequestSessionBase

    30 from cubicweb.cwvreg import CWRegistryStore

    31 from cubicweb.web.request import ConnectionCubicWebRequestBase

    32 

    33 from cubicweb.devtools import BASE_URL, BaseApptestConfiguration

    34 

    35 

    36 class FakeConfig(dict, BaseApptestConfiguration):

    37     translations = {}

    38     uiprops = {}

    39     https_uiprops = {}

    40     apphome = None

    41     debugmode = False

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

    43         self.appid = appid

    44         self.apphome = apphome

    45         self._cubes = cubes

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

    47         self['uid'] = None

    48         self['base-url'] = BASE_URL

    49         self['rql-cache-size'] = 3000

    50         self.datadir_url = BASE_URL + 'data/'

    51         self.https_datadir_url = (BASE_URL + 'data/').replace('http://', 'https://')

    52 

    53     def cubes(self, expand=False):

    54         return self._cubes

    55 

    56     def sources(self):

    57         return {'system': {'db-driver': 'sqlite'}}

    58 

    59 

    60 class FakeRequest(ConnectionCubicWebRequestBase):

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

    62 

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

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

    65             kwargs['vreg'] = CWRegistryStore(FakeConfig(), initlog=False)

    66         kwargs['https'] = False

    67         self._http_method = kwargs.pop('method', 'GET')

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

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

    70         self._session_data = {}

    71 

    72     def set_cookie(self, name, value, maxage=300, expires=None, secure=False, httponly=False):

    73         super(FakeRequest, self).set_cookie(name, value, maxage, expires, secure, httponly)

    74         cookie = self.get_response_header('Set-Cookie')

    75         self._headers_in.setHeader('Cookie', cookie)

    76 

    77     ## Implement request abstract API

    78     def http_method(self):

    79         return self._http_method

    80 

    81     def relative_path(self, includeparams=True):

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

    83         to the instance's root, but some other normalization may be needed

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

    85         """

    86         if self._url.startswith(BASE_URL):

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

    88         else:

    89             url = self._url

    90         if includeparams:

    91             return url

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

    93 

    94     def set_request_header(self, header, value, raw=False):

    95         """set an incoming HTTP header (for test purpose only)"""

    96         if isinstance(value, string_types):

    97             value = [value]

    98         if raw:

    99             # adding encoded header is important, else page content

   100             # will be reconverted back to unicode and apart unefficiency, this

   101             # may cause decoding problem (e.g. when downloading a file)

   102             self._headers_in.setRawHeaders(header, value)

   103         else:

   104             self._headers_in.setHeader(header, value) #

   105 

   106     def get_response_header(self, header, default=None, raw=False):

   107         """return output header (for test purpose only)"""

   108         if raw:

   109             return self.headers_out.getRawHeaders(header, [default])[0]

   110         return self.headers_out.getHeader(header, default)

   111 

   112     def build_url_params(self, **kwargs):

   113         # overriden to get predictable resultts

   114         args = []

   115         for param, values in sorted(kwargs.items()):

   116             if not isinstance(values, (list, tuple)):

   117                 values = (values,)

   118             for value in values:

   119                 assert value is not None

   120                 args.append(u'%s=%s' % (param, self.url_quote(value)))

   121         return '&'.join(args)

   122 

   123 class FakeUser(object):

   124     login = 'toto'

   125     eid = 0

   126     def in_groups(self, groups):

   127         return True

   128 

   129 

   130 class FakeSession(RequestSessionBase):

   131 

   132     def __init__(self, repo=None, user=None, vreg=None):

   133         self.repo = repo

   134         if vreg is None:

   135             vreg = getattr(self.repo, 'vreg', None)

   136         if vreg is None:

   137             vreg = CWRegistryStore(FakeConfig(), initlog=False)

   138         self.vreg = vreg

   139         self.cnxset = FakeConnectionsSet()

   140         self.user = user or FakeUser()

   141         self.is_internal_session = False

   142         self.transaction_data = {}

   143 

   144     def execute(self, *args, **kwargs):

   145         pass

   146 

   147     def commit(self, *args):

   148         self.transaction_data.clear()

   149     def close(self, *args):

   150         pass

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

   152         pass

   153 

   154     def set_entity_cache(self, entity):

   155         pass

   156 

   157     def security_enabled(self, read=False, write=False):

   158         class FakeCM(object):

   159             def __enter__(self): pass

   160             def __exit__(self, exctype, exc, traceback): pass

   161         return FakeCM()

   162 

   163     # for use with enabled_security context manager

   164     read_security = write_security = True

   165 

   166     @contextmanager

   167     def running_hooks_ops(self):

   168         yield

   169 

   170 class FakeRepo(object):

   171     querier = None

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

   173         self.extids = {}

   174         self.eids = {}

   175         self._count = 0

   176         self.schema = schema

   177         self.config = config or FakeConfig()

   178         self.vreg = vreg or CWRegistryStore(self.config, initlog=False)

   179         self.vreg.schema = schema

   180 

   181     def internal_session(self):

   182         return FakeSession(self)

   183 

   184     def extid2eid(self, source, extid, etype, cnx, insert=True):

   185         try:

   186             return self.extids[extid]

   187         except KeyError:

   188             if not insert:

   189                 return None

   190             self._count += 1

   191             eid = self._count

   192             entity = source.before_entity_insertion(cnx, extid, etype, eid)

   193             self.extids[extid] = eid

   194             self.eids[eid] = extid

   195             source.after_entity_insertion(cnx, extid, entity)

   196             return eid

   197 

   198 

   199 class FakeSource(object):

   200     dbhelper = get_db_helper('sqlite')

   201     def __init__(self, uri):

   202         self.uri = uri

   203 

   204 

   205 class FakeConnectionsSet(object):

   206     def source(self, uri):

   207         return FakeSource(uri)