--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/configurationengine/source/cone/storage/tests/unittest_zipstorage.py Thu Mar 11 17:04:37 2010 +0200
@@ -0,0 +1,255 @@
+#
+# Copyright (c) 2009 Nokia Corporation and/or its subsidiary(-ies).
+# All rights reserved.
+# This component and the accompanying materials are made available
+# under the terms of "Eclipse Public License v1.0"
+# which accompanies this distribution, and is available
+# at the URL "http://www.eclipse.org/legal/epl-v10.html".
+#
+# Initial Contributors:
+# Nokia Corporation - initial contribution.
+#
+# Contributors:
+#
+# Description:
+#
+
+
+"""
+Test the CPF root file parsing routines
+"""
+
+import zipfile
+import unittest
+import string
+import sys,os,shutil
+
+ROOT_PATH = os.path.dirname(os.path.abspath(__file__))
+
+import __init__
+from cone.public import exceptions, api
+from cone.storage import zipstorage
+
+datazip = os.path.join(ROOT_PATH,"data.zip")
+
+#class TestZipStorageDummy(unittest.TestCase):
+# def test_open_storage(self):
+# storage = zipstorage.ZipStorage("TestProlog.zip","r")
+# res = storage.open_resource("TestProlog")
+# storage.close()
+
+class TestZipStorage(unittest.TestCase):
+ def test_create_new_storage(self):
+ storage = zipstorage.ZipStorage("new.zip","w")
+ storage.close()
+ self.assertTrue(os.path.exists("new.zip"))
+ os.unlink("new.zip")
+
+ def test_open_existing_storage(self):
+ storage = zipstorage.ZipStorage(datazip,"r")
+ storage.close()
+
+ def test_open_existing_storage_and_close_twice(self):
+ storage = zipstorage.ZipStorage(datazip,"r")
+ storage.close()
+ try:
+ storage.close()
+ self.fail('closing twice succeeds!')
+ except exceptions.StorageException:
+ pass
+
+ def test_open_nonexisting_storage_fails(self):
+ try:
+ storage = zipstorage.ZipStorage("foo.zip","r")
+ storage.close()
+ self.fail("opening a non existing ZipStorage succeeded?")
+ except zipstorage.ZipException:
+ self.assertTrue(True)
+
+ def test_open_a_non_zipfile_fails(self):
+ try:
+ storage = zipstorage.ZipStorage("data/onefile/test.txt","r")
+ storage.close()
+ self.fail("opening a non zipfile for ZipStorage succeeded?")
+ except zipstorage.ZipException,e:
+ self.assertTrue(True)
+
+class TestStorage(unittest.TestCase):
+ def setUp(self):
+ shutil.copyfile(datazip,"temptests.zip")
+ self.storage = zipstorage.ZipStorage("temptests.zip","a")
+
+ def tearDown(self):
+ self.storage.close()
+ os.unlink("temptests.zip")
+
+ def test_open_resource_existing_file_for_reading(self):
+ res = self.storage.open_resource("data/simple.confml","w")
+ self.assertTrue(res)
+ self.assertTrue(isinstance(res,api.Resource))
+
+ def test_open_resource_new_file(self):
+ storage = zipstorage.ZipStorage("testnewfile.zip","w")
+ res = storage.open_resource("data/newfile.txt","w")
+ res.write("test write")
+ self.assertTrue(res)
+ self.assertTrue(isinstance(res,api.Resource))
+ res.close()
+ self.assertEquals(storage.list_resources("",True), ['data/newfile.txt'])
+ storage.close()
+ os.unlink("testnewfile.zip")
+
+
+ def test_open_resource_nonexisting(self):
+ try:
+ res = self.storage.open_resource("iamnothere.txt")
+ self.fail("Opening of a non existing file succeeds!??")
+ except exceptions.NotResource:
+ self.assertTrue(True)
+
+ def test_list_resources_nonrecurse(self):
+ storage = zipstorage.ZipStorage("testnonrecurse.zip","w")
+ res = storage.open_resource("data/morestuff.confml","w")
+ res.close()
+ res = storage.open_resource("data/prodX.confml","w")
+ res.close()
+ file_array = storage.list_resources("data")
+ self.assertEquals(file_array[0],"data/morestuff.confml")
+ self.assertEquals(file_array[1],"data/prodX.confml")
+ storage.close()
+ os.unlink("testnonrecurse.zip")
+
+ def test_list_resources_recurse(self):
+ storage = zipstorage.ZipStorage("testrecurse.zip","w")
+ res = storage.open_resource("data/foo/morestuff.confml","w")
+ res.close()
+ res = storage.open_resource("data/prodX.confml","w")
+ res.close()
+ res = storage.open_resource("data/ncp11/confml/jallaa.confml","w")
+ res.close()
+ file_array = storage.list_resources("data",True)
+ self.assertEquals(file_array,['data/foo/morestuff.confml', 'data/prodX.confml', 'data/ncp11/confml/jallaa.confml'])
+ storage.close()
+ os.unlink("testrecurse.zip")
+
+ def test_is_resource_true(self):
+ res = self.storage.open_resource("data/simple.confml","w")
+ res.close()
+ self.assertTrue(self.storage.is_resource("data/simple.confml"))
+
+ def test_is_resource_true_for_dotted_file(self):
+ res = self.storage.open_resource(".metadata","w")
+ res.close()
+ self.assertTrue(self.storage.is_resource(".metadata"))
+
+ def test_is_resource_true_with_slash(self):
+ res = self.storage.open_resource("data/simple.confml","w")
+ res.close()
+ self.assertTrue(self.storage.is_resource("/data/simple.confml"))
+
+ def test_is_resource_false(self):
+ self.assertFalse(self.storage.is_resource("data"))
+
+ def test_metadata_writing(self):
+ fs = zipstorage.ZipStorage("testtemp.zip","w")
+ fs.set_active_configuration('testing.confml')
+ fs.close()
+ fs = zipstorage.ZipStorage("testtemp.zip","r")
+ self.assertEquals(fs.get_active_configuration(),'testing.confml')
+ fs.close()
+ os.unlink("testtemp.zip")
+
+ def test_open_resource_new_file_and_overwrite(self):
+ storage = api.Storage.open("testoverwrite.zip","w")
+ res = storage.open_resource("data/newfile.txt","w")
+ res.write("test write")
+ self.assertTrue(res)
+ self.assertTrue(isinstance(res,api.Resource))
+ res.close()
+ res = storage.open_resource("data/newfile.txt","w")
+ res.write("Hahaaa")
+ res.close()
+ storage.close()
+ storage = api.Storage.open("testoverwrite.zip","r")
+ self.assertEquals(storage.open_resource("data/newfile.txt").read(), "Hahaaa")
+ storage.close()
+ os.unlink("testoverwrite.zip")
+
+ def test_delete_resource(self):
+ storage = api.Storage.open("testdelete.zip","w")
+ res = storage.open_resource("data/newfile.txt","w")
+ res.write("test write")
+ res.close()
+ res = storage.open_resource("readme.txt","w")
+ res.write("test 2")
+ res.close()
+ storage.close()
+ storage2 = api.Storage.open("testdelete.zip","a")
+ #self.assertEquals(storage2.list_resources("",True), ['.metadata', 'data/newfile.txt', 'readme.txt'])
+ self.assertEquals(storage2.open_resource("data/newfile.txt").read(),"test write")
+ storage2.delete_resource("data/newfile.txt")
+ self.assertEquals(len(storage2.list_resources("",True)),2)
+ storage2.close()
+ storage3 = api.Storage.open("testdelete.zip","a")
+ self.assertEquals(storage3.list_resources("",True), ['readme.txt','.metadata'])
+ storage3.close()
+ os.unlink("testdelete.zip")
+
+
+ def test_create_folder(self):
+ storage = api.Storage.open("empty_folder.zip","w")
+ res = storage.open_resource("test.txt","w")
+ res.write('test')
+ res.close()
+ storage.create_folder("data")
+ storage.create_folder("data2/folder1")
+ storage.create_folder("data3\\")
+ storage.close()
+
+ storage2 = api.Storage.open("empty_folder.zip","a")
+ self.assertEquals(storage2.is_folder("data"),True)
+ self.assertEquals(storage2.is_folder("data2/folder1"),True)
+ self.assertEquals(storage2.is_folder("data3"),True)
+ self.assertEquals(storage2.list_resources('.',True),['test.txt','.metadata'])
+ self.assertEquals(storage2.list_resources(''),['test.txt','.metadata'])
+ storage2.close()
+ os.unlink("empty_folder.zip")
+
+class TestZipStorageListResources(unittest.TestCase):
+
+ def _run_test_list_resources(self, zip_file):
+ full_path = os.path.join(ROOT_PATH, 'list_resources_data', zip_file)
+ zs = zipstorage.ZipStorage(full_path, 'r')
+ res_list = zs.list_resources('/', recurse=True, empty_folders=True)
+
+ expected = [
+ ('folder', 'test'),
+ ('folder', 'test/layer'),
+ ('folder', 'test/layer/confml'),
+ ('folder', 'test/layer/content'),
+ ('folder', 'test/layer/content/empty'),
+ ('folder', 'test/layer/content/something'),
+ ('resource', 'test/layer/content/something/x.txt'),
+ ('folder', 'test/layer/doc'),
+ ('folder', 'test/layer/implml'),
+ ('resource', 'test/layer/root.confml'),
+ ('resource', 'test/root.confml')]
+ for type, res in expected:
+ if type == 'resource':
+ self.assertTrue(zs.is_resource(res), "zs.is_resource('%s') returns False" % res)
+ elif type == 'folder':
+ self.assertTrue(zs.is_folder(res), "zs.is_folder('%s') returns False" % res)
+ else:
+ raise RuntimeError('Invalid type')
+
+ def test_list_resources_7zip_zipped(self):
+ self._run_test_list_resources('7zip.zip')
+
+ def test_list_resources_winzip_zipped(self):
+ self._run_test_list_resources('winzip.zip')
+
+ def test_list_resources_carbide_ct_zipped(self):
+ self._run_test_list_resources('carbide.ct.cpf')
+
+if __name__ == '__main__':
+ unittest.main()