ceph
/
ceph-ansible
mirror of https://github.com/ceph/ceph-ansible.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

tests/mons: update tests to use new host fixture 

Signed-off-by: Alfredo Deza <adeza@redhat.com>
pull/1781/head
Alfredo Deza 2017-08-21 15:42:24 -04:00
parent ecf917d354
commit 31a960c323
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
tests/functional/tests/mon/test_mons.py
View File

@ -4,46 +4,46 @@ import pytest
class TestMons(object):
@pytest.mark.no_docker
def test_ceph_mon_package_is_installed(self, node, Package):
assert Package("ceph-mon").is_installed
def test_ceph_mon_package_is_installed(self, node, host):
assert host.package("ceph-mon").is_installed
def test_mon_listens_on_6789(self, node, Socket):
assert Socket("tcp://%s:6789" % node["address"]).is_listening
def test_mon_listens_on_6789(self, node, host):
assert host.socket("tcp://%s:6789" % node["address"]).is_listening
def test_mon_service_is_running(self, node, Service):
def test_mon_service_is_running(self, node, host):
service_name = "ceph-mon@ceph-{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert Service(service_name).is_running
assert host.service(service_name).is_running
def test_mon_service_is_enabled(self, node, Service):
def test_mon_service_is_enabled(self, node, host):
service_name = "ceph-mon@ceph-{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert Service(service_name).is_enabled
assert host.service(service_name).is_enabled
@pytest.mark.no_docker
def test_can_get_cluster_health(self, node, Command):
def test_can_get_cluster_health(self, node, host):
cmd = "sudo ceph --cluster={} --connect-timeout 5 -s".format(node["cluster_name"])
output = Command.check_output(cmd)
output = host.check_output(cmd)
assert output.strip().startswith("cluster")
class TestOSDs(object):
@pytest.mark.no_docker
def test_all_osds_are_up_and_in(self, node, Command):
def test_all_osds_are_up_and_in(self, node, host):
cmd = "sudo ceph --cluster={} --connect-timeout 5 -s".format(node["cluster_name"])
output = Command.check_output(cmd)
output = host.check_output(cmd)
phrase = "{num_osds} osds: {num_osds} up, {num_osds} in".format(num_osds=node["total_osds"])
assert phrase in output
@pytest.mark.docker
def test_all_docker_osds_are_up_and_in(self, node, Command):
def test_all_docker_osds_are_up_and_in(self, node, host):
cmd = "sudo docker exec ceph-mon-ceph-{} ceph --cluster={} --connect-timeout 5 -s".format(
node["vars"]["inventory_hostname"],
node["cluster_name"]
)
output = Command.check_output(cmd)
output = host.check_output(cmd)
phrase = "{num_osds} osds: {num_osds} up, {num_osds} in".format(num_osds=node["total_osds"])
assert phrase in output