ceph/qa/tasks/cephfs/test_cephfs_shell.py

   1 import logging
   2 from StringIO import StringIO
   3 from tasks.cephfs.cephfs_test_case import CephFSTestCase
   4 from tasks.cephfs.fuse_mount import FuseMount
   5 from teuthology.exceptions import CommandFailedError
   6
   7 log = logging.getLogger(__name__)
   8
   9
  10 class TestCephFSShell(CephFSTestCase):
  11     CLIENTS_REQUIRED = 1
  12     py_version = 'python'
  13
  14     def setUp(self):
  15         CephFSTestCase.setUp(self)
  16         self.py_version = self.ctx.config.get('overrides', {}).get('python', 'python')
  17         log.info("using python version: {}".format(self.py_version))
  18
  19     def _cephfs_shell(self, cmd, opts=None):
  20         args = ["cephfs-shell", "-c", self.mount_a.config_path]
  21         if opts is not None:
  22             args.extend(opts)
  23         args.extend(("--", cmd))
  24         log.info("Running command: {}".format(" ".join(args)))
  25         status = self.mount_a.client_remote.run(args=args, stdout=StringIO())
  26         return status.stdout.getvalue().strip()
  27
  28     def test_help(self):
  29         """
  30         Test that help outputs commands.
  31         """
  32
  33         o = self._cephfs_shell("help")
  34
  35         log.info("output:\n{}".format(o))