# copyright 2003-2012 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/>.
"""cubicweb.cwconfig unit tests"""
import contextlib
import functools
import sys
import os
import pkgutil
from os.path import dirname, join
from pkg_resources import EntryPoint, Distribution
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
from logilab.common.modutils import cleanup_sys_modules
from cubicweb.devtools import ApptestConfiguration
from cubicweb.devtools.testlib import BaseTestCase
from cubicweb.cwconfig import (
CubicWebConfiguration, _expand_modname)
def templibdir(func):
"""create a temporary directory and insert it in sys.path"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
with TemporaryDirectory() as libdir:
sys.path.insert(0, libdir)
try:
args = args + (libdir,)
return func(*args, **kwargs)
finally:
sys.path.remove(libdir)
return wrapper
def create_filepath(filepath):
filedir = dirname(filepath)
if not os.path.exists(filedir):
os.makedirs(filedir)
with open(filepath, 'a'):
pass
@contextlib.contextmanager
def temp_config(appid, instance_dir, cubes):
"""context manager that create a config object with specified appid,
instance_dir and cubes"""
cls = CubicWebConfiguration
old = (cls._INSTANCES_DIR,
sys.path[:], sys.meta_path[:])
old_modules = set(sys.modules)
try:
cls._INSTANCES_DIR = instance_dir
config = cls(appid)
config._cubes = cubes
config.adjust_sys_path()
yield config
finally:
(cls._INSTANCES_DIR, sys.path[:], sys.meta_path[:]) = old
for module in set(sys.modules) - old_modules:
del sys.modules[module]
def iter_entry_points(group, name):
"""Mock pkg_resources.iter_entry_points to yield EntryPoint from
packages found in test/data even though these are not
installed.
"""
libpython = CubicWebConfigurationTC.datapath()
prefix = 'cubicweb_'
for pkgname in os.listdir(libpython):
if not pkgname.startswith(prefix):
continue
location = join(libpython, pkgname)
yield EntryPoint(pkgname[len(prefix):], pkgname,
dist=Distribution(location))
class CubicWebConfigurationTC(BaseTestCase):
@classmethod
def setUpClass(cls):
sys.path.append(cls.datapath())
@classmethod
def tearDownClass(cls):
sys.path.remove(cls.datapath())
def setUp(self):
self.config = ApptestConfiguration('data', __file__)
self.config._cubes = ('email', 'file')
def tearDown(self):
cleanup_sys_modules([self.datapath()])
def test_migration_scripts_dir(self):
mscripts = os.listdir(self.config.migration_scripts_dir())
self.assertIn('bootstrapmigration_repository.py', mscripts)
self.assertIn('postcreate.py', mscripts)
self.assertIn('3.24.0_Any.py', mscripts)
@patch('pkg_resources.iter_entry_points', side_effect=iter_entry_points)
def test_available_cubes(self, mock_iter_entry_points):
expected_cubes = [
'cubicweb_card',
'cubicweb_comment',
'cubicweb_email',
'cubicweb_file',
'cubicweb_forge',
'cubicweb_localperms',
'cubicweb_mycube',
'cubicweb_tag',
]
self.assertEqual(self.config.available_cubes(), expected_cubes)
mock_iter_entry_points.assert_called_once_with(
group='cubicweb.cubes', name=None)
def test_reorder_cubes(self):
# forge depends on email and file and comment
# email depends on file
self.assertEqual(self.config.reorder_cubes(['file', 'email', 'forge']),
('forge', 'email', 'file'))
self.assertEqual(self.config.reorder_cubes(['email', 'file', 'forge']),
('forge', 'email', 'file'))
self.assertEqual(self.config.reorder_cubes(['email', 'forge', 'file']),
('forge', 'email', 'file'))
self.assertEqual(self.config.reorder_cubes(['file', 'forge', 'email']),
('forge', 'email', 'file'))
self.assertEqual(self.config.reorder_cubes(['forge', 'file', 'email']),
('forge', 'email', 'file'))
self.assertEqual(self.config.reorder_cubes(('forge', 'email', 'file')),
('forge', 'email', 'file'))
def test_reorder_cubes_recommends(self):
from cubicweb_comment import __pkginfo__ as comment_pkginfo
self._test_reorder_cubes_recommends(comment_pkginfo)
def _test_reorder_cubes_recommends(self, comment_pkginfo):
comment_pkginfo.__recommends_cubes__ = {'file': None}
try:
# email recommends comment
# comment recommends file
self.assertEqual(self.config.reorder_cubes(('forge', 'email', 'file', 'comment')),
('forge', 'email', 'comment', 'file'))
self.assertEqual(self.config.reorder_cubes(('forge', 'email', 'comment', 'file')),
('forge', 'email', 'comment', 'file'))
self.assertEqual(self.config.reorder_cubes(('forge', 'comment', 'email', 'file')),
('forge', 'email', 'comment', 'file'))
self.assertEqual(self.config.reorder_cubes(('comment', 'forge', 'email', 'file')),
('forge', 'email', 'comment', 'file'))
finally:
comment_pkginfo.__recommends_cubes__ = {}
def test_expand_cubes(self):
self.assertEqual(self.config.expand_cubes(('email', 'comment')),
['email', 'comment', 'file'])
def test_init_cubes_ignore_pyramid_cube(self):
warning_msg = 'cubicweb-pyramid got integrated into CubicWeb'
with self.assertLogs('cubicweb.configuration', level='WARNING') as cm:
self.config.init_cubes(['pyramid', 'card'])
self.assertIn(warning_msg, cm.output[0])
self.assertNotIn('pyramid', self.config._cubes)
@patch('pkg_resources.iter_entry_points', side_effect=iter_entry_points)
def test_ccplugin_modname(self, mock_iter_entry_points):
self.config.load_cwctl_plugins()
mock_iter_entry_points.assert_called_once_with(
group='cubicweb.cubes', name=None)
self.assertIn('cubicweb_mycube.ccplugin', sys.modules, sorted(sys.modules))
def test_config_value_from_environment_str(self):
self.assertIsNone(self.config['base-url'])
os.environ['CW_BASE_URL'] = 'https://www.cubicweb.org'
try:
self.assertEqual(self.config['base-url'],
'https://www.cubicweb.org')
finally:
del os.environ['CW_BASE_URL']
def test_config_value_from_environment_int(self):
self.assertEqual(self.config['connections-pool-size'], 4)
os.environ['CW_CONNECTIONS_POOL_SIZE'] = '6'
try:
self.assertEqual(self.config['connections-pool-size'], 6)
finally:
del os.environ['CW_CONNECTIONS_POOL_SIZE']
def test_config_value_from_environment_yn(self):
self.assertEqual(self.config['allow-email-login'], False)
try:
for val, result in (('yes', True), ('no', False),
('y', True), ('n', False),):
os.environ['CW_ALLOW_EMAIL_LOGIN'] = val
self.assertEqual(self.config['allow-email-login'], result)
finally:
del os.environ['CW_ALLOW_EMAIL_LOGIN']
class ModnamesTC(unittest.TestCase):
@templibdir
def test_expand_modnames(self, libdir):
tempdir = join(libdir, 'lib')
filepaths = [
join(tempdir, '__init__.py'),
join(tempdir, 'a.py'),
join(tempdir, 'b.py'),
join(tempdir, 'c.py'),
join(tempdir, 'b', '__init__.py'),
join(tempdir, 'b', 'a.py'),
join(tempdir, 'b', 'c.py'),
join(tempdir, 'b', 'd', '__init__.py'),
join(tempdir, 'e', 'e.py'),
]
for filepath in filepaths:
create_filepath(filepath)
# not importable
self.assertEqual(list(_expand_modname('isnotimportable')), [])
# not a python package
self.assertEqual(list(_expand_modname('lib.e')), [])
self.assertEqual(list(_expand_modname('lib.a')), [
('lib.a', join(tempdir, 'a.py')),
])
# lib.b.d should be imported
self.assertEqual(list(_expand_modname('lib.b')), [
('lib.b', join(tempdir, 'b', '__init__.py')),
('lib.b.a', join(tempdir, 'b', 'a.py')),
('lib.b.c', join(tempdir, 'b', 'c.py')),
('lib.b.d', join(tempdir, 'b', 'd', '__init__.py')),
])
# lib.b.d should not be imported without recursive mode
self.assertEqual(list(_expand_modname('lib.b', recursive=False)), [
('lib.b', join(tempdir, 'b', '__init__.py')),
('lib.b.a', join(tempdir, 'b', 'a.py')),
('lib.b.c', join(tempdir, 'b', 'c.py')),
])
self.assertEqual(list(_expand_modname('lib')), [
('lib', join(tempdir, '__init__.py')),
('lib.a', join(tempdir, 'a.py')),
('lib.b', join(tempdir, 'b', '__init__.py')),
('lib.b.a', join(tempdir, 'b', 'a.py')),
('lib.b.c', join(tempdir, 'b', 'c.py')),
('lib.b.d', join(tempdir, 'b', 'd', '__init__.py')),
('lib.c', join(tempdir, 'c.py')),
])
for source in (
join(tempdir, 'c.py'),
join(tempdir, 'b', 'c.py'),
join(tempdir, 'b', 'd', '__init__.py'),
):
# remove source file
os.remove(source)
self.assertEqual(list(_expand_modname('lib.c')), [])
self.assertEqual(list(_expand_modname('lib.b')), [
('lib.b', join(tempdir, 'b', '__init__.py')),
('lib.b.a', join(tempdir, 'b', 'a.py')),
])
self.assertEqual(list(_expand_modname('lib')), [
('lib', join(tempdir, '__init__.py')),
('lib.a', join(tempdir, 'a.py')),
('lib.b', join(tempdir, 'b', '__init__.py')),
('lib.b.a', join(tempdir, 'b', 'a.py')),
])
@templibdir
def test_schema_modnames(self, libdir):
for filepath in (
join(libdir, 'schema.py'),
join(libdir, 'cubicweb_foo', '__init__.py'),
join(libdir, 'cubicweb_foo', 'schema', '__init__.py'),
join(libdir, 'cubicweb_foo', 'schema', 'a.py'),
join(libdir, 'cubicweb_foo', 'schema', 'b.py'),
# subpackages should not be loaded
join(libdir, 'cubicweb_foo', 'schema', 'c', '__init__.py'),
join(libdir, '_instance_dir', 'data1', 'schema.py'),
join(libdir, '_instance_dir', 'data2', 'noschema.py'),
):
create_filepath(filepath)
expected = [
('cubicweb', 'cubicweb.schemas.bootstrap'),
('cubicweb', 'cubicweb.schemas.base'),
('cubicweb', 'cubicweb.schemas.workflow'),
('cubicweb', 'cubicweb.schemas.Bookmark'),
('foo', 'cubicweb_foo.schema'),
('foo', 'cubicweb_foo.schema.a'),
('foo', 'cubicweb_foo.schema.b'),
]
# app has schema file
instance_dir = join(libdir, '_instance_dir')
with temp_config('data1', instance_dir, ('foo',)) as config:
self.assertEqual(pkgutil.find_loader('schema').get_filename(),
join(libdir, '_instance_dir',
'data1', 'schema.py'))
self.assertEqual(config.schema_modnames(),
expected + [('data', 'schema')])
# app doesn't have schema file
with temp_config('data2', instance_dir, ('foo',)) as config:
self.assertEqual(pkgutil.find_loader('schema').get_filename(),
join(libdir, 'schema.py'))
self.assertEqual(config.schema_modnames(), expected)
@templibdir
def test_appobjects_modnames(self, libdir):
for filepath in (
join(libdir, 'entities.py'),
join(libdir, 'cubicweb_foo', '__init__.py'),
join(libdir, 'cubicweb_foo', 'entities', '__init__.py'),
join(libdir, 'cubicweb_foo', 'entities', 'a.py'),
# subpackages should be loaded recursively
join(libdir, 'cubicweb_foo', 'entities', 'b', '__init__.py'),
join(libdir, 'cubicweb_foo', 'entities', 'b', 'a.py'),
join(libdir, 'cubicweb_foo', 'entities', 'b', 'c', '__init__.py'),
join(libdir, 'cubicweb_foo', 'hooks.py'),
join(libdir, '_instance_dir', 'data1', 'entities.py'),
join(libdir, '_instance_dir', 'data2', 'hooks.py'),
):
create_filepath(filepath)
instance_dir = join(libdir, '_instance_dir')
expected = [
'cubicweb.entities',
'cubicweb.entities.adapters',
'cubicweb.entities.authobjs',
'cubicweb.entities.lib',
'cubicweb.entities.schemaobjs',
'cubicweb.entities.sources',
'cubicweb.entities.wfobjs',
'cubicweb_foo.entities',
'cubicweb_foo.entities.a',
'cubicweb_foo.entities.b',
'cubicweb_foo.entities.b.a',
'cubicweb_foo.entities.b.c',
'cubicweb_foo.hooks',
]
# data1 has entities
with temp_config('data1', instance_dir, ('foo',)) as config:
config.cube_appobject_path = set(['entities', 'hooks'])
self.assertEqual(config.appobjects_modnames(),
expected + ['entities'])
# data2 has hooks
with temp_config('data2', instance_dir, ('foo',)) as config:
config.cube_appobject_path = set(['entities', 'hooks'])
self.assertEqual(config.appobjects_modnames(),
expected + ['hooks'])
if __name__ == '__main__':
unittest.main()