diff -r 000000000000 -r 2e8eeb919028 configurationengine/source/cone/storage/tests/unittest_zipstorage.py --- /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()