diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
commit | 19fcec84d8d7d21e796c7624e521b60d28ee21ed (patch) | |
tree | 42d26aa27d1e3f7c0b8bd3fd14e7d7082f5008dc /src/cephadm/tests/fixtures.py | |
parent | Initial commit. (diff) | |
download | ceph-19fcec84d8d7d21e796c7624e521b60d28ee21ed.tar.xz ceph-19fcec84d8d7d21e796c7624e521b60d28ee21ed.zip |
Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/cephadm/tests/fixtures.py')
-rw-r--r-- | src/cephadm/tests/fixtures.py | 126 |
1 files changed, 126 insertions, 0 deletions
diff --git a/src/cephadm/tests/fixtures.py b/src/cephadm/tests/fixtures.py new file mode 100644 index 000000000..45990d910 --- /dev/null +++ b/src/cephadm/tests/fixtures.py @@ -0,0 +1,126 @@ +import mock +import os +import pytest +import time + +from contextlib import contextmanager +from pyfakefs import fake_filesystem + +from typing import Callable, Dict, List, Optional + + +with mock.patch('builtins.open', create=True): + from importlib.machinery import SourceFileLoader + cd = SourceFileLoader('cephadm', 'cephadm').load_module() + + +def mock_docker(): + docker = mock.Mock(cd.Docker) + docker.path = '/usr/bin/docker' + return docker + + +def mock_podman(): + podman = mock.Mock(cd.Podman) + podman.path = '/usr/bin/podman' + podman.version = (2, 1, 0) + return podman + + +def _daemon_path(): + return os.getcwd() + + +def mock_bad_firewalld(): + def raise_bad_firewalld(): + raise Exception('Called bad firewalld') + f = mock.Mock(cd.Firewalld) + f.enable_service_for = lambda _ : raise_bad_firewalld() + f.apply_rules = lambda : raise_bad_firewalld() + f.open_ports = lambda _ : raise_bad_firewalld() + +def _mock_scrape_host(obj, interval): + try: + raise ValueError("wah") + except Exception as e: + obj._handle_thread_exception(e, 'host') + + +def _mock_run(obj): + t = obj._create_thread(obj._scrape_host_facts, 'host', 5) + time.sleep(1) + if not t.is_alive(): + obj.cephadm_cache.update_health('host', "inactive", "host thread stopped") + + +@pytest.fixture +def exporter(): + with mock.patch('cephadm.CephadmDaemon.daemon_path', _daemon_path()), \ + mock.patch('cephadm.CephadmDaemon.can_run', return_value=True), \ + mock.patch('cephadm.CephadmDaemon.run', _mock_run), \ + mock.patch('cephadm.CephadmDaemon._scrape_host_facts', _mock_scrape_host): + + ctx = cd.CephadmContext() + exporter = cd.CephadmDaemon(ctx, fsid='foobar', daemon_id='test') + assert exporter.token == 'MyAccessToken' + yield exporter + + +@pytest.fixture() +def cephadm_fs( + fs: fake_filesystem.FakeFilesystem, +): + """ + use pyfakefs to stub filesystem calls + """ + uid = os.getuid() + gid = os.getgid() + + with mock.patch('os.fchown'), \ + mock.patch('os.fchmod'), \ + mock.patch('platform.processor', return_value='x86_64'), \ + mock.patch('cephadm.extract_uid_gid', return_value=(uid, gid)): + + if not fake_filesystem.is_root(): + fake_filesystem.set_uid(0) + + fs.create_dir(cd.DATA_DIR) + fs.create_dir(cd.LOG_DIR) + fs.create_dir(cd.LOCK_DIR) + fs.create_dir(cd.LOGROTATE_DIR) + fs.create_dir(cd.UNIT_DIR) + + yield fs + + +@contextmanager +def with_cephadm_ctx( + cmd: List[str], + container_engine: Callable = mock_podman(), + list_networks: Optional[Dict[str,Dict[str,List[str]]]] = None, + hostname: Optional[str] = None, +): + """ + :param cmd: cephadm command argv + :param container_engine: container engine mock (podman or docker) + :param list_networks: mock 'list-networks' return + :param hostname: mock 'socket.gethostname' return + """ + if not hostname: + hostname = 'host1' + + with mock.patch('cephadm.attempt_bind'), \ + mock.patch('cephadm.call', return_value=('', '', 0)), \ + mock.patch('cephadm.call_timeout', return_value=0), \ + mock.patch('cephadm.find_executable', return_value='foo'), \ + mock.patch('cephadm.is_available', return_value=True), \ + mock.patch('cephadm.json_loads_retry', return_value={'epoch' : 1}), \ + mock.patch('socket.gethostname', return_value=hostname): + ctx: cd.CephadmContext = cd.cephadm_init_ctx(cmd) + ctx.container_engine = container_engine + if list_networks is not None: + with mock.patch('cephadm.list_networks', return_value=list_networks): + yield ctx + else: + yield ctx + |