self._check_task_status_na()
+ def test_scrub_when_mds_is_inactive(self):
+ test_dir = "scrub_control_test_path"
+ abs_test_path = f"/{test_dir}"
+
+ self.create_scrub_data(test_dir)
+
+ # allow standby-replay
+ self.fs.set_max_mds(1)
+ self.fs.set_allow_standby_replay(True)
+ status = self.fs.wait_for_daemons()
+ sr_mds_id = self.fs.get_daemon_names('up:standby-replay', status=status)[0]
+
+ # start the scrub and verify
+ with self.assertRaises(CommandFailedError) as ce:
+ self.run_ceph_cmd('tell', f'mds.{sr_mds_id}', 'scrub',
+ 'start', abs_test_path, 'recursive')
+ self.assertEqual(ce.exception.exitstatus, errno.EINVAL)
+
+ # pause and verify
+ with self.assertRaises(CommandFailedError) as ce:
+ self.run_ceph_cmd('tell', f'mds.{sr_mds_id}', 'scrub', 'pause')
+ self.assertEqual(ce.exception.exitstatus, errno.EINVAL)
+
+ # abort and verify
+ with self.assertRaises(CommandFailedError) as ce:
+ self.run_ceph_cmd('tell', f'mds.{sr_mds_id}', 'scrub', 'abort')
+ self.assertEqual(ce.exception.exitstatus, errno.EINVAL)
+
+ # resume and verify
+ with self.assertRaises(CommandFailedError) as ce:
+ self.run_ceph_cmd('tell', f'mds.{sr_mds_id}', 'scrub', 'resume')
+ self.assertEqual(ce.exception.exitstatus, errno.EINVAL)
+
+
class TestScrubChecks(CephFSTestCase):
"""
Run flush and scrub commands on the specified files in the filesystem. This