Commit a7ecbaa3 authored by Daniele Venzano's avatar Daniele Venzano

Add some unit tests and mock objects, generate coverage report

parent cb7a1bcd
......@@ -36,7 +36,7 @@ unittests:
- pip install -U pip setuptools
- pip install -r requirements.txt
- pip install -r requirements_tests.txt
- pytest --ignore tests --tb=short
- pytest --ignore tests --tb=short --cov-report=term --cov zoe_api --cov zoe_lib --cov zoe_master
docs-test:
stage: static-test
......
......@@ -3,7 +3,7 @@
set -e
pylint *.py zoe_* tests/*.py
pytest --ignore tests --tb=short
pytest --ignore tests --tb=short --cov-report=term --cov zoe_api --cov zoe_master --cov zoe_lib
doc8 docs/
sh ./build_docs.sh
......@@ -37,9 +37,9 @@ class APIEndpoint:
:type master: zoe_api.master_api.APIManager
:type sql: zoe_lib.sql_manager.SQLManager
"""
def __init__(self):
self.master = zoe_api.master_api.APIManager()
self.sql = zoe_lib.state.SQLManager(get_conf())
def __init__(self, master_api, sql_manager):
self.master = master_api
self.sql = sql_manager
def execution_by_id(self, uid, role, execution_id) -> zoe_lib.state.sql_manager.Execution:
"""Lookup an execution by its ID."""
......
......@@ -23,9 +23,11 @@ from tornado.ioloop import IOLoop, PeriodicCallback
from tornado.web import Application
import zoe_lib.config as config
import zoe_lib.state
import zoe_api.db_init
import zoe_api.api_endpoint
import zoe_api.rest_api
import zoe_api.master_api
import zoe_api.web
import zoe_api.auth.ldap
from zoe_api.web.custom_request_handler import JinjaApp
......@@ -57,7 +59,9 @@ def zoe_web_main() -> int:
zoe_api.db_init.init()
api_endpoint = zoe_api.api_endpoint.APIEndpoint()
master_api = zoe_api.master_api.APIManager()
sql_manager = zoe_lib.state.SQLManager(config.get_conf())
api_endpoint = zoe_api.api_endpoint.APIEndpoint(master_api, sql_manager)
app_settings = {
'static_path': os.path.join(os.path.dirname(__file__), "web", "static"),
......
# Copyright (c) 2017, Daniele Venzano
#
# 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 module for the API endpoint."""
import pytest
from zoe_api.api_endpoint import APIEndpoint
from zoe_api.tests.mock_master_api import MockAPIManager
from zoe_lib.state.tests.mock_sql_manager import MockSQLManager
class TestAPIEndpoint:
"""The test class."""
@pytest.fixture(params=[True, False])
def master_api(self, request):
"""A mock master API manager that can succeed or fail."""
ret = MockAPIManager()
ret.fails = request.param
return ret
@pytest.fixture
def sql_manager(self):
"""A mock SQLManager that can succeed or fail."""
return MockSQLManager()
def test_statistics_scheduler(self, master_api, sql_manager):
"""Test the scheduler statistics API."""
api = APIEndpoint(master_api, sql_manager)
ret = api.statistics_scheduler('nouser', 'norole')
assert ret == 'No error message' or ret is None
# Copyright (c) 2017, Daniele Venzano
#
# 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.
"""Mock master API for unit testing."""
import logging
from typing import Tuple
log = logging.getLogger(__name__)
APIReturnType = Tuple[bool, str]
class MockAPIManager:
"""Main class for the API."""
def __init__(self):
self.fails = False
def _request_reply(self) -> APIReturnType:
"""
Fake connection.
"""
if not self.fails:
return True, 'No error message'
else:
return False, 'Fake error message'
def execution_start(self, exec_id: int) -> APIReturnType:
"""Start an execution."""
assert isinstance(exec_id, int)
return self._request_reply()
def execution_terminate(self, exec_id: int) -> APIReturnType:
"""Terminate an execution."""
assert isinstance(exec_id, int)
return self._request_reply()
def execution_delete(self, exec_id: int) -> APIReturnType:
"""Delete an execution."""
assert isinstance(exec_id, int)
return self._request_reply()
def scheduler_statistics(self) -> APIReturnType:
"""Query scheduler statistics."""
return self._request_reply()
# Copyright (c) 2017, Daniele Venzano
#
# 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.
"""Mock SQL Manager for unit testing."""
import sqlite3
from collections import namedtuple
from zoe_lib.state.sql_manager import SQLManager
Conf = namedtuple('Conf', ['dbuser', 'dbpass', 'dbhost', 'dbport', 'dbname', 'deployment_name'])
class MockSQLManager(SQLManager):
"""A mock SQL manager."""
def __init__(self):
fake_conf = Conf(dbuser='', dbpass='', dbhost='', dbport=5432, dbname='', deployment_name='test')
super().__init__(fake_conf)
def _connect(self):
self.conn = sqlite3.connect(':memory:')
def _cursor(self):
return self.conn.cursor()
......@@ -31,7 +31,7 @@ class TestApplicationsMethods:
def test_pass_for_complex_zapp(self):
"""Test zapp validation code."""
zapp_fp = json.load(open('contrib/zoeapps/complex_zapp.json', 'r'))
zapp_fp = json.load(open('tests/complex_zapp.json', 'r'))
applications.app_validate(zapp_fp)
def test_fails_for_zapp(self):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment