]>
git.proxmox.com Git - ceph.git/blob - ceph/qa/tasks/mgr/dashboard/test_cephfs.py
1 # -*- coding: utf-8 -*-
2 from __future__
import absolute_import
5 from contextlib
import contextmanager
7 from .helper
import DashboardTestCase
, JObj
, JList
, JLeaf
10 class CephfsTest(DashboardTestCase
):
13 AUTH_ROLES
= ['cephfs-manager']
15 QUOTA_PATH
= '/quotas'
17 def assertToHave(self
, data
, key
):
18 self
.assertIn(key
, data
)
19 self
.assertIsNotNone(data
[key
])
22 return self
.fs
.get_namespace_id()
24 def mk_dirs(self
, path
, expectedStatus
=200):
25 self
._post
("/api/cephfs/{}/mk_dirs".format(self
.get_fs_id()),
26 params
={'path': path
})
27 self
.assertStatus(expectedStatus
)
29 def rm_dir(self
, path
, expectedStatus
=200):
30 self
._post
("/api/cephfs/{}/rm_dir".format(self
.get_fs_id()),
31 params
={'path': path
})
32 self
.assertStatus(expectedStatus
)
34 def get_root_directory(self
, expectedStatus
=200):
35 data
= self
._get
("/api/cephfs/{}/get_root_directory".format(self
.get_fs_id()))
36 self
.assertStatus(expectedStatus
)
37 self
.assertIsInstance(data
, dict)
40 def ls_dir(self
, path
, expectedLength
, depth
= None):
41 return self
._ls
_dir
(path
, expectedLength
, depth
, "api")
43 def ui_ls_dir(self
, path
, expectedLength
, depth
= None):
44 return self
._ls
_dir
(path
, expectedLength
, depth
, "ui-api")
46 def _ls_dir(self
, path
, expectedLength
, depth
, baseApiPath
):
47 params
= {'path': path
}
49 params
['depth'] = depth
50 data
= self
._get
("/{}/cephfs/{}/ls_dir".format(baseApiPath
, self
.get_fs_id()),
52 self
.assertStatus(200)
53 self
.assertIsInstance(data
, list)
54 self
.assertEqual(len(data
), expectedLength
)
57 def setQuotas(self
, bytes
=None, files
=None):
62 self
._post
("/api/cephfs/{}/set_quotas".format(self
.get_fs_id()), data
=quotas
,
63 params
={'path': self
.QUOTA_PATH
})
64 self
.assertStatus(200)
66 def assertQuotas(self
, bytes
, files
):
67 data
= self
.ls_dir('/', 1)[0]
68 self
.assertEqual(data
['quotas']['max_bytes'], bytes
)
69 self
.assertEqual(data
['quotas']['max_files'], files
)
72 def new_quota_dir(self
):
73 self
.mk_dirs(self
.QUOTA_PATH
)
74 self
.setQuotas(1024**3, 1024)
76 self
.rm_dir(self
.QUOTA_PATH
)
78 @DashboardTestCase.RunAs('test', 'test', ['block-manager'])
79 def test_access_permissions(self
):
80 fs_id
= self
.get_fs_id()
81 self
._get
("/api/cephfs/{}/clients".format(fs_id
))
82 self
.assertStatus(403)
83 self
._get
("/api/cephfs/{}".format(fs_id
))
84 self
.assertStatus(403)
85 self
._get
("/api/cephfs/{}/mds_counters".format(fs_id
))
86 self
.assertStatus(403)
87 self
._get
("/ui-api/cephfs/{}/tabs".format(fs_id
))
88 self
.assertStatus(403)
90 def test_cephfs_clients(self
):
91 fs_id
= self
.get_fs_id()
92 data
= self
._get
("/api/cephfs/{}/clients".format(fs_id
))
93 self
.assertStatus(200)
95 self
.assertIn('status', data
)
96 self
.assertIn('data', data
)
98 def test_cephfs_evict_client_does_not_exist(self
):
99 fs_id
= self
.get_fs_id()
100 self
._delete
("/api/cephfs/{}/client/1234".format(fs_id
))
101 self
.assertStatus(404)
103 def test_cephfs_evict_invalid_client_id(self
):
104 fs_id
= self
.get_fs_id()
105 self
._delete
("/api/cephfs/{}/client/xyz".format(fs_id
))
106 self
.assertStatus(400)
107 self
.assertJsonBody({
108 "component": 'cephfs',
109 "code": "invalid_cephfs_client_id",
110 "detail": "Invalid cephfs client ID xyz"
113 def test_cephfs_get(self
):
114 fs_id
= self
.get_fs_id()
115 data
= self
._get
("/api/cephfs/{}/".format(fs_id
))
116 self
.assertStatus(200)
118 self
.assertToHave(data
, 'cephfs')
119 self
.assertToHave(data
, 'standbys')
120 self
.assertToHave(data
, 'versions')
122 def test_cephfs_mds_counters(self
):
123 fs_id
= self
.get_fs_id()
124 data
= self
._get
("/api/cephfs/{}/mds_counters".format(fs_id
))
125 self
.assertStatus(200)
127 self
.assertIsInstance(data
, dict)
128 self
.assertIsNotNone(data
)
130 def test_cephfs_mds_counters_wrong(self
):
131 self
._get
("/api/cephfs/baadbaad/mds_counters")
132 self
.assertStatus(400)
133 self
.assertJsonBody({
134 "component": 'cephfs',
135 "code": "invalid_cephfs_id",
136 "detail": "Invalid cephfs ID baadbaad"
139 def test_cephfs_list(self
):
140 data
= self
._get
("/api/cephfs/")
141 self
.assertStatus(200)
143 self
.assertIsInstance(data
, list)
145 self
.assertToHave(cephfs
, 'id')
146 self
.assertToHave(cephfs
, 'mdsmap')
148 def test_cephfs_get_quotas(self
):
149 fs_id
= self
.get_fs_id()
150 data
= self
._get
("/api/cephfs/{}/get_quotas?path=/".format(fs_id
))
151 self
.assertStatus(200)
152 self
.assertSchema(data
, JObj({
157 def test_cephfs_tabs(self
):
158 fs_id
= self
.get_fs_id()
159 data
= self
._get
("/ui-api/cephfs/{}/tabs".format(fs_id
))
160 self
.assertStatus(200)
161 self
.assertIsInstance(data
, dict)
164 pools
= data
['pools']
165 self
.assertIsInstance(pools
, list)
166 self
.assertGreater(len(pools
), 0)
168 self
.assertEqual(pool
['size'], pool
['used'] + pool
['avail'])
171 self
.assertToHave(data
, 'ranks')
172 self
.assertIsInstance(data
['ranks'], list)
175 self
.assertToHave(data
, 'name')
176 self
.assertIsInstance(data
['name'], six
.string_types
)
179 self
.assertToHave(data
, 'standbys')
180 self
.assertIsInstance(data
['standbys'], six
.string_types
)
183 counters
= data
['mds_counters']
184 self
.assertIsInstance(counters
, dict)
185 self
.assertGreater(len(counters
.keys()), 0)
186 for k
, v
in counters
.items():
187 self
.assertEqual(v
['name'], k
)
190 self
.assertToHave(data
, 'clients')
191 clients
= data
['clients']
192 self
.assertToHave(clients
, 'data')
193 self
.assertIsInstance(clients
['data'], list)
194 self
.assertToHave(clients
, 'status')
195 self
.assertIsInstance(clients
['status'], int)
197 def test_ls_mk_rm_dir(self
):
200 self
.mk_dirs('/pictures/birds')
201 self
.ls_dir('/', 2, 3)
202 self
.ls_dir('/pictures', 1)
204 self
.rm_dir('/pictures', 500)
205 self
.rm_dir('/pictures/birds')
206 self
.rm_dir('/pictures')
210 def test_snapshots(self
):
211 fs_id
= self
.get_fs_id()
212 self
.mk_dirs('/movies/dune/extended_version')
214 self
._post
("/api/cephfs/{}/mk_snapshot".format(fs_id
),
215 params
={'path': '/movies/dune', 'name': 'test'})
216 self
.assertStatus(200)
218 data
= self
.ls_dir('/movies', 1)
219 self
.assertSchema(data
[0], JObj(sub_elems
={
222 'parent': JLeaf(str),
223 'snapshots': JList(JObj(sub_elems
={
226 'created': JLeaf(str)
228 'quotas': JObj(sub_elems
={
229 'max_bytes': JLeaf(int),
230 'max_files': JLeaf(int)
233 snapshots
= data
[0]['snapshots']
234 self
.assertEqual(len(snapshots
), 1)
235 snapshot
= snapshots
[0]
236 self
.assertEqual(snapshot
['name'], "test")
237 self
.assertEqual(snapshot
['path'], "/movies/dune/.snap/test")
239 # Should have filtered out "_test_$timestamp"
240 data
= self
.ls_dir('/movies/dune', 1)
241 snapshots
= data
[0]['snapshots']
242 self
.assertEqual(len(snapshots
), 0)
244 self
._post
("/api/cephfs/{}/rm_snapshot".format(fs_id
),
245 params
={'path': '/movies/dune', 'name': 'test'})
246 self
.assertStatus(200)
248 data
= self
.ls_dir('/movies', 1)
249 self
.assertEqual(len(data
[0]['snapshots']), 0)
251 # Cleanup. Note, the CephFS Python extension (and therefor the Dashboard
252 # REST API) does not support recursive deletion of a directory.
253 self
.rm_dir('/movies/dune/extended_version')
254 self
.rm_dir('/movies/dune')
255 self
.rm_dir('/movies')
257 def test_quotas_default(self
):
258 self
.mk_dirs(self
.QUOTA_PATH
)
259 self
.assertQuotas(0, 0)
260 self
.rm_dir(self
.QUOTA_PATH
)
262 def test_quotas_set_both(self
):
263 with self
.new_quota_dir():
264 self
.assertQuotas(1024**3, 1024)
266 def test_quotas_set_only_bytes(self
):
267 with self
.new_quota_dir():
268 self
.setQuotas(2048**3)
269 self
.assertQuotas(2048**3, 1024)
271 def test_quotas_set_only_files(self
):
272 with self
.new_quota_dir():
273 self
.setQuotas(None, 2048)
274 self
.assertQuotas(1024**3, 2048)
276 def test_quotas_unset_both(self
):
277 with self
.new_quota_dir():
279 self
.assertQuotas(0, 0)
281 def test_listing_of_root_dir(self
):
282 self
.ls_dir('/', 0) # Should not list root
283 ui_root
= self
.ui_ls_dir('/', 1)[0] # Should list root by default
284 root
= self
.get_root_directory()
285 self
.assertEqual(ui_root
, root
)
287 def test_listing_of_ui_api_ls_on_deeper_levels(self
):
288 # The UI-API and API ls_dir methods should behave the same way on deeper levels
289 self
.mk_dirs('/pictures')
290 api_ls
= self
.ls_dir('/pictures', 0)
291 ui_api_ls
= self
.ui_ls_dir('/pictures', 0)
292 self
.assertEqual(api_ls
, ui_api_ls
)
293 self
.rm_dir('/pictures')