# type: ignore
-from typing import List, Optional
+
+import errno
import mock
-from mock import patch, call
import os
+import pytest
+import socket
import sys
-import unittest
-import threading
import time
-import errno
-import socket
+import threading
+import unittest
+
from http.server import HTTPServer
from urllib.request import Request, urlopen
from urllib.error import HTTPError
-import pytest
+from typing import List, Optional
-from .fixtures import exporter
+from .fixtures import (
+ cephadm_fs,
+ exporter,
+ mock_docker,
+ mock_podman,
+ with_cephadm_ctx,
+)
-with patch('builtins.open', create=True):
+
+with mock.patch('builtins.open', create=True):
from importlib.machinery import SourceFileLoader
cd = SourceFileLoader('cephadm', 'cephadm').load_module()
-class TestCephAdm(object):
-
- @staticmethod
- def mock_docker():
- docker = mock.Mock(cd.Docker)
- docker.path = '/usr/bin/docker'
- return docker
- @staticmethod
- def mock_podman():
- podman = mock.Mock(cd.Podman)
- podman.path = '/usr/bin/podman'
- podman.version = (2, 1, 0)
- return podman
+class TestCephAdm(object):
def test_docker_unit_file(self):
ctx = mock.Mock()
- ctx.container_engine = self.mock_docker()
+ ctx.container_engine = mock_docker()
r = cd.get_unit_file(ctx, '9b9d7609-f4d5-4aba-94c8-effa764d96c9')
assert 'Requires=docker.service' in r
- ctx.container_engine = self.mock_podman()
+ ctx.container_engine = mock_podman()
r = cd.get_unit_file(ctx, '9b9d7609-f4d5-4aba-94c8-effa764d96c9')
assert 'Requires=docker.service' not in r
for side_effect, expected_exception in (
(os_error(errno.EADDRINUSE), cd.PortOccupiedError),
- (os_error(errno.EAFNOSUPPORT), OSError),
- (os_error(errno.EADDRNOTAVAIL), OSError),
+ (os_error(errno.EAFNOSUPPORT), cd.Error),
+ (os_error(errno.EADDRNOTAVAIL), cd.Error),
(None, None),
):
_socket = mock.Mock()
except:
assert False
else:
- assert _socket.call_args == call(address_family, socket.SOCK_STREAM)
+ assert _socket.call_args == mock.call(address_family, socket.SOCK_STREAM)
@mock.patch('socket.socket')
@mock.patch('cephadm.logger')
):
for side_effect, expected_exception in (
(os_error(errno.EADDRINUSE), cd.PortOccupiedError),
- (os_error(errno.EADDRNOTAVAIL), OSError),
- (os_error(errno.EAFNOSUPPORT), OSError),
+ (os_error(errno.EADDRNOTAVAIL), cd.Error),
+ (os_error(errno.EAFNOSUPPORT), cd.Error),
(None, None),
):
mock_socket_obj = mock.Mock()
# test normal valid login with url, username and password specified
call_throws.return_value = '', '', 0
- ctx: Optional[cd.CephadmContext] = cd.cephadm_init_ctx(
+ ctx: cd.CephadmContext = cd.cephadm_init_ctx(
['registry-login', '--registry-url', 'sample-url',
'--registry-username', 'sample-user', '--registry-password',
'sample-pass'])
- ctx.container_engine = self.mock_docker()
- assert ctx
+ ctx.container_engine = mock_docker()
retval = cd.command_registry_login(ctx)
assert retval == 0
# test bad login attempt with invalid arguments given
- ctx: Optional[cd.CephadmContext] = cd.cephadm_init_ctx(
+ ctx: cd.CephadmContext = cd.cephadm_init_ctx(
['registry-login', '--registry-url', 'bad-args-url'])
- assert ctx
with pytest.raises(Exception) as e:
assert cd.command_registry_login(ctx)
assert str(e.value) == ('Invalid custom registry arguments received. To login to a custom registry include '
# test normal valid login with json file
get_parm.return_value = {"url": "sample-url", "username": "sample-username", "password": "sample-password"}
- ctx: Optional[cd.CephadmContext] = cd.cephadm_init_ctx(
+ ctx: cd.CephadmContext = cd.cephadm_init_ctx(
['registry-login', '--registry-json', 'sample-json'])
- ctx.container_engine = self.mock_docker()
- assert ctx
+ ctx.container_engine = mock_docker()
retval = cd.command_registry_login(ctx)
assert retval == 0
# test bad login attempt with bad json file
get_parm.return_value = {"bad-json": "bad-json"}
- ctx: Optional[cd.CephadmContext] = cd.cephadm_init_ctx(
+ ctx: cd.CephadmContext = cd.cephadm_init_ctx(
['registry-login', '--registry-json', 'sample-json'])
- assert ctx
with pytest.raises(Exception) as e:
assert cd.command_registry_login(ctx)
assert str(e.value) == ("json provided for custom registry login did not include all necessary fields. "
# test login attempt with valid arguments where login command fails
call_throws.side_effect = Exception
- ctx: Optional[cd.CephadmContext] = cd.cephadm_init_ctx(
+ ctx: cd.CephadmContext = cd.cephadm_init_ctx(
['registry-login', '--registry-url', 'sample-url',
'--registry-username', 'sample-user', '--registry-password',
'sample-pass'])
- assert ctx
with pytest.raises(Exception) as e:
cd.command_registry_login(ctx)
assert str(e.value) == "Failed to login to custom registry @ sample-url as sample-user with given password"
image = cd._filter_last_local_ceph_image(out)
assert image == 'docker.io/ceph/ceph:v15.2.5'
+ def test_normalize_image_digest(self):
+ s = 'myhostname:5000/ceph/ceph@sha256:753886ad9049004395ae990fbb9b096923b5a518b819283141ee8716ddf55ad1'
+ assert cd.normalize_image_digest(s) == s
+
+ s = 'ceph/ceph:latest'
+ assert cd.normalize_image_digest(s) == f'{cd.DEFAULT_REGISTRY}/{s}'
class TestCustomContainer(unittest.TestCase):
cc: cd.CustomContainer
_call.return_value = '', '{}, version 0.16.1'.format(daemon_type.replace('-', '_')), 0
version = cd.Monitoring.get_version(ctx, 'container_id', daemon_type)
assert version == '0.16.1'
+
+ @mock.patch('cephadm.os.fchown')
+ @mock.patch('cephadm.get_parm')
+ @mock.patch('cephadm.makedirs')
+ @mock.patch('cephadm.open')
+ @mock.patch('cephadm.make_log_dir')
+ @mock.patch('cephadm.make_data_dir')
+ def test_create_daemon_dirs_prometheus(self, make_data_dir, make_log_dir, _open, makedirs,
+ get_parm, fchown):
+ """
+ Ensures the required and optional files given in the configuration are
+ created and mapped correctly inside the container. Tests absolute and
+ relative file paths given in the configuration.
+ """
+
+ fsid = 'aaf5a720-13fe-4a3b-82b9-2d99b7fd9704'
+ daemon_type = 'prometheus'
+ uid, gid = 50, 50
+ daemon_id = 'home'
+ ctx = mock.Mock()
+ ctx.data_dir = '/somedir'
+ files = {
+ 'files': {
+ 'prometheus.yml': 'foo',
+ '/etc/prometheus/alerting/ceph_alerts.yml': 'bar'
+ }
+ }
+ get_parm.return_value = files
+
+ cd.create_daemon_dirs(ctx,
+ fsid,
+ daemon_type,
+ daemon_id,
+ uid,
+ gid,
+ config=None,
+ keyring=None)
+
+ prefix = '{data_dir}/{fsid}/{daemon_type}.{daemon_id}'.format(
+ data_dir=ctx.data_dir,
+ fsid=fsid,
+ daemon_type=daemon_type,
+ daemon_id=daemon_id
+ )
+ assert _open.call_args_list == [
+ mock.call('{}/etc/prometheus/prometheus.yml'.format(prefix), 'w',
+ encoding='utf-8'),
+ mock.call('{}/etc/prometheus/alerting/ceph_alerts.yml'.format(prefix), 'w',
+ encoding='utf-8'),
+ ]
+ assert mock.call().__enter__().write('foo') in _open.mock_calls
+ assert mock.call().__enter__().write('bar') in _open.mock_calls
+
+
+class TestBootstrap(object):
+
+ @staticmethod
+ def _get_cmd(*args):
+ return [
+ 'bootstrap',
+ '--allow-mismatched-release',
+ '--skip-prepare-host',
+ '--skip-dashboard',
+ *args,
+ ]
+
+ def test_config(self, cephadm_fs):
+ conf_file = 'foo'
+ cmd = self._get_cmd(
+ '--mon-ip', '192.168.1.1',
+ '--skip-mon-network',
+ '--config', conf_file,
+ )
+
+ with with_cephadm_ctx(cmd) as ctx:
+ msg = r'No such file or directory'
+ with pytest.raises(cd.Error, match=msg):
+ cd.command_bootstrap(ctx)
+
+ cephadm_fs.create_file(conf_file)
+ with with_cephadm_ctx(cmd) as ctx:
+ retval = cd.command_bootstrap(ctx)
+ assert retval == 0
+
+ def test_no_mon_addr(self, cephadm_fs):
+ cmd = self._get_cmd()
+ with with_cephadm_ctx(cmd) as ctx:
+ msg = r'must specify --mon-ip or --mon-addrv'
+ with pytest.raises(cd.Error, match=msg):
+ cd.command_bootstrap(ctx)
+
+ def test_skip_mon_network(self, cephadm_fs):
+ cmd = self._get_cmd('--mon-ip', '192.168.1.1')
+
+ with with_cephadm_ctx(cmd, list_networks={}) as ctx:
+ msg = r'--skip-mon-network'
+ with pytest.raises(cd.Error, match=msg):
+ cd.command_bootstrap(ctx)
+
+ cmd += ['--skip-mon-network']
+ with with_cephadm_ctx(cmd, list_networks={}) as ctx:
+ retval = cd.command_bootstrap(ctx)
+ assert retval == 0
+
+ @pytest.mark.parametrize('mon_ip, list_networks, result',
+ [
+ # IPv4
+ (
+ 'eth0',
+ {'192.168.1.0/24': {'eth0': ['192.168.1.1']}},
+ False,
+ ),
+ (
+ '0.0.0.0',
+ {'192.168.1.0/24': {'eth0': ['192.168.1.1']}},
+ False,
+ ),
+ (
+ '192.168.1.0',
+ {'192.168.1.0/24': {'eth0': ['192.168.1.1']}},
+ False,
+ ),
+ (
+ '192.168.1.1',
+ {'192.168.1.0/24': {'eth0': ['192.168.1.1']}},
+ True,
+ ),
+ (
+ '192.168.1.1:1234',
+ {'192.168.1.0/24': {'eth0': ['192.168.1.1']}},
+ True,
+ ),
+ # IPv6
+ (
+ '::',
+ {'192.168.1.0/24': {'eth0': ['192.168.1.1']}},
+ False,
+ ),
+ (
+ '::ffff:192.168.1.0',
+ {"ffff::/64": {"eth0": ["::ffff:c0a8:101"]}},
+ False,
+ ),
+ (
+ '::ffff:192.168.1.1',
+ {"ffff::/64": {"eth0": ["::ffff:c0a8:101"]}},
+ True,
+ ),
+ (
+ '::ffff:c0a8:101',
+ {"ffff::/64": {"eth0": ["::ffff:c0a8:101"]}},
+ True,
+ ),
+ (
+ '0000:0000:0000:0000:0000:FFFF:C0A8:0101',
+ {"ffff::/64": {"eth0": ["::ffff:c0a8:101"]}},
+ True,
+ ),
+ ])
+ def test_mon_ip(self, mon_ip, list_networks, result, cephadm_fs):
+ cmd = self._get_cmd('--mon-ip', mon_ip)
+ if not result:
+ with with_cephadm_ctx(cmd, list_networks=list_networks) as ctx:
+ msg = r'--skip-mon-network'
+ with pytest.raises(cd.Error, match=msg):
+ cd.command_bootstrap(ctx)
+ else:
+ with with_cephadm_ctx(cmd, list_networks=list_networks) as ctx:
+ retval = cd.command_bootstrap(ctx)
+ assert retval == 0
+
+ def test_allow_fqdn_hostname(self, cephadm_fs):
+ hostname = 'foo.bar'
+ cmd = self._get_cmd(
+ '--mon-ip', '192.168.1.1',
+ '--skip-mon-network',
+ )
+
+ with with_cephadm_ctx(cmd, hostname=hostname) as ctx:
+ msg = r'--allow-fqdn-hostname'
+ with pytest.raises(cd.Error, match=msg):
+ cd.command_bootstrap(ctx)
+
+ cmd += ['--allow-fqdn-hostname']
+ with with_cephadm_ctx(cmd, hostname=hostname) as ctx:
+ retval = cd.command_bootstrap(ctx)
+ assert retval == 0
+
+ @pytest.mark.parametrize('fsid, err',
+ [
+ ('', None),
+ ('00000000-0000-0000-0000-0000deadbeef', None),
+ ('00000000-0000-0000-0000-0000deadbeez', 'not an fsid'),
+ ])
+ def test_fsid(self, fsid, err, cephadm_fs):
+ cmd = self._get_cmd(
+ '--mon-ip', '192.168.1.1',
+ '--skip-mon-network',
+ '--fsid', fsid,
+ )
+
+ with with_cephadm_ctx(cmd) as ctx:
+ if err:
+ with pytest.raises(cd.Error, match=err):
+ cd.command_bootstrap(ctx)
+ else:
+ retval = cd.command_bootstrap(ctx)
+ assert retval == 0
projects / ceph.git / blobdiff