+++ /dev/null
-import logging
-import os
-import subprocess
-import sys
-
-
-LOG = logging.getLogger(__name__)
-
-
-def _prepend_path(env):
- """
- Make sure the PATH contains the location where the Python binary
- lives. This makes sure cli tools installed in a virtualenv work.
- """
- if env is None:
- env = os.environ
- env = dict(env)
- new = os.path.dirname(sys.executable)
- path = env.get('PATH')
- if path is not None:
- new = new + ':' + path
- env['PATH'] = new
- return env
-
-
-class CLIFailed(Exception):
- """CLI tool failed"""
-
- def __init__(self, args, status):
- self.args = args
- self.status = status
-
- def __str__(self):
- return '{doc}: {args}: exited with status {status}'.format(
- doc=self.__doc__,
- args=self.args,
- status=self.status,
- )
-
-
-class CLIProcess(object):
- def __init__(self, **kw):
- self.kw = kw
-
- def __enter__(self):
- try:
- self.p = subprocess.Popen(**self.kw)
- except OSError as e:
- raise AssertionError(
- 'CLI tool {args!r} does not work: {err}'.format(
- args=self.kw['args'],
- err=e,
- ),
- )
- else:
- return self.p
-
- def __exit__(self, exc_type, exc_val, exc_tb):
- self.p.wait()
- if self.p.returncode != 0:
- err = CLIFailed(
- args=self.kw['args'],
- status=self.p.returncode,
- )
- if exc_type is None:
- # nothing else raised, so we should complain; if
- # something else failed, we'll just log
- raise err
- else:
- LOG.error(str(err))
-
-
-class CLITester(object):
- # provide easy way for caller to access the exception class
- # without importing us
- Failed = CLIFailed
-
- def __init__(self, tmpdir):
- self.tmpdir = tmpdir
-
- def __call__(self, **kw):
- kw.setdefault('cwd', str(self.tmpdir))
- kw['env'] = _prepend_path(kw.get('env'))
- kw['env']['COLUMNS'] = '80'
- return CLIProcess(**kw)
-
-
-def pytest_funcarg__cli(request):
- """
- Test command line behavior.
- """
-
- # the tmpdir here will be the same value as the test function
- # sees; we rely on that to let caller prepare and introspect
- # any files the cli tool will read or create
- tmpdir = request.getfuncargvalue('tmpdir')
-
- return CLITester(tmpdir=tmpdir)
+++ /dev/null
-import contextlib
-import os
-
-
-@contextlib.contextmanager
-def directory(path):
- prev = os.open('.', os.O_RDONLY | os.O_DIRECTORY)
- try:
- os.chdir(path)
- yield
- finally:
- os.fchdir(prev)
- os.close(prev)
+++ /dev/null
-
-
-def fake_getaddrinfo(*a, **kw):
- return_host = kw.get('return_host', 'host1')
- return [[0,0,0,0, return_host]]
+++ /dev/null
-import pytest
-import subprocess
-
-
-def test_help(tmpdir, cli):
- with cli(
- args=['ceph-deploy', '--help'],
- stdout=subprocess.PIPE,
- ) as p:
- result = p.stdout.read()
- assert 'usage: ceph-deploy' in result
- assert 'Deploy Ceph' in result
- assert 'optional arguments:' in result
- assert 'commands:' in result
-
-
-def test_bad_command(tmpdir, cli):
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', 'bork'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'usage: ceph-deploy' in result
- assert err.value.status == 2
- assert {p.basename for p in tmpdir.listdir()} == set()
-
-
-def test_bad_cluster(tmpdir, cli):
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', '--cluster=/evil-this-should-not-be-created', 'new'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'usage: ceph-deploy' in result
- assert err.value.status == 2
- assert {p.basename for p in tmpdir.listdir()} == set()
+++ /dev/null
-import argparse
-import collections
-import mock
-import pytest
-import subprocess
-
-from ..cli import main
-from .. import install
-
-from .directory import directory
-
-
-def test_help(tmpdir, cli):
- with cli(
- args=['ceph-deploy', 'install', '--help'],
- stdout=subprocess.PIPE,
- ) as p:
- result = p.stdout.read()
- assert 'usage: ceph-deploy' in result
- assert 'positional arguments:' in result
- assert 'optional arguments:' in result
-
-
-def test_bad_no_host(tmpdir, cli):
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', 'install'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'usage: ceph-deploy install' in result
- assert 'too few arguments' in result
- assert err.value.status == 2
-
-
-def test_simple(tmpdir):
- ns = argparse.Namespace()
- ns.pushy = mock.Mock()
- conn = mock.NonCallableMock(name='PushyClient')
- ns.pushy.return_value = conn
-
- mock_compiled = collections.defaultdict(mock.Mock)
- conn.compile.return_value = mock.Mock(return_value = ('Ubuntu', 'precise','cuttlefish'))
-
-
- try:
- with directory(str(tmpdir)):
- main(
- args=['-v', 'install', 'storehost1'],
- namespace=ns,
- )
- except SystemExit as e:
- raise AssertionError('Unexpected exit: %s', e)
-
- ns.pushy.assert_has_calls([
- mock.call('ssh+sudo:storehost1'),
- ])
-
- call_list = conn.compile.call_args_list
- mock.call(install.lsb.check_lsb_release) == call_list[0]
- mock.call(install.lsb.lsb_release) == call_list[1]
- mock.call(install.install_debian) == call_list[2]
- assert len(call_list) == 3
+++ /dev/null
-import argparse
-import collections
-import mock
-import pytest
-import subprocess
-
-from ..cli import main
-from .. import mon
-
-from .directory import directory
-from .fakes import fake_getaddrinfo
-
-def test_help(tmpdir, cli):
- with cli(
- args=['ceph-deploy', 'mon', '--help'],
- stdout=subprocess.PIPE,
- ) as p:
- result = p.stdout.read()
- assert 'usage: ceph-deploy' in result
- assert 'Deploy ceph monitor on remote hosts.' in result
- assert 'positional arguments:'
- assert 'optional arguments:'
-
-
-def test_bad_no_conf(tmpdir, cli):
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', 'mon'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'usage: ceph-deploy' in result
- assert 'too few arguments' in result
- assert err.value.status == 2
-
-
-def test_bad_no_mon(tmpdir, cli):
- with tmpdir.join('ceph.conf').open('w'):
- pass
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', 'mon'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'usage: ceph-deploy mon' in result
- assert 'too few arguments' in result
- assert err.value.status == 2
-
-
-def test_simple(tmpdir, capsys):
- with tmpdir.join('ceph.conf').open('w') as f:
- f.write("""\
-[global]
-fsid = 6ede5564-3cf1-44b5-aa96-1c77b0c3e1d0
-mon initial members = host1
-""")
-
- ns = argparse.Namespace()
- ns.pushy = mock.Mock()
- conn = mock.NonCallableMock(name='PushyClient')
- ns.pushy.return_value = conn
-
- mock_compiled = collections.defaultdict(mock.Mock)
- conn.compile.side_effect = mock_compiled.__getitem__
-
- MON_SECRET = 'AQBWDj5QAP6LHhAAskVBnUkYHJ7eYREmKo5qKA=='
-
- def _create_mon(cluster, get_monitor_secret):
- secret = get_monitor_secret()
- assert secret == MON_SECRET
-
- mock_compiled[mon.create_mon].side_effect = _create_mon
-
- try:
- with mock.patch('socket.getaddrinfo', fake_getaddrinfo):
- with directory(str(tmpdir)):
- main(
- args=['-v', 'new', 'host1'],
- namespace=ns,
- )
- main(
- args=['-v', 'mon', 'create', 'host1'],
- namespace=ns,
- )
- except SystemExit as e:
- raise AssertionError('Unexpected exit: %s', e)
- out, err = capsys.readouterr()
- err = err.lower()
- assert 'creating new cluster named ceph' in err
- assert 'monitor host1 at h' in err
- assert 'resolving host host1' in err
- assert "monitor initial members are ['host1']" in err
- assert "monitor addrs are ['h']" in err
+++ /dev/null
-import pytest
-from mock import patch
-import re
-import subprocess
-import uuid
-
-from .. import conf
-from ..cli import main
-from .directory import directory
-from .fakes import fake_getaddrinfo
-
-
-def test_help(tmpdir, cli):
- with cli(
- args=['ceph-deploy', 'new', '--help'],
- stdout=subprocess.PIPE,
- ) as p:
- result = p.stdout.read()
- assert 'usage: ceph-deploy new' in result
- assert 'positional arguments' in result
- assert 'optional arguments' in result
-
-
-def test_write_global_conf_section(tmpdir, cli):
- with patch('ceph_deploy.new.socket.getaddrinfo', fake_getaddrinfo):
- with directory(str(tmpdir)):
- main(args=['new', 'host1'])
- with tmpdir.join('ceph.conf').open() as f:
- cfg = conf.parse(f)
- assert cfg.sections() == ['global']
-
-
-def pytest_funcarg__newcfg(request):
- tmpdir = request.getfuncargvalue('tmpdir')
- cli = request.getfuncargvalue('cli')
-
- def new(*args):
- with patch('ceph_deploy.new.socket.getaddrinfo', fake_getaddrinfo):
- with directory(str(tmpdir)):
- main( args=['new'] + list(args))
- with tmpdir.join('ceph.conf').open() as f:
- cfg = conf.parse(f)
- return cfg
- return new
-
-
-def test_uuid(newcfg):
- cfg = newcfg('host1')
- fsid = cfg.get('global', 'fsid')
- # make sure it's a valid uuid
- uuid.UUID(hex=fsid)
- # make sure it looks pretty, too
- UUID_RE = re.compile(
- r'^[0-9a-f]{8}-'
- + r'[0-9a-f]{4}-'
- # constant 4 here, we want to enforce randomness and not leak
- # MACs or time
- + r'4[0-9a-f]{3}-'
- + r'[0-9a-f]{4}-'
- + r'[0-9a-f]{12}$',
- )
- assert UUID_RE.match(fsid)
-
-
-def test_mons(newcfg):
- cfg = newcfg('node01', 'node07', 'node34')
- mon_initial_members = cfg.get('global', 'mon_initial_members')
- assert mon_initial_members == 'node01, node07, node34'
-
-
-def test_defaults(newcfg):
- cfg = newcfg('host1')
- assert cfg.get('global', 'auth_supported') == 'cephx'
- assert cfg.get('global', 'osd_journal_size') == '1024'
- assert cfg.get('global', 'filestore_xattr_use_omap') == 'true'
+++ /dev/null
-import argparse
-import collections
-import mock
-import pytest
-import subprocess
-
-from ..cli import main
-from .. import osd
-
-from .directory import directory
-
-
-def test_help(tmpdir, cli):
- with cli(
- args=['ceph-deploy', 'osd', '--help'],
- stdout=subprocess.PIPE,
- ) as p:
- result = p.stdout.read()
- assert 'usage: ceph-deploy osd' in result
- assert 'positional arguments' in result
- assert 'optional arguments' in result
-
-
-def test_bad_no_conf(tmpdir, cli):
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', 'osd', 'fakehost:/does-not-exist'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'ceph-deploy osd: error' in result
- assert 'invalid choice' in result
- assert err.value.status == 2
-
-
-def test_bad_no_disk(tmpdir, cli):
- with tmpdir.join('ceph.conf').open('w'):
- pass
- with pytest.raises(cli.Failed) as err:
- with cli(
- args=['ceph-deploy', 'osd'],
- stderr=subprocess.PIPE,
- ) as p:
- result = p.stderr.read()
- assert 'usage: ceph-deploy osd' in result
- assert err.value.status == 2
-
-
-def test_simple(tmpdir, capsys):
- with tmpdir.join('ceph.conf').open('w') as f:
- f.write("""\
-[global]
-fsid = 6ede5564-3cf1-44b5-aa96-1c77b0c3e1d0
-mon host = host1
-""")
-
- ns = argparse.Namespace()
-
- conn_osd = mock.NonCallableMock(name='PushyClient-osd')
- mock_compiled_osd = collections.defaultdict(mock.Mock)
- #conn_osd.compile.side_effect = mock_compiled_osd.__getitem__
- conn_osd.compile.return_value = mock.Mock(return_value='fakekeyring')
-
- conn_mon = mock.NonCallableMock(name='PushyClient-mon')
- mock_compiled_mon = collections.defaultdict(mock.Mock)
- conn_mon.compile.side_effect = mock_compiled_mon.__getitem__
-
- ns.pushy = mock.Mock(name='pushy namespace')
-
- def _conn(url):
- if url == 'ssh+sudo:host1':
- return conn_mon
- elif url == 'ssh+sudo:storehost1:sdc':
- return conn_osd
- else:
- raise AssertionError('Unexpected connection url: %r', url)
- ns.pushy.side_effect = _conn
-
- BOOTSTRAP_KEY = 'fakekeyring'
-
- mock_compiled_mon[osd.get_bootstrap_osd_key].side_effect = BOOTSTRAP_KEY
-
- def _create_osd(cluster, find_key):
- key = find_key()
- assert key == BOOTSTRAP_KEY
-
- mock_compiled_osd[osd.create_osd].side_effect = _create_osd
-
- with directory(str(tmpdir)):
- main(
- args=['-v', 'gatherkeys', 'storehost1:sdc'],
- namespace=ns,
- )
- main(
- args=['-v', 'osd', 'prepare', 'storehost1:sdc'],
- namespace=ns,
- )
- out, err = capsys.readouterr()
- err = err.lower()
- assert 'have ceph.mon.keyring' in err
- assert 'have ceph.client.admin.keyring' in err
- assert 'have ceph.bootstrap-osd.keyring' in err
- assert 'got ceph.bootstrap-mds.keyring key from storehost1:sdc' in err
- assert 'got ceph.bootstrap-osd.keyring key from storehost1:sdc' in err
+++ /dev/null
-from cStringIO import StringIO
-from .. import conf
-
-
-def test_simple():
- f = StringIO("""\
-[foo]
-bar = baz
-""")
- cfg = conf.parse(f)
- assert cfg.get('foo', 'bar') == 'baz'
-
-
-def test_indent_space():
- f = StringIO("""\
-[foo]
- bar = baz
-""")
- cfg = conf.parse(f)
- assert cfg.get('foo', 'bar') == 'baz'
-
-
-def test_indent_tab():
- f = StringIO("""\
-[foo]
-\tbar = baz
-""")
- cfg = conf.parse(f)
- assert cfg.get('foo', 'bar') == 'baz'
-
-
-def test_words_underscore():
- f = StringIO("""\
-[foo]
-bar_thud = baz
-""")
- cfg = conf.parse(f)
- assert cfg.get('foo', 'bar_thud') == 'baz'
- assert cfg.get('foo', 'bar thud') == 'baz'
-
-
-def test_words_space():
- f = StringIO("""\
-[foo]
-bar thud = baz
-""")
- cfg = conf.parse(f)
- assert cfg.get('foo', 'bar_thud') == 'baz'
- assert cfg.get('foo', 'bar thud') == 'baz'
-
-
-def test_words_many():
- f = StringIO("""\
-[foo]
-bar__ thud quux = baz
-""")
- cfg = conf.parse(f)
- assert cfg.get('foo', 'bar_thud_quux') == 'baz'
- assert cfg.get('foo', 'bar thud quux') == 'baz'
--- /dev/null
+import logging
+import os
+import subprocess
+import sys
+
+
+LOG = logging.getLogger(__name__)
+
+
+def _prepend_path(env):
+ """
+ Make sure the PATH contains the location where the Python binary
+ lives. This makes sure cli tools installed in a virtualenv work.
+ """
+ if env is None:
+ env = os.environ
+ env = dict(env)
+ new = os.path.dirname(sys.executable)
+ path = env.get('PATH')
+ if path is not None:
+ new = new + ':' + path
+ env['PATH'] = new
+ return env
+
+
+class CLIFailed(Exception):
+ """CLI tool failed"""
+
+ def __init__(self, args, status):
+ self.args = args
+ self.status = status
+
+ def __str__(self):
+ return '{doc}: {args}: exited with status {status}'.format(
+ doc=self.__doc__,
+ args=self.args,
+ status=self.status,
+ )
+
+
+class CLIProcess(object):
+ def __init__(self, **kw):
+ self.kw = kw
+
+ def __enter__(self):
+ try:
+ self.p = subprocess.Popen(**self.kw)
+ except OSError as e:
+ raise AssertionError(
+ 'CLI tool {args!r} does not work: {err}'.format(
+ args=self.kw['args'],
+ err=e,
+ ),
+ )
+ else:
+ return self.p
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.p.wait()
+ if self.p.returncode != 0:
+ err = CLIFailed(
+ args=self.kw['args'],
+ status=self.p.returncode,
+ )
+ if exc_type is None:
+ # nothing else raised, so we should complain; if
+ # something else failed, we'll just log
+ raise err
+ else:
+ LOG.error(str(err))
+
+
+class CLITester(object):
+ # provide easy way for caller to access the exception class
+ # without importing us
+ Failed = CLIFailed
+
+ def __init__(self, tmpdir):
+ self.tmpdir = tmpdir
+
+ def __call__(self, **kw):
+ kw.setdefault('cwd', str(self.tmpdir))
+ kw['env'] = _prepend_path(kw.get('env'))
+ kw['env']['COLUMNS'] = '80'
+ return CLIProcess(**kw)
+
+
+def pytest_funcarg__cli(request):
+ """
+ Test command line behavior.
+ """
+
+ # the tmpdir here will be the same value as the test function
+ # sees; we rely on that to let caller prepare and introspect
+ # any files the cli tool will read or create
+ tmpdir = request.getfuncargvalue('tmpdir')
+
+ return CLITester(tmpdir=tmpdir)
--- /dev/null
+import contextlib
+import os
+
+
+@contextlib.contextmanager
+def directory(path):
+ prev = os.open('.', os.O_RDONLY | os.O_DIRECTORY)
+ try:
+ os.chdir(path)
+ yield
+ finally:
+ os.fchdir(prev)
+ os.close(prev)
--- /dev/null
+
+
+def fake_getaddrinfo(*a, **kw):
+ return_host = kw.get('return_host', 'host1')
+ return [[0,0,0,0, return_host]]
--- /dev/null
+import pytest
+import subprocess
+
+
+def test_help(tmpdir, cli):
+ with cli(
+ args=['ceph-deploy', '--help'],
+ stdout=subprocess.PIPE,
+ ) as p:
+ result = p.stdout.read()
+ assert 'usage: ceph-deploy' in result
+ assert 'Deploy Ceph' in result
+ assert 'optional arguments:' in result
+ assert 'commands:' in result
+
+
+def test_bad_command(tmpdir, cli):
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', 'bork'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'usage: ceph-deploy' in result
+ assert err.value.status == 2
+ assert {p.basename for p in tmpdir.listdir()} == set()
+
+
+def test_bad_cluster(tmpdir, cli):
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', '--cluster=/evil-this-should-not-be-created', 'new'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'usage: ceph-deploy' in result
+ assert err.value.status == 2
+ assert {p.basename for p in tmpdir.listdir()} == set()
--- /dev/null
+import argparse
+import collections
+import mock
+import pytest
+import subprocess
+
+from ..cli import main
+from .. import install
+
+from .directory import directory
+
+
+def test_help(tmpdir, cli):
+ with cli(
+ args=['ceph-deploy', 'install', '--help'],
+ stdout=subprocess.PIPE,
+ ) as p:
+ result = p.stdout.read()
+ assert 'usage: ceph-deploy' in result
+ assert 'positional arguments:' in result
+ assert 'optional arguments:' in result
+
+
+def test_bad_no_host(tmpdir, cli):
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', 'install'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'usage: ceph-deploy install' in result
+ assert 'too few arguments' in result
+ assert err.value.status == 2
+
+
+def test_simple(tmpdir):
+ ns = argparse.Namespace()
+ ns.pushy = mock.Mock()
+ conn = mock.NonCallableMock(name='PushyClient')
+ ns.pushy.return_value = conn
+
+ mock_compiled = collections.defaultdict(mock.Mock)
+ conn.compile.return_value = mock.Mock(return_value = ('Ubuntu', 'precise','cuttlefish'))
+
+
+ try:
+ with directory(str(tmpdir)):
+ main(
+ args=['-v', 'install', 'storehost1'],
+ namespace=ns,
+ )
+ except SystemExit as e:
+ raise AssertionError('Unexpected exit: %s', e)
+
+ ns.pushy.assert_has_calls([
+ mock.call('ssh+sudo:storehost1'),
+ ])
+
+ call_list = conn.compile.call_args_list
+ mock.call(install.lsb.check_lsb_release) == call_list[0]
+ mock.call(install.lsb.lsb_release) == call_list[1]
+ mock.call(install.install_debian) == call_list[2]
+ assert len(call_list) == 3
--- /dev/null
+import argparse
+import collections
+import mock
+import pytest
+import subprocess
+
+from ..cli import main
+from .. import mon
+
+from .directory import directory
+from .fakes import fake_getaddrinfo
+
+def test_help(tmpdir, cli):
+ with cli(
+ args=['ceph-deploy', 'mon', '--help'],
+ stdout=subprocess.PIPE,
+ ) as p:
+ result = p.stdout.read()
+ assert 'usage: ceph-deploy' in result
+ assert 'Deploy ceph monitor on remote hosts.' in result
+ assert 'positional arguments:'
+ assert 'optional arguments:'
+
+
+def test_bad_no_conf(tmpdir, cli):
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', 'mon'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'usage: ceph-deploy' in result
+ assert 'too few arguments' in result
+ assert err.value.status == 2
+
+
+def test_bad_no_mon(tmpdir, cli):
+ with tmpdir.join('ceph.conf').open('w'):
+ pass
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', 'mon'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'usage: ceph-deploy mon' in result
+ assert 'too few arguments' in result
+ assert err.value.status == 2
+
+
+def test_simple(tmpdir, capsys):
+ with tmpdir.join('ceph.conf').open('w') as f:
+ f.write("""\
+[global]
+fsid = 6ede5564-3cf1-44b5-aa96-1c77b0c3e1d0
+mon initial members = host1
+""")
+
+ ns = argparse.Namespace()
+ ns.pushy = mock.Mock()
+ conn = mock.NonCallableMock(name='PushyClient')
+ ns.pushy.return_value = conn
+
+ mock_compiled = collections.defaultdict(mock.Mock)
+ conn.compile.side_effect = mock_compiled.__getitem__
+
+ MON_SECRET = 'AQBWDj5QAP6LHhAAskVBnUkYHJ7eYREmKo5qKA=='
+
+ def _create_mon(cluster, get_monitor_secret):
+ secret = get_monitor_secret()
+ assert secret == MON_SECRET
+
+ mock_compiled[mon.create_mon].side_effect = _create_mon
+
+ try:
+ with mock.patch('socket.getaddrinfo', fake_getaddrinfo):
+ with directory(str(tmpdir)):
+ main(
+ args=['-v', 'new', 'host1'],
+ namespace=ns,
+ )
+ main(
+ args=['-v', 'mon', 'create', 'host1'],
+ namespace=ns,
+ )
+ except SystemExit as e:
+ raise AssertionError('Unexpected exit: %s', e)
+ out, err = capsys.readouterr()
+ err = err.lower()
+ assert 'creating new cluster named ceph' in err
+ assert 'monitor host1 at h' in err
+ assert 'resolving host host1' in err
+ assert "monitor initial members are ['host1']" in err
+ assert "monitor addrs are ['h']" in err
--- /dev/null
+import pytest
+from mock import patch
+import re
+import subprocess
+import uuid
+
+from .. import conf
+from ..cli import main
+from .directory import directory
+from .fakes import fake_getaddrinfo
+
+
+def test_help(tmpdir, cli):
+ with cli(
+ args=['ceph-deploy', 'new', '--help'],
+ stdout=subprocess.PIPE,
+ ) as p:
+ result = p.stdout.read()
+ assert 'usage: ceph-deploy new' in result
+ assert 'positional arguments' in result
+ assert 'optional arguments' in result
+
+
+def test_write_global_conf_section(tmpdir, cli):
+ with patch('ceph_deploy.new.socket.getaddrinfo', fake_getaddrinfo):
+ with directory(str(tmpdir)):
+ main(args=['new', 'host1'])
+ with tmpdir.join('ceph.conf').open() as f:
+ cfg = conf.parse(f)
+ assert cfg.sections() == ['global']
+
+
+def pytest_funcarg__newcfg(request):
+ tmpdir = request.getfuncargvalue('tmpdir')
+ cli = request.getfuncargvalue('cli')
+
+ def new(*args):
+ with patch('ceph_deploy.new.socket.getaddrinfo', fake_getaddrinfo):
+ with directory(str(tmpdir)):
+ main( args=['new'] + list(args))
+ with tmpdir.join('ceph.conf').open() as f:
+ cfg = conf.parse(f)
+ return cfg
+ return new
+
+
+def test_uuid(newcfg):
+ cfg = newcfg('host1')
+ fsid = cfg.get('global', 'fsid')
+ # make sure it's a valid uuid
+ uuid.UUID(hex=fsid)
+ # make sure it looks pretty, too
+ UUID_RE = re.compile(
+ r'^[0-9a-f]{8}-'
+ + r'[0-9a-f]{4}-'
+ # constant 4 here, we want to enforce randomness and not leak
+ # MACs or time
+ + r'4[0-9a-f]{3}-'
+ + r'[0-9a-f]{4}-'
+ + r'[0-9a-f]{12}$',
+ )
+ assert UUID_RE.match(fsid)
+
+
+def test_mons(newcfg):
+ cfg = newcfg('node01', 'node07', 'node34')
+ mon_initial_members = cfg.get('global', 'mon_initial_members')
+ assert mon_initial_members == 'node01, node07, node34'
+
+
+def test_defaults(newcfg):
+ cfg = newcfg('host1')
+ assert cfg.get('global', 'auth_supported') == 'cephx'
+ assert cfg.get('global', 'osd_journal_size') == '1024'
+ assert cfg.get('global', 'filestore_xattr_use_omap') == 'true'
--- /dev/null
+import argparse
+import collections
+import mock
+import pytest
+import subprocess
+
+from ..cli import main
+from .. import osd
+
+from .directory import directory
+
+
+def test_help(tmpdir, cli):
+ with cli(
+ args=['ceph-deploy', 'osd', '--help'],
+ stdout=subprocess.PIPE,
+ ) as p:
+ result = p.stdout.read()
+ assert 'usage: ceph-deploy osd' in result
+ assert 'positional arguments' in result
+ assert 'optional arguments' in result
+
+
+def test_bad_no_conf(tmpdir, cli):
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', 'osd', 'fakehost:/does-not-exist'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'ceph-deploy osd: error' in result
+ assert 'invalid choice' in result
+ assert err.value.status == 2
+
+
+def test_bad_no_disk(tmpdir, cli):
+ with tmpdir.join('ceph.conf').open('w'):
+ pass
+ with pytest.raises(cli.Failed) as err:
+ with cli(
+ args=['ceph-deploy', 'osd'],
+ stderr=subprocess.PIPE,
+ ) as p:
+ result = p.stderr.read()
+ assert 'usage: ceph-deploy osd' in result
+ assert err.value.status == 2
+
+
+def test_simple(tmpdir, capsys):
+ with tmpdir.join('ceph.conf').open('w') as f:
+ f.write("""\
+[global]
+fsid = 6ede5564-3cf1-44b5-aa96-1c77b0c3e1d0
+mon host = host1
+""")
+
+ ns = argparse.Namespace()
+
+ conn_osd = mock.NonCallableMock(name='PushyClient-osd')
+ mock_compiled_osd = collections.defaultdict(mock.Mock)
+ #conn_osd.compile.side_effect = mock_compiled_osd.__getitem__
+ conn_osd.compile.return_value = mock.Mock(return_value='fakekeyring')
+
+ conn_mon = mock.NonCallableMock(name='PushyClient-mon')
+ mock_compiled_mon = collections.defaultdict(mock.Mock)
+ conn_mon.compile.side_effect = mock_compiled_mon.__getitem__
+
+ ns.pushy = mock.Mock(name='pushy namespace')
+
+ def _conn(url):
+ if url == 'ssh+sudo:host1':
+ return conn_mon
+ elif url == 'ssh+sudo:storehost1:sdc':
+ return conn_osd
+ else:
+ raise AssertionError('Unexpected connection url: %r', url)
+ ns.pushy.side_effect = _conn
+
+ BOOTSTRAP_KEY = 'fakekeyring'
+
+ mock_compiled_mon[osd.get_bootstrap_osd_key].side_effect = BOOTSTRAP_KEY
+
+ def _create_osd(cluster, find_key):
+ key = find_key()
+ assert key == BOOTSTRAP_KEY
+
+ mock_compiled_osd[osd.create_osd].side_effect = _create_osd
+
+ with directory(str(tmpdir)):
+ main(
+ args=['-v', 'gatherkeys', 'storehost1:sdc'],
+ namespace=ns,
+ )
+ main(
+ args=['-v', 'osd', 'prepare', 'storehost1:sdc'],
+ namespace=ns,
+ )
+ out, err = capsys.readouterr()
+ err = err.lower()
+ assert 'have ceph.mon.keyring' in err
+ assert 'have ceph.client.admin.keyring' in err
+ assert 'have ceph.bootstrap-osd.keyring' in err
+ assert 'got ceph.bootstrap-mds.keyring key from storehost1:sdc' in err
+ assert 'got ceph.bootstrap-osd.keyring key from storehost1:sdc' in err
--- /dev/null
+from cStringIO import StringIO
+from .. import conf
+
+
+def test_simple():
+ f = StringIO("""\
+[foo]
+bar = baz
+""")
+ cfg = conf.parse(f)
+ assert cfg.get('foo', 'bar') == 'baz'
+
+
+def test_indent_space():
+ f = StringIO("""\
+[foo]
+ bar = baz
+""")
+ cfg = conf.parse(f)
+ assert cfg.get('foo', 'bar') == 'baz'
+
+
+def test_indent_tab():
+ f = StringIO("""\
+[foo]
+\tbar = baz
+""")
+ cfg = conf.parse(f)
+ assert cfg.get('foo', 'bar') == 'baz'
+
+
+def test_words_underscore():
+ f = StringIO("""\
+[foo]
+bar_thud = baz
+""")
+ cfg = conf.parse(f)
+ assert cfg.get('foo', 'bar_thud') == 'baz'
+ assert cfg.get('foo', 'bar thud') == 'baz'
+
+
+def test_words_space():
+ f = StringIO("""\
+[foo]
+bar thud = baz
+""")
+ cfg = conf.parse(f)
+ assert cfg.get('foo', 'bar_thud') == 'baz'
+ assert cfg.get('foo', 'bar thud') == 'baz'
+
+
+def test_words_many():
+ f = StringIO("""\
+[foo]
+bar__ thud quux = baz
+""")
+ cfg = conf.parse(f)
+ assert cfg.get('foo', 'bar_thud_quux') == 'baz'
+ assert cfg.get('foo', 'bar thud quux') == 'baz'
projects / ceph-deploy.git / commitdiff