ceph/src/ceph-volume/ceph_volume/tests/functional/tests/osd/test_osds.py

   1 import json
   2
   3
   4 class TestOSDs(object):
   5
   6     def test_ceph_osd_package_is_installed(self, node, host):
   7         assert host.package("ceph-osd").is_installed
   8
   9     def test_osds_listen_on_public_network(self, node, host):
  10         # TODO: figure out way to paramaterize this test
  11         nb_port = (node["num_osds"] * node["num_osd_ports"])
  12         assert host.check_output(
  13             "netstat -lntp | grep ceph-osd | grep %s | wc -l" % (node["address"])) == str(nb_port)  # noqa E501
  14
  15     def test_osds_listen_on_cluster_network(self, node, host):
  16         # TODO: figure out way to paramaterize this test
  17         nb_port = (node["num_osds"] * node["num_osd_ports"])
  18         assert host.check_output("netstat -lntp | grep ceph-osd | grep %s | wc -l" %  # noqa E501
  19                                  (node["cluster_address"])) == str(nb_port)
  20
  21     def test_osd_services_are_running(self, node, host):
  22         # TODO: figure out way to paramaterize node['osds'] for this test
  23         for osd in node["osds"]:
  24             assert host.service("ceph-osd@%s" % osd).is_running
  25
  26     def test_osd_are_mounted(self, node, host):
  27         # TODO: figure out way to paramaterize node['osd_ids'] for this test
  28         for osd_id in node["osd_ids"]:
  29             osd_path = "/var/lib/ceph/osd/{cluster}-{osd_id}".format(
  30                 cluster=node["cluster_name"],
  31                 osd_id=osd_id,
  32             )
  33             assert host.mount_point(osd_path).exists
  34
  35     def test_ceph_volume_is_installed(self, node, host):
  36         host.exists('ceph-volume')
  37
  38     def test_ceph_volume_systemd_is_installed(self, node, host):
  39         host.exists('ceph-volume-systemd')
  40
  41     def _get_osd_id_from_host(self, node, osd_tree):
  42         children = []
  43         for n in osd_tree['nodes']:
  44             if n['name'] == node['vars']['inventory_hostname'] and n['type'] == 'host':  # noqa E501
  45                 children = n['children']
  46         return children
  47
  48     def _get_nb_up_osds_from_ids(self, node, osd_tree):
  49         nb_up = 0
  50         ids = self._get_osd_id_from_host(node, osd_tree)
  51         for n in osd_tree['nodes']:
  52             if n['id'] in ids and n['status'] == 'up':
  53                 nb_up += 1
  54         return nb_up
  55
  56     def test_all_osds_are_up_and_in(self, node, host):
  57         cmd = "sudo ceph --cluster={cluster} --connect-timeout 5 --keyring /var/lib/ceph/bootstrap-osd/{cluster}.keyring -n client.bootstrap-osd osd tree -f json".format(  # noqa E501
  58             cluster=node["cluster_name"])
  59         output = json.loads(host.check_output(cmd))
  60         assert node["num_osds"] == self._get_nb_up_osds_from_ids(node, output)