endif()
add_subdirectory(insights)
add_subdirectory(ansible)
-add_subdirectory(orchestrator_cli)
if(WITH_TESTS)
include(AddCephTest)
from __future__ import absolute_import
+
import os
+if 'UNITTEST' in os.environ:
+ import tests
-if 'UNITTEST' not in os.environ:
- from .module import Module
-else:
- import sys
- import mock
- sys.modules['ceph_module'] = mock.Mock()
+from .module import Module
os.environ['PATH'] = '{}:{}'.format(os.path.abspath('../../../../build/bin'),
os.environ['PATH'])
- # Mock ceph module otherwise every module that is involved in a testcase and imports it will
- # raise an ImportError
- import sys
- try:
- import mock
- except ImportError:
- import unittest.mock as mock
-
- sys.modules['ceph_module'] = mock.Mock()
+ from tests import mock
mgr = mock.Mock()
mgr.get_frontend_path.side_effect = lambda: os.path.abspath("./frontend/dist")
+++ /dev/null
-if(WITH_TESTS)
- include(AddCephTest)
- add_tox_test(mgr-orchestrator_cli)
-endif()
+++ /dev/null
-tox==2.9.1
-../../../python-common
-pytest
-mock
-requests-mock
+++ /dev/null
-from __future__ import absolute_import
-import json
-
-import pytest
-
-from ceph.deployment import inventory
-from orchestrator import ReadCompletion, raise_if_exception, RGWSpec
-from orchestrator import InventoryNode, ServiceDescription
-from orchestrator import OrchestratorValidationError
-
-
-def _test_resource(data, resource_class, extra=None):
- # ensure we can deserialize and serialize
- rsc = resource_class.from_json(data)
- rsc.to_json()
-
- if extra:
- # if there is an unexpected data provided
- data.update(extra)
- with pytest.raises(OrchestratorValidationError):
- resource_class.from_json(data)
-
-
-def test_inventory():
- json_data = {
- 'name': 'host0',
- 'devices': [
- {
- 'sys_api': {
- 'rotational': '1',
- 'size': 1024,
- },
- 'path': '/dev/sda',
- 'available': False,
- 'rejected_reasons': [],
- 'lvs': []
- }
- ]
- }
- _test_resource(json_data, InventoryNode, {'abc': False})
- for devices in json_data['devices']:
- _test_resource(devices, inventory.Device)
-
- json_data = [{}, {'name': 'host0'}, {'devices': []}]
- for data in json_data:
- with pytest.raises(OrchestratorValidationError):
- InventoryNode.from_json(data)
-
-
-def test_service_description():
- json_data = {
- 'nodename': 'test',
- 'service_type': 'mon',
- 'service_instance': 'a'
- }
- _test_resource(json_data, ServiceDescription, {'abc': False})
-
-
-def test_raise():
- c = ReadCompletion()
- c.exception = ZeroDivisionError()
- with pytest.raises(ZeroDivisionError):
- raise_if_exception(c)
-
-
-def test_rgwspec():
- """
- {
- "rgw_zone": "zonename",
- "rgw_frontend_port": 8080,
- "rgw_zonegroup": "group",
- "rgw_zone_user": "user",
- "rgw_realm": "realm",
- "count": 3
- }
- """
- example = json.loads(test_rgwspec.__doc__.strip())
- spec = RGWSpec.from_json(example)
- assert spec.validate_add() is None
pytest-cov==2.7.1
+mock; python_version <= '3.3'
+../../python-common
--- /dev/null
+from __future__ import absolute_import
+
+
+import os
+
+if 'UNITTEST' in os.environ:
+
+ # Mock ceph_module. Otherwise every module that is involved in a testcase and imports it will
+ # raise an ImportError
+
+ import sys
+
+ try:
+ from unittest import mock
+ except ImportError:
+ import mock
+
+ class M(object):
+ def __init__(self, *args):
+ super(M, self).__init__()
+ self._ceph_get_version = mock.Mock()
+ self._ceph_get = mock.MagicMock()
+ self._ceph_get_module_option = mock.MagicMock()
+ self._ceph_log = mock.MagicMock()
+ self._ceph_get_option = mock.MagicMock()
+ self._ceph_get_store = lambda _: ''
+ self._ceph_get_store_prefix = lambda _: {}
+
+
+ cm = mock.Mock()
+ cm.BaseMgrModule = M
+ cm.BaseMgrStandbyModule = M
+ sys.modules['ceph_module'] = cm
+ sys.modules['rados'] = mock.Mock()
--- /dev/null
+from __future__ import absolute_import
+import json
+
+import pytest
+
+from ceph.deployment import inventory
+from orchestrator import ReadCompletion, raise_if_exception, RGWSpec
+from orchestrator import InventoryNode, ServiceDescription
+from orchestrator import OrchestratorValidationError
+
+
+def _test_resource(data, resource_class, extra=None):
+ # ensure we can deserialize and serialize
+ rsc = resource_class.from_json(data)
+ rsc.to_json()
+
+ if extra:
+ # if there is an unexpected data provided
+ data.update(extra)
+ with pytest.raises(OrchestratorValidationError):
+ resource_class.from_json(data)
+
+
+def test_inventory():
+ json_data = {
+ 'name': 'host0',
+ 'devices': [
+ {
+ 'sys_api': {
+ 'rotational': '1',
+ 'size': 1024,
+ },
+ 'path': '/dev/sda',
+ 'available': False,
+ 'rejected_reasons': [],
+ 'lvs': []
+ }
+ ]
+ }
+ _test_resource(json_data, InventoryNode, {'abc': False})
+ for devices in json_data['devices']:
+ _test_resource(devices, inventory.Device)
+
+ json_data = [{}, {'name': 'host0'}, {'devices': []}]
+ for data in json_data:
+ with pytest.raises(OrchestratorValidationError):
+ InventoryNode.from_json(data)
+
+
+def test_service_description():
+ json_data = {
+ 'nodename': 'test',
+ 'service_type': 'mon',
+ 'service_instance': 'a'
+ }
+ _test_resource(json_data, ServiceDescription, {'abc': False})
+
+
+def test_raise():
+ c = ReadCompletion()
+ c.exception = ZeroDivisionError()
+ with pytest.raises(ZeroDivisionError):
+ raise_if_exception(c)
+
+
+def test_rgwspec():
+ """
+ {
+ "rgw_zone": "zonename",
+ "rgw_frontend_port": 8080,
+ "rgw_zonegroup": "group",
+ "rgw_zone_user": "user",
+ "rgw_realm": "realm",
+ "count": 3
+ }
+ """
+ example = json.loads(test_rgwspec.__doc__.strip())
+ spec = RGWSpec.from_json(example)
+ assert spec.validate_add() is None
skipsdist = true
[testenv]
+setenv = UNITTEST = true
deps = -rrequirements.txt
-commands = pytest --cov --cov-append --cov-report=term --doctest-modules mgr_util.py {posargs}
+commands = pytest --cov --cov-append --cov-report=term --doctest-modules {posargs:mgr_util.py tests/}
\ No newline at end of file