def get_ceph_cluster_count(ctx: CephadmContext) -> int:
- return len([c for c in os.listdir(ctx.data_dir) if is_fsid(c)])
+ if os.path.isdir(ctx.data_dir):
+ return len([c for c in os.listdir(ctx.data_dir) if is_fsid(c)])
+ return 0
def command_rm_cluster(ctx: CephadmContext) -> None:
])
def test_get_ceph_cluster_count(self, test_input, expected):
ctx = _cephadm.CephadmContext()
- with mock.patch('os.listdir', return_value=test_input):
- assert _cephadm.get_ceph_cluster_count(ctx) == expected
+ with mock.patch('os.path.isdir', return_value=True):
+ with mock.patch('os.listdir', return_value=test_input):
+ assert _cephadm.get_ceph_cluster_count(ctx) == expected
+
+ def test_get_ceph_cluster_count_missing_datadir(self):
+ ctx = _cephadm.CephadmContext()
+ with mock.patch('os.path.isdir', return_value=False):
+ assert _cephadm.get_ceph_cluster_count(ctx) == 0
def test_set_image_minimize_config(self):
def throw_cmd(cmd):