|
1 # |
|
2 # Copyright (c) 2009 Nokia Corporation and/or its subsidiary(-ies). |
|
3 # All rights reserved. |
|
4 # This component and the accompanying materials are made available |
|
5 # under the terms of "Eclipse Public License v1.0" |
|
6 # which accompanies this distribution, and is available |
|
7 # at the URL "http://www.eclipse.org/legal/epl-v10.html". |
|
8 # |
|
9 # Initial Contributors: |
|
10 # Nokia Corporation - initial contribution. |
|
11 # |
|
12 # Contributors: |
|
13 # |
|
14 # Description: |
|
15 # |
|
16 |
|
17 """ |
|
18 Test Respource |
|
19 """ |
|
20 import unittest |
|
21 import string |
|
22 import sys,os |
|
23 import pickle |
|
24 import __init__ |
|
25 |
|
26 from cone.public import api, exceptions, utils |
|
27 |
|
28 ROOT_PATH = os.path.dirname(os.path.abspath(__file__)) |
|
29 temp_dir = os.path.join(ROOT_PATH, "temp/storage") |
|
30 storage_path = os.path.join(temp_dir, "layertest.pk") |
|
31 |
|
32 class TestStorage(unittest.TestCase): |
|
33 def setUp(self): |
|
34 pass |
|
35 |
|
36 def test_create_storage(self): |
|
37 res = api.Storage("") |
|
38 self.assertTrue(res) |
|
39 |
|
40 def test_get_root(self): |
|
41 path = os.path.join(temp_dir, "foo/faa.pk") |
|
42 store = api.Storage.open(path,"w") |
|
43 self.assertEquals(store.get_path(),path) |
|
44 |
|
45 def test_set_path(self): |
|
46 path = os.path.join(temp_dir, "foo/faa.pk") |
|
47 store = api.Storage.open(path,"w") |
|
48 self.assertEquals(store.get_path(),path) |
|
49 store.set_current_path("faa") |
|
50 self.assertEquals(store.get_current_path(),"faa") |
|
51 |
|
52 def test_get_more_read(self): |
|
53 storage = api.Storage("") |
|
54 self.assertEquals(storage.get_mode("r"),storage.MODE_READ) |
|
55 self.assertEquals(storage.get_mode("rb"),storage.MODE_READ) |
|
56 |
|
57 def test_get_more_write(self): |
|
58 storage = api.Storage("") |
|
59 self.assertEquals(storage.get_mode("w"),storage.MODE_WRITE) |
|
60 self.assertEquals(storage.get_mode("wb"),storage.MODE_WRITE) |
|
61 |
|
62 def test_get_more_append(self): |
|
63 storage = api.Storage("") |
|
64 self.assertEquals(storage.get_mode("a"),storage.MODE_APPEND) |
|
65 self.assertEquals(storage.get_mode("ab"),storage.MODE_APPEND) |
|
66 |
|
67 def test_get_more_unknown(self): |
|
68 storage = api.Storage("") |
|
69 self.assertEquals(storage.get_mode("1"),storage.MODE_UNKNOWN) |
|
70 self.assertEquals(storage.get_mode("2b"),storage.MODE_UNKNOWN) |
|
71 |
|
72 class TestStorageGeneric(unittest.TestCase): |
|
73 def test_list_resources(self): |
|
74 store = api.Storage.open(storage_path,"w") |
|
75 self.assertTrue(store) |
|
76 self.assertEquals(store.list_resources(""), []) |
|
77 |
|
78 def test_open_resource_nonexisting(self): |
|
79 store = api.Storage.open(storage_path,"w") |
|
80 self.assertTrue(store) |
|
81 try: |
|
82 res = store.open_resource("test") |
|
83 res.close() |
|
84 self.fail("Opening non existing resource succeeds") |
|
85 except exceptions.NotResource,e: |
|
86 pass |
|
87 |
|
88 def test_is_resource_nonexisting(self): |
|
89 store = api.Storage.open(storage_path,"w") |
|
90 self.assertTrue(store) |
|
91 self.assertFalse(store.is_resource("test")) |
|
92 |
|
93 def test_open_resource_is_resource(self): |
|
94 store = api.Storage.open(storage_path,"w") |
|
95 self.assertTrue(store) |
|
96 res = store.open_resource("test","w") |
|
97 res.write("Testing writing more") |
|
98 res.close() |
|
99 self.assertTrue(store.is_resource('test')) |
|
100 |
|
101 def test_open_resource_write_and_write(self): |
|
102 store = api.Storage.open(storage_path,"w") |
|
103 self.assertTrue(store) |
|
104 res = store.open_resource("test","w") |
|
105 res.write("Testing writing more") |
|
106 res.close() |
|
107 res = store.open_resource("test","w") |
|
108 res.write("writing") |
|
109 res.close() |
|
110 self.assertEquals(store._get('test').data, 'writing') |
|
111 |
|
112 def test_get_size_on_write_only_resource_fails(self): |
|
113 store = api.Storage.open(storage_path,"w") |
|
114 self.assertTrue(store) |
|
115 res = store.open_resource("test","w") |
|
116 res.write("Writing foobar") |
|
117 self.assertRaises(exceptions.StorageException, res.get_size) |
|
118 res.close() |
|
119 |
|
120 def test_open_resource_and_get_size(self): |
|
121 store = api.Storage.open(storage_path,"w") |
|
122 self.assertTrue(store) |
|
123 res = store.open_resource("test","w") |
|
124 res.write("Writing foobar") |
|
125 res.close() |
|
126 |
|
127 res = store.open_resource("test","r") |
|
128 self.assertEquals(res.get_size(), 14) |
|
129 res.close() |
|
130 |
|
131 def test_write_fails_on_read_mode(self): |
|
132 store = api.Storage.open(storage_path, "w") |
|
133 self.assertTrue(store) |
|
134 res = store.open_resource("test","w") |
|
135 res.write("Testing writing more") |
|
136 res.close() |
|
137 resr = store.open_resource("test","r") |
|
138 try: |
|
139 resr.write("Testing writing more") |
|
140 resr.close() |
|
141 self.fail("Writing succeeds on read mode?") |
|
142 except exceptions.StorageException, e: |
|
143 pass |
|
144 |
|
145 def test_open_resource_append(self): |
|
146 store = api.Storage.open(storage_path,"w") |
|
147 self.assertTrue(store) |
|
148 res = store.open_resource("test","a") |
|
149 res.write("Testing append.\n") |
|
150 res.close() |
|
151 res = store.open_resource("test","a") |
|
152 res.write("appending!") |
|
153 res.close() |
|
154 self.assertEquals(store.test.data, 'Testing append.\nappending!') |
|
155 |
|
156 def test_open_multiple_no_close_and_write_closed(self): |
|
157 store = api.Storage.open(storage_path,"w") |
|
158 self.assertTrue(store) |
|
159 res1 = store.open_resource("test","a") |
|
160 res1.write("Testing append.\n") |
|
161 res1.close() |
|
162 res2 = store.open_resource("test","a") |
|
163 res2.write("appending!") |
|
164 res2.close() |
|
165 try: |
|
166 res2.write("sss") |
|
167 self.fail("writing on closed object succeeds?") |
|
168 except ValueError: |
|
169 pass |
|
170 self.assertEquals(store.test.data, 'Testing append.\nappending!') |
|
171 |
|
172 def test_open_many_to_one(self): |
|
173 store = api.Storage.open(storage_path,"w") |
|
174 self.assertTrue(store) |
|
175 res1 = store.open_resource("test/foo.txt","a") |
|
176 res1.write("Testing\n") |
|
177 res1.close() |
|
178 res1 = store.open_resource("test/foo.txt","w") |
|
179 res1.close() |
|
180 self.assertEquals(store.list_resources('', True),['test/foo.txt']) |
|
181 |
|
182 def test_open_many(self): |
|
183 store = api.Storage.open(storage_path,"w") |
|
184 self.assertTrue(store) |
|
185 res1 = store.open_resource("test/foo.txt","a") |
|
186 res1.write("Testing\n") |
|
187 res1.close() |
|
188 res2 = store.open_resource("test/bar.txt","a") |
|
189 res2.write("Writing bar!") |
|
190 res2.close() |
|
191 self.assertEquals(store.test.foo.data, 'Testing\n') |
|
192 self.assertEquals(store.test.bar.data, 'Writing bar!') |
|
193 |
|
194 def test_open_many_to_a_hierarchy_and_list_folders(self): |
|
195 store = api.Storage.open(storage_path,"w") |
|
196 self.assertTrue(store) |
|
197 root= store.open_resource("root.txt","a") |
|
198 root.write("root\n") |
|
199 root.close() |
|
200 res1 = store.open_resource("test/foo.txt","a") |
|
201 res1.write("Testing\n") |
|
202 res1.close() |
|
203 res2 = store.open_resource("test/bar.txt","a") |
|
204 res2.write("Writing bar!") |
|
205 res2.close() |
|
206 self.assertEquals(store.list_resources('/'), ['root.txt']) |
|
207 self.assertEquals(store.list_resources('/test'), ['test/bar.txt', |
|
208 'test/foo.txt']) |
|
209 self.assertEquals(store.list_resources('/',True), ['root.txt', |
|
210 'test/bar.txt', |
|
211 'test/foo.txt']) |
|
212 |
|
213 def test_open_resource_and_delete(self): |
|
214 store = api.Storage.open(storage_path,"w") |
|
215 self.assertTrue(store) |
|
216 res = store.open_resource("test1.txt","a") |
|
217 res.write("Testing append.\n") |
|
218 res.close() |
|
219 res = store.open_resource("test2.txt","w") |
|
220 res.write("Testing append.\n") |
|
221 res.close() |
|
222 res = store.open_resource("test3.txt","w") |
|
223 res.write("Testing append.\n") |
|
224 res.close() |
|
225 for res in store.list_resources(''): |
|
226 store.delete_resource(res) |
|
227 self.assertEquals(store.list_resources(''), []) |
|
228 |
|
229 def test_open_resources_and_read(self): |
|
230 store = api.Storage.open(storage_path,"w") |
|
231 self.assertTrue(store) |
|
232 res = store.open_resource("test1.txt","w") |
|
233 res.write("Testing reading.\n") |
|
234 res.close() |
|
235 res = store.open_resource("test2.txt","w") |
|
236 res.write("Testing reading.\n") |
|
237 res.close() |
|
238 res = store.open_resource("test3.txt","w") |
|
239 res.write("Testing reading.\n") |
|
240 res.close() |
|
241 for res in store.list_resources(''): |
|
242 res = store.open_resource(res) |
|
243 self.assertEquals(res.read(), "Testing reading.\n") |
|
244 |
|
245 def test_open_resources_and_save_and_dump(self): |
|
246 temp_file = os.path.join(temp_dir, "FooStore.pk") |
|
247 store = api.Storage.open(temp_file, "w") |
|
248 self.assertTrue(store) |
|
249 res1 = store.open_resource("test1.txt","w") |
|
250 res1.write("Testing reading.\n") |
|
251 res2 = store.open_resource("test2.txt","w") |
|
252 res2.write("Testing reading.\n") |
|
253 res3 = store.open_resource("test3.txt","w") |
|
254 res3.write("Testing reading.\n") |
|
255 store.save() |
|
256 |
|
257 def test_open_resources_and_load(self): |
|
258 temp_file = os.path.join(temp_dir, "store.pk") |
|
259 store = api.Storage.open(temp_file,"w") |
|
260 self.assertTrue(store) |
|
261 res1 = store.open_resource("test1.txt","w") |
|
262 res1.write("Testing reading.\n") |
|
263 res2 = store.open_resource("test2.txt","w") |
|
264 res2.write("Testing reading.\n") |
|
265 res3 = store.open_resource("test3.txt","w") |
|
266 res3.write("Testing reading.\n") |
|
267 store.close() |
|
268 store2 = api.Storage.open(temp_file) |
|
269 self.assertEquals(store2.list_resources(''), ['test1.txt', |
|
270 'test2.txt', |
|
271 'test3.txt']) |
|
272 self.assertEquals(store2.open_resource("test1.txt").read(),'Testing reading.\n') |
|
273 self.assertEquals(store2.open_resource("test2.txt").read(),'Testing reading.\n') |
|
274 self.assertEquals(store2.open_resource("test3.txt").read(),'Testing reading.\n') |
|
275 |
|
276 def test_import_resources(self): |
|
277 temp_file = os.path.join(temp_dir, "importsource.pk") |
|
278 store = api.Storage.open(temp_file, "w") |
|
279 self.assertTrue(store) |
|
280 res1 = store.open_resource("test1.txt","w") |
|
281 res1.write("Testing reading.\n") |
|
282 res2 = store.open_resource("test2.txt","w") |
|
283 res2.write("Testing reading.\n") |
|
284 res3 = store.open_resource("test3.txt","w") |
|
285 res3.write("Testing reading.\n") |
|
286 store.save() |
|
287 res3.close() |
|
288 store2 = api.Storage.open(temp_file, "w") |
|
289 store2.import_resources(store.list_resources(''), store) |
|
290 self.assertEquals(store2.open_resource("test1.txt").read(),store.open_resource('test1.txt').read()) |
|
291 self.assertEquals(store2.open_resource("test2.txt").read(),store.open_resource('test2.txt').read()) |
|
292 self.assertEquals(store2.open_resource("test3.txt").read(),store.open_resource('test3.txt').read()) |
|
293 store2.close() |
|
294 |
|
295 def test_export_resources(self): |
|
296 temp_file = os.path.join(temp_dir, "exportsource.pk") |
|
297 store = api.Storage.open(temp_file, "w") |
|
298 self.assertTrue(store) |
|
299 res1 = store.open_resource("test1.txt","w") |
|
300 res1.write("Testing reading.\n") |
|
301 res2 = store.open_resource("test2.txt","w") |
|
302 res2.write("Testing reading.\n") |
|
303 res3 = store.open_resource("test3.txt","w") |
|
304 res3.write("Testing reading.\n") |
|
305 store.save() |
|
306 store2 = api.Storage.open(temp_file, "w") |
|
307 store.export_resources(store.list_resources(''), store2) |
|
308 res2 = store2.open_resource("test1.txt") |
|
309 res1 = store.open_resource('test1.txt') |
|
310 self.assertEquals(res1.read(),res2.read()) |
|
311 self.assertEquals(store2.open_resource("test2.txt").read(),store.open_resource('test2.txt').read()) |
|
312 self.assertEquals(store2.open_resource("test3.txt").read(),store.open_resource('test3.txt').read()) |
|
313 store2.close() |
|
314 |
|
315 def test_export_modify_and_close_and_open(self): |
|
316 temp_file = os.path.join(temp_dir, "exportsource.pk") |
|
317 store = api.Storage.open(temp_file,"w") |
|
318 self.assertTrue(store) |
|
319 res1 = store.open_resource("test1.txt","w") |
|
320 res1.write("Testing reading.\n") |
|
321 res1.close() |
|
322 |
|
323 res2 = store.open_resource("test2.txt","w") |
|
324 res2.write("Testing reading.\n") |
|
325 res3 = store.open_resource("test3.txt","w") |
|
326 res3.write("Testing reading.\n") |
|
327 |
|
328 res1 = store.open_resource("test1.txt","w") |
|
329 res1.write("Testing reading too.\n") |
|
330 res1.close() |
|
331 |
|
332 store.close() |
|
333 |
|
334 modified_temp_file = os.path.join(temp_dir, "modified.pk") |
|
335 store2 = api.Storage.open(modified_temp_file, "w") |
|
336 store.export_resources(store.list_resources(''), store2) |
|
337 store2.delete_resource('test3.txt') |
|
338 resr = store2.open_resource('test2.txt') |
|
339 resw = store2.open_resource('test2.txt','w') |
|
340 resw.write("Now this sould be different") |
|
341 store2.close() |
|
342 store3 = api.Storage.open(modified_temp_file) |
|
343 self.assertEquals(store3.list_resources(''), ['test1.txt', |
|
344 'test2.txt']) |
|
345 self.assertEquals(store3.open_resource("test2.txt").read(),'Now this sould be different') |
|
346 self.assertEquals(store3.open_resource("test1.txt").read(),'Testing reading too.\n') |
|
347 |
|
348 |
|
349 def test_get_path_set_path(self): |
|
350 temp_file = os.path.join(temp_dir, "subpath.pk") |
|
351 store = api.Storage.open(temp_file,"w") |
|
352 self.assertEquals(store.get_path(),temp_file) |
|
353 self.assertEquals(store.get_current_path(),"") |
|
354 store.set_current_path("subdir") |
|
355 self.assertEquals(store.get_current_path(),"subdir") |
|
356 |
|
357 def test_set_path_open_resource(self): |
|
358 temp_file = os.path.join(temp_dir, "subpath.pk") |
|
359 if os.path.exists(temp_file): os.unlink(temp_file) |
|
360 |
|
361 store = api.Storage.open(temp_file,"w") |
|
362 self.assertEquals(store.get_path(),temp_file) |
|
363 store.set_current_path("subdir") |
|
364 self.assertEquals(store.get_current_path(),"subdir") |
|
365 res = store.open_resource("foo.txt","w") |
|
366 res.write("foo") |
|
367 res.close() |
|
368 self.assertEquals(store.list_resources(""), ["foo.txt"]) |
|
369 store.set_current_path("/") |
|
370 self.assertEquals(store.list_resources("", True), ["subdir/foo.txt"]) |
|
371 store.close() |
|
372 os.unlink(temp_file) |
|
373 |
|
374 def test_create_folder(self): |
|
375 temp_file = os.path.join(temp_dir, "subpath.pk") |
|
376 store = api.Storage.open(temp_file,"w") |
|
377 store.create_folder("subdir/test") |
|
378 self.assertTrue(store.is_folder("subdir/test")) |
|
379 |
|
380 def test_create_folder_and_delete_folder(self): |
|
381 temp_file = os.path.join(temp_dir, "subpath.pk") |
|
382 store = api.Storage.open(temp_file,"w") |
|
383 store.create_folder("subdir/test") |
|
384 self.assertTrue(store.is_folder("subdir/test")) |
|
385 store.delete_folder("subdir/test") |
|
386 self.assertFalse(store.is_folder("subdir/test")) |
|
387 self.assertTrue(store.is_folder("subdir")) |
|
388 |
|
389 class TestFolder(unittest.TestCase): |
|
390 |
|
391 def test_create_folder(self): |
|
392 store = api.Storage.open(storage_path,"w") |
|
393 layer = api.Folder(store, "foo") |
|
394 self.assertTrue(layer) |
|
395 |
|
396 def test_get_path(self): |
|
397 store = api.Storage.open(storage_path,"w") |
|
398 layer = api.Folder(store, "foo") |
|
399 self.assertTrue(layer) |
|
400 self.assertEquals(layer.get_current_path(),"foo") |
|
401 |
|
402 def test_open_resource(self): |
|
403 store = api.Storage.open(storage_path,"w") |
|
404 layer = api.Folder(store, "foo") |
|
405 self.assertTrue(layer) |
|
406 res = layer.open_resource("confml/test.confml","w") |
|
407 res.write("foo.conf") |
|
408 res.close() |
|
409 self.assertEquals(layer.list_resources("", True),["confml/test.confml"]) |
|
410 self.assertEquals(store.list_resources("", True),["foo/confml/test.confml"]) |
|
411 |
|
412 def test_create_two_layers_and_open_resource(self): |
|
413 store = api.Storage.open(storage_path,"w") |
|
414 foo_layer = api.Folder(store, "foo") |
|
415 bar_layer = api.Folder(store, "bar") |
|
416 res = foo_layer.open_resource("confml/test.confml","w") |
|
417 res.write("foo.conf") |
|
418 res.close() |
|
419 res = foo_layer.open_resource("root.confml","w") |
|
420 res.close() |
|
421 res = bar_layer.open_resource("confml/root.confml","w") |
|
422 res.write("foo.conf") |
|
423 res.close() |
|
424 self.assertEquals(foo_layer.list_resources("", True),['confml/test.confml', 'root.confml']) |
|
425 self.assertEquals(store.list_resources("", True),['bar/confml/root.confml','foo/confml/test.confml','foo/root.confml']) |
|
426 |
|
427 foo_layer.delete_resource("confml/test.confml") |
|
428 self.assertEquals(foo_layer.list_resources("", True),["root.confml"]) |
|
429 self.assertEquals(store.list_resources("", True),["bar/confml/root.confml","foo/root.confml"]) |
|
430 |
|
431 |
|
432 if __name__ == '__main__': |
|
433 unittest.main() |
|
434 |