Browse Source

Add basic unit test structure

Partially implements: blueprint add-unit-tests

Change-Id: Ic96dcaf401f30bc098738818630777fc62040aca
changes/84/479984/3
Lingxian Kong 5 years ago
parent
commit
1171944560
  1. 1
      qinling/config.py
  2. 3
      qinling/context.py
  3. 9
      qinling/db/api.py
  4. 13
      qinling/db/sqlalchemy/api.py
  5. 23
      qinling/tests/base.py
  6. 28
      qinling/tests/test_qinling.py
  7. 0
      qinling/tests/unit/__init__.py
  8. 0
      qinling/tests/unit/api/__init__.py
  9. 77
      qinling/tests/unit/api/base.py
  10. 0
      qinling/tests/unit/api/controllers/__init__.py
  11. 0
      qinling/tests/unit/api/controllers/v1/__init__.py
  12. 48
      qinling/tests/unit/api/controllers/v1/test_runtime.py
  13. 117
      qinling/tests/unit/base.py
  14. 26
      qinling/tests/unit/config.py
  15. 4
      tox.ini

1
qinling/config.py

@ -102,7 +102,6 @@ STORAGE_GROUP = 'storage'
storage_opts = [
cfg.StrOpt(
'file_system_dir',
default='/opt/qinling/funtion/packages',
help='Directory to store funtion packages.'
),
cfg.StrOpt(

3
qinling/context.py

@ -21,11 +21,8 @@ from qinling import exceptions as exc
from qinling.utils import thread_local
CONF = cfg.CONF
ALLOWED_WITHOUT_AUTH = ['/', '/v1/']
CTX_THREAD_LOCAL_NAME = "QINLING_APP_CTX_THREAD_LOCAL"
DEFAULT_PROJECT_ID = "default"

9
qinling/db/api.py

@ -54,6 +54,11 @@ def transaction():
yield
# A helper function for test.
def delete_all():
delete_runtimes(insecure=True)
# Function
@ -108,6 +113,10 @@ def update_runtime(id, values):
return IMPL.update_runtime(id, values)
def delete_runtimes(**kwargs):
return IMPL.delete_runtimes(**kwargs)
# Execution

13
qinling/db/sqlalchemy/api.py

@ -182,6 +182,14 @@ def _get_db_object_by_id(model, id, insecure=False):
return query.filter_by(id=id).first()
def _delete_all(model, insecure=False, **kwargs):
# NOTE(kong): Because we use 'in_' operator in _secure_query(), delete()
# method will raise error with default parameter. Please refer to
# http://docs.sqlalchemy.org/en/rel_1_0/orm/query.html#sqlalchemy.orm.query.Query.delete
query = db_base.model_query(model) if insecure else _secure_query(model)
query.filter_by(**kwargs).delete(synchronize_session=False)
@db_base.session_aware()
def get_function(id, session=None):
function = _get_db_object_by_id(models.Function, id)
@ -272,6 +280,11 @@ def update_runtime(id, values, session=None):
return runtime
@db_base.session_aware()
def delete_runtimes(session=None, insecure=False, **kwargs):
return _delete_all(models.Runtime, insecure=insecure, **kwargs)
@db_base.session_aware()
def create_execution(values, session=None):
execution = models.Execution()

23
qinling/tests/base.py

@ -1,23 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslotest import base
class TestCase(base.BaseTestCase):
"""Test case base class for all unit tests."""

28
qinling/tests/test_qinling.py

@ -1,28 +0,0 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
test_qinling
----------------------------------
Tests for `qinling` module.
"""
from qinling.tests import base
class TestQinling(base.TestCase):
def test_something(self):
pass

0
qinling/tests/unit/__init__.py

0
qinling/tests/unit/api/__init__.py

77
qinling/tests/unit/api/base.py

@ -0,0 +1,77 @@
# Copyright 2017 Catalyst IT Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import shutil
import tempfile
import mock
from oslo_config import cfg
import pecan
import pecan.testing
from webtest import app as webtest_app
from qinling.tests.unit import base
# Disable authentication by default for API tests.
cfg.CONF.set_default('auth_enable', False, group='pecan')
class APITest(base.DbTestCase):
def setUp(self):
super(APITest, self).setUp()
# Config package directory before app starts.
package_dir = tempfile.mkdtemp(prefix='tmp_qinling')
self.override_config('file_system_dir', package_dir, 'storage')
self.addCleanup(shutil.rmtree, package_dir, True)
pecan_opts = cfg.CONF.pecan
self.app = pecan.testing.load_test_app({
'app': {
'root': pecan_opts.root,
'modules': pecan_opts.modules,
'debug': pecan_opts.debug,
'auth_enable': False,
}
})
self.addCleanup(pecan.set_config, {}, overwrite=True)
self.addCleanup(
cfg.CONF.set_default, 'auth_enable', False, group='pecan'
)
self.patch_ctx = mock.patch('qinling.context.Context.from_environ')
self.mock_ctx = self.patch_ctx.start()
self.mock_ctx.return_value = self.ctx
self.addCleanup(self.patch_ctx.stop)
def assertNotFound(self, url):
try:
self.app.get(url, headers={'Accept': 'application/json'})
except webtest_app.AppError as error:
self.assertIn('Bad response: 404 Not Found', str(error))
return
self.fail('Expected 404 Not found but got OK')
def assertUnauthorized(self, url):
try:
self.app.get(url, headers={'Accept': 'application/json'})
except webtest_app.AppError as error:
self.assertIn('Bad response: 401 Unauthorized', str(error))
return
self.fail('Expected 401 Unauthorized but got OK')

0
qinling/tests/unit/api/controllers/__init__.py

0
qinling/tests/unit/api/controllers/v1/__init__.py

48
qinling/tests/unit/api/controllers/v1/test_runtime.py

@ -0,0 +1,48 @@
# Copyright 2017 Catalyst IT Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from qinling.db import api as db_api
from qinling import status
from qinling.tests.unit.api import base
from qinling.tests.unit import base as test_base
class TestRuntimeController(base.APITest):
def setUp(self):
super(TestRuntimeController, self).setUp()
# Insert a runtime record in db. The data will be removed in clean up.
db_runtime = db_api.create_runtime(
{
'name': 'test_runtime',
'image': 'python2.7',
'project_id': test_base.DEFAULT_PROJECT_ID,
'status': status.AVAILABLE
}
)
self.runtime_id = db_runtime.id
def test_get(self):
resp = self.app.get('/v1/runtimes/%s' % self.runtime_id)
expected = {
'id': self.runtime_id,
"image": "python2.7",
"name": "test_runtime",
"project_id": test_base.DEFAULT_PROJECT_ID,
"status": status.AVAILABLE
}
self.assertEqual(200, resp.status_int)
self._assertDictContainsSubset(resp.json, expected)

117
qinling/tests/unit/base.py

@ -0,0 +1,117 @@
# -*- coding: utf-8 -*-
# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import cfg
from oslotest import base
from qinling import context as auth_context
from qinling.db import api as db_api
from qinling.db import base as db_base
from qinling.db.sqlalchemy import sqlite_lock
from qinling.tests.unit import config as test_config
test_config.parse_args()
DEFAULT_PROJECT_ID = 'default'
OPT_PROJECT_ID = '55-66-77-88'
def get_context(default=True, admin=False):
if default:
return auth_context.Context.from_dict({
'user_name': 'test-default-user',
'user': '1-2-3-4',
'tenant': DEFAULT_PROJECT_ID,
'project_name': 'test-default-project',
'is_admin': admin
})
else:
return auth_context.Context.from_dict({
'user_name': 'test-opt-user',
'user': '5-6-7-8',
'tenant': OPT_PROJECT_ID,
'project_name': 'test-opt-project',
'is_admin': admin
})
class BaseTest(base.BaseTestCase):
def override_config(self, name, override, group=None):
"""Cleanly override CONF variables."""
cfg.CONF.set_override(name, override, group)
self.addCleanup(cfg.CONF.clear_override, name, group)
def _assertDictContainsSubset(self, parent, child, msg=None):
"""Checks whether child dict is a superset of parent.
assertDictContainsSubset() in standard Python 2.7 has been deprecated
since Python 3.2
"""
self.assertTrue(
set(child.items()).issubset(set(parent.items()))
)
class DbTestCase(BaseTest):
is_heavy_init_called = False
def setUp(self):
super(DbTestCase, self).setUp()
self._heavy_init()
self.ctx = get_context()
auth_context.set_ctx(self.ctx)
self.addCleanup(auth_context.set_ctx, None)
self.addCleanup(self._clean_db)
@classmethod
def heavy_init(cls):
"""Runs a long initialization.
This method runs long initialization once by class
and can be extended by child classes.
"""
# If using sqlite, change to memory. The default is file based.
if cfg.CONF.database.connection.startswith('sqlite'):
cfg.CONF.set_default('connection', 'sqlite://', group='database')
cfg.CONF.set_default('max_overflow', -1, group='database')
cfg.CONF.set_default('max_pool_size', 1000, group='database')
db_api.setup_db()
@classmethod
def _heavy_init(cls):
"""Method that runs heavy_init().
Make this method private to prevent extending this one.
It runs heavy_init() only once.
Note: setUpClass() can be used, but it magically is not invoked
from child class in another module.
"""
if not cls.is_heavy_init_called:
cls.heavy_init()
cls.is_heavy_init_called = True
def _clean_db(self):
db_api.delete_all()
sqlite_lock.cleanup()
if not cfg.CONF.database.connection.startswith('sqlite'):
db_base.get_engine().dispose()

26
qinling/tests/unit/config.py

@ -0,0 +1,26 @@
# Copyright 2017 Catalyst IT Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
from qinling import config
def parse_args():
# Look for .mistral.conf in the project directory by default.
project_dir = '%s/../..' % os.path.dirname(__file__)
config_file = '%s/.qinling.conf' % os.path.realpath(project_dir)
config_files = [config_file] if os.path.isfile(config_file) else None
config.parse_args(args=[], default_config_files=config_files)

4
tox.ini

@ -12,7 +12,7 @@ setenv =
deps = -r{toxinidir}/test-requirements.txt
commands =
find . -type f -name "*.pyc" -delete
python setup.py test --slowest --testr-args='{posargs}'
python setup.py testr --slowest --testr-args='{posargs}'
whitelist_externals =
rm
find
@ -25,7 +25,9 @@ commands =
oslo-config-generator --config-file tools/config/config-generator.qinling.conf \
--output-file etc/qinling.conf.sample
#set PYTHONHASHSEED=0 to prevent wsmeext.sphinxext from randomly failing.
[testenv:venv]
setenv = PYTHONHASHSEED=0
commands = {posargs}
[testenv:cover]

Loading…
Cancel
Save