0
|
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 |
"""
|
|
19 |
Test the CPF root file parsing routines
|
|
20 |
"""
|
|
21 |
|
|
22 |
import unittest
|
3
|
23 |
import os,shutil
|
0
|
24 |
|
|
25 |
ROOT_PATH = os.path.dirname(os.path.abspath(__file__))
|
|
26 |
|
|
27 |
from cone.public import exceptions, api
|
|
28 |
from cone.storage import zipstorage
|
|
29 |
|
|
30 |
datazip = os.path.join(ROOT_PATH,"data.zip")
|
|
31 |
|
|
32 |
#class TestZipStorageDummy(unittest.TestCase):
|
|
33 |
# def test_open_storage(self):
|
|
34 |
# storage = zipstorage.ZipStorage("TestProlog.zip","r")
|
|
35 |
# res = storage.open_resource("TestProlog")
|
|
36 |
# storage.close()
|
|
37 |
|
|
38 |
class TestZipStorage(unittest.TestCase):
|
|
39 |
def test_create_new_storage(self):
|
|
40 |
storage = zipstorage.ZipStorage("new.zip","w")
|
|
41 |
storage.close()
|
|
42 |
self.assertTrue(os.path.exists("new.zip"))
|
|
43 |
os.unlink("new.zip")
|
|
44 |
|
|
45 |
def test_open_existing_storage(self):
|
|
46 |
storage = zipstorage.ZipStorage(datazip,"r")
|
|
47 |
storage.close()
|
|
48 |
|
|
49 |
def test_open_existing_storage_and_close_twice(self):
|
|
50 |
storage = zipstorage.ZipStorage(datazip,"r")
|
|
51 |
storage.close()
|
|
52 |
try:
|
|
53 |
storage.close()
|
|
54 |
self.fail('closing twice succeeds!')
|
|
55 |
except exceptions.StorageException:
|
|
56 |
pass
|
|
57 |
|
|
58 |
def test_open_nonexisting_storage_fails(self):
|
|
59 |
try:
|
|
60 |
storage = zipstorage.ZipStorage("foo.zip","r")
|
|
61 |
storage.close()
|
|
62 |
self.fail("opening a non existing ZipStorage succeeded?")
|
|
63 |
except zipstorage.ZipException:
|
|
64 |
self.assertTrue(True)
|
|
65 |
|
|
66 |
def test_open_a_non_zipfile_fails(self):
|
|
67 |
try:
|
|
68 |
storage = zipstorage.ZipStorage("data/onefile/test.txt","r")
|
|
69 |
storage.close()
|
|
70 |
self.fail("opening a non zipfile for ZipStorage succeeded?")
|
|
71 |
except zipstorage.ZipException,e:
|
|
72 |
self.assertTrue(True)
|
|
73 |
|
|
74 |
class TestStorage(unittest.TestCase):
|
|
75 |
def setUp(self):
|
|
76 |
shutil.copyfile(datazip,"temptests.zip")
|
|
77 |
self.storage = zipstorage.ZipStorage("temptests.zip","a")
|
|
78 |
|
|
79 |
def tearDown(self):
|
|
80 |
self.storage.close()
|
|
81 |
os.unlink("temptests.zip")
|
|
82 |
|
|
83 |
def test_open_resource_existing_file_for_reading(self):
|
|
84 |
res = self.storage.open_resource("data/simple.confml","w")
|
|
85 |
self.assertTrue(res)
|
|
86 |
self.assertTrue(isinstance(res,api.Resource))
|
|
87 |
|
|
88 |
def test_open_resource_new_file(self):
|
|
89 |
storage = zipstorage.ZipStorage("testnewfile.zip","w")
|
|
90 |
res = storage.open_resource("data/newfile.txt","w")
|
|
91 |
res.write("test write")
|
|
92 |
self.assertTrue(res)
|
|
93 |
self.assertTrue(isinstance(res,api.Resource))
|
|
94 |
res.close()
|
3
|
95 |
self.assertEquals(storage.list_resources("",recurse=True), ['data/newfile.txt'])
|
0
|
96 |
storage.close()
|
|
97 |
os.unlink("testnewfile.zip")
|
|
98 |
|
|
99 |
|
|
100 |
def test_open_resource_nonexisting(self):
|
|
101 |
try:
|
|
102 |
res = self.storage.open_resource("iamnothere.txt")
|
|
103 |
self.fail("Opening of a non existing file succeeds!??")
|
|
104 |
except exceptions.NotResource:
|
|
105 |
self.assertTrue(True)
|
|
106 |
|
|
107 |
def test_list_resources_nonrecurse(self):
|
|
108 |
storage = zipstorage.ZipStorage("testnonrecurse.zip","w")
|
|
109 |
res = storage.open_resource("data/morestuff.confml","w")
|
|
110 |
res.close()
|
|
111 |
res = storage.open_resource("data/prodX.confml","w")
|
|
112 |
res.close()
|
|
113 |
file_array = storage.list_resources("data")
|
|
114 |
self.assertEquals(file_array[0],"data/morestuff.confml")
|
|
115 |
self.assertEquals(file_array[1],"data/prodX.confml")
|
|
116 |
storage.close()
|
|
117 |
os.unlink("testnonrecurse.zip")
|
|
118 |
|
|
119 |
def test_list_resources_recurse(self):
|
|
120 |
storage = zipstorage.ZipStorage("testrecurse.zip","w")
|
|
121 |
res = storage.open_resource("data/foo/morestuff.confml","w")
|
|
122 |
res.close()
|
|
123 |
res = storage.open_resource("data/prodX.confml","w")
|
|
124 |
res.close()
|
|
125 |
res = storage.open_resource("data/ncp11/confml/jallaa.confml","w")
|
|
126 |
res.close()
|
3
|
127 |
file_array = storage.list_resources("data",recurse=True)
|
0
|
128 |
self.assertEquals(file_array,['data/foo/morestuff.confml', 'data/prodX.confml', 'data/ncp11/confml/jallaa.confml'])
|
|
129 |
storage.close()
|
|
130 |
os.unlink("testrecurse.zip")
|
|
131 |
|
|
132 |
def test_is_resource_true(self):
|
|
133 |
res = self.storage.open_resource("data/simple.confml","w")
|
|
134 |
res.close()
|
|
135 |
self.assertTrue(self.storage.is_resource("data/simple.confml"))
|
|
136 |
|
|
137 |
def test_is_resource_true_for_dotted_file(self):
|
|
138 |
res = self.storage.open_resource(".metadata","w")
|
|
139 |
res.close()
|
|
140 |
self.assertTrue(self.storage.is_resource(".metadata"))
|
|
141 |
|
|
142 |
def test_is_resource_true_with_slash(self):
|
|
143 |
res = self.storage.open_resource("data/simple.confml","w")
|
|
144 |
res.close()
|
|
145 |
self.assertTrue(self.storage.is_resource("/data/simple.confml"))
|
|
146 |
|
|
147 |
def test_is_resource_false(self):
|
|
148 |
self.assertFalse(self.storage.is_resource("data"))
|
|
149 |
|
|
150 |
def test_metadata_writing(self):
|
|
151 |
fs = zipstorage.ZipStorage("testtemp.zip","w")
|
|
152 |
fs.set_active_configuration('testing.confml')
|
|
153 |
fs.close()
|
|
154 |
fs = zipstorage.ZipStorage("testtemp.zip","r")
|
|
155 |
self.assertEquals(fs.get_active_configuration(),'testing.confml')
|
|
156 |
fs.close()
|
|
157 |
os.unlink("testtemp.zip")
|
|
158 |
|
|
159 |
def test_open_resource_new_file_and_overwrite(self):
|
|
160 |
storage = api.Storage.open("testoverwrite.zip","w")
|
|
161 |
res = storage.open_resource("data/newfile.txt","w")
|
|
162 |
res.write("test write")
|
|
163 |
self.assertTrue(res)
|
|
164 |
self.assertTrue(isinstance(res,api.Resource))
|
|
165 |
res.close()
|
|
166 |
res = storage.open_resource("data/newfile.txt","w")
|
|
167 |
res.write("Hahaaa")
|
|
168 |
res.close()
|
|
169 |
storage.close()
|
|
170 |
storage = api.Storage.open("testoverwrite.zip","r")
|
|
171 |
self.assertEquals(storage.open_resource("data/newfile.txt").read(), "Hahaaa")
|
|
172 |
storage.close()
|
|
173 |
os.unlink("testoverwrite.zip")
|
|
174 |
|
|
175 |
def test_delete_resource(self):
|
|
176 |
storage = api.Storage.open("testdelete.zip","w")
|
|
177 |
res = storage.open_resource("data/newfile.txt","w")
|
|
178 |
res.write("test write")
|
|
179 |
res.close()
|
|
180 |
res = storage.open_resource("readme.txt","w")
|
|
181 |
res.write("test 2")
|
|
182 |
res.close()
|
|
183 |
storage.close()
|
|
184 |
storage2 = api.Storage.open("testdelete.zip","a")
|
3
|
185 |
#self.assertEquals(storage2.list_resources("",recurse=True), ['.metadata', 'data/newfile.txt', 'readme.txt'])
|
0
|
186 |
self.assertEquals(storage2.open_resource("data/newfile.txt").read(),"test write")
|
|
187 |
storage2.delete_resource("data/newfile.txt")
|
3
|
188 |
self.assertEquals(len(storage2.list_resources("",recurse=True)),2)
|
0
|
189 |
storage2.close()
|
|
190 |
storage3 = api.Storage.open("testdelete.zip","a")
|
3
|
191 |
self.assertEquals(storage3.list_resources("",recurse=True), ['readme.txt','.metadata'])
|
0
|
192 |
storage3.close()
|
|
193 |
os.unlink("testdelete.zip")
|
|
194 |
|
|
195 |
|
|
196 |
def test_create_folder(self):
|
|
197 |
storage = api.Storage.open("empty_folder.zip","w")
|
|
198 |
res = storage.open_resource("test.txt","w")
|
|
199 |
res.write('test')
|
|
200 |
res.close()
|
|
201 |
storage.create_folder("data")
|
|
202 |
storage.create_folder("data2/folder1")
|
|
203 |
storage.create_folder("data3\\")
|
|
204 |
storage.close()
|
|
205 |
|
|
206 |
storage2 = api.Storage.open("empty_folder.zip","a")
|
|
207 |
self.assertEquals(storage2.is_folder("data"),True)
|
|
208 |
self.assertEquals(storage2.is_folder("data2/folder1"),True)
|
|
209 |
self.assertEquals(storage2.is_folder("data3"),True)
|
3
|
210 |
self.assertEquals(storage2.list_resources('.',recurse=True),['test.txt','.metadata'])
|
0
|
211 |
self.assertEquals(storage2.list_resources(''),['test.txt','.metadata'])
|
|
212 |
storage2.close()
|
|
213 |
os.unlink("empty_folder.zip")
|
|
214 |
|
|
215 |
class TestZipStorageListResources(unittest.TestCase):
|
|
216 |
|
|
217 |
def _run_test_list_resources(self, zip_file):
|
|
218 |
full_path = os.path.join(ROOT_PATH, 'list_resources_data', zip_file)
|
|
219 |
zs = zipstorage.ZipStorage(full_path, 'r')
|
|
220 |
res_list = zs.list_resources('/', recurse=True, empty_folders=True)
|
|
221 |
|
|
222 |
expected = [
|
|
223 |
('folder', 'test'),
|
|
224 |
('folder', 'test/layer'),
|
|
225 |
('folder', 'test/layer/confml'),
|
|
226 |
('folder', 'test/layer/content'),
|
|
227 |
('folder', 'test/layer/content/empty'),
|
|
228 |
('folder', 'test/layer/content/something'),
|
|
229 |
('resource', 'test/layer/content/something/x.txt'),
|
|
230 |
('folder', 'test/layer/doc'),
|
|
231 |
('folder', 'test/layer/implml'),
|
|
232 |
('resource', 'test/layer/root.confml'),
|
|
233 |
('resource', 'test/root.confml')]
|
|
234 |
for type, res in expected:
|
|
235 |
if type == 'resource':
|
|
236 |
self.assertTrue(zs.is_resource(res), "zs.is_resource('%s') returns False" % res)
|
|
237 |
elif type == 'folder':
|
|
238 |
self.assertTrue(zs.is_folder(res), "zs.is_folder('%s') returns False" % res)
|
|
239 |
else:
|
|
240 |
raise RuntimeError('Invalid type')
|
|
241 |
|
|
242 |
def test_list_resources_7zip_zipped(self):
|
|
243 |
self._run_test_list_resources('7zip.zip')
|
|
244 |
|
|
245 |
def test_list_resources_winzip_zipped(self):
|
|
246 |
self._run_test_list_resources('winzip.zip')
|
|
247 |
|
|
248 |
def test_list_resources_carbide_ct_zipped(self):
|
|
249 |
self._run_test_list_resources('carbide.ct.cpf')
|
|
250 |
|
|
251 |
if __name__ == '__main__':
|
|
252 |
unittest.main()
|