X-Git-Url: https://vcs.fsf.org/?a=blobdiff_plain;f=mediagoblin%2Ftests%2Ftest_workbench.py;h=f3ff57ed669687f0f8d4ef179cd8bcb73da7b0c8;hb=9c602458d830c1ebcd0ede7c14f7ddef79e2a73f;hp=89f2ef337cafb1e1e0ec7727e034120eefa09c4e;hpb=da579dec005834f92fb7d968e14610be9936d9a5;p=mediagoblin.git diff --git a/mediagoblin/tests/test_workbench.py b/mediagoblin/tests/test_workbench.py index 89f2ef33..f3ff57ed 100644 --- a/mediagoblin/tests/test_workbench.py +++ b/mediagoblin/tests/test_workbench.py @@ -1,5 +1,5 @@ # GNU MediaGoblin -- federated, autonomous media hosting -# Copyright (C) 2011 Free Software Foundation, Inc +# Copyright (C) 2011, 2012 MediaGoblin contributors. See AUTHORS. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by @@ -17,47 +17,54 @@ import os import tempfile -from nose.tools import assert_raises -from mediagoblin import workbench -from mediagoblin.tests.test_storage import get_tmp_filestorage +from mediagoblin.tools import workbench +from mediagoblin.mg_globals import setup_globals +from mediagoblin.decorators import get_workbench +from mediagoblin.tests.test_storage import get_tmp_filestorage, cleanup_storage class TestWorkbench(object): - def setUp(self): + def setup(self): + self.workbench_base = tempfile.mkdtemp(prefix='gmg_workbench_testing') self.workbench_manager = workbench.WorkbenchManager( - os.path.join(tempfile.gettempdir(), u'mgoblin_workbench_testing')) + self.workbench_base) + + def teardown(self): + # If the workbench is empty, this should work. + os.rmdir(self.workbench_base) def test_create_workbench(self): - workbench = self.workbench_manager.create_workbench() - assert os.path.isdir(workbench) - assert workbench.startswith(self.workbench_manager.base_workbench_dir) + workbench = self.workbench_manager.create() + assert os.path.isdir(workbench.dir) + assert workbench.dir.startswith(self.workbench_manager.base_workbench_dir) + workbench.destroy() + + def test_joinpath(self): + this_workbench = self.workbench_manager.create() + tmpname = this_workbench.joinpath('temp.txt') + assert tmpname == os.path.join(this_workbench.dir, 'temp.txt') + this_workbench.destroy() def test_destroy_workbench(self): # kill a workbench - this_workbench = self.workbench_manager.create_workbench() - tmpfile = file(os.path.join(this_workbench, 'temp.txt'), 'w') + this_workbench = self.workbench_manager.create() + tmpfile_name = this_workbench.joinpath('temp.txt') + tmpfile = open(tmpfile_name, 'w') with tmpfile: tmpfile.write('lollerskates') - assert os.path.exists(os.path.join(this_workbench, 'temp.txt')) - - self.workbench_manager.destroy_workbench(this_workbench) - assert not os.path.exists(os.path.join(this_workbench, 'temp.txt')) - assert not os.path.exists(this_workbench) + assert os.path.exists(tmpfile_name) - # make sure we can't kill other stuff though - dont_kill_this = tempfile.mkdtemp() - - assert_raises( - workbench.WorkbenchOutsideScope, - self.workbench_manager.destroy_workbench, - dont_kill_this) + wb_dir = this_workbench.dir + this_workbench.destroy() + assert not os.path.exists(tmpfile_name) + assert not os.path.exists(wb_dir) def test_localized_file(self): tmpdir, this_storage = get_tmp_filestorage() - this_workbench = self.workbench_manager.create_workbench() - + this_workbench = self.workbench_manager.create() + # Write a brand new file filepath = ['dir1', 'dir2', 'ourfile.txt'] @@ -65,10 +72,11 @@ class TestWorkbench(object): our_file.write('Our file') # with a local file storage - filename = self.workbench_manager.localized_file( - this_workbench, this_storage, filepath) + filename = this_workbench.localized_file(this_storage, filepath) assert filename == os.path.join( tmpdir, 'dir1/dir2/ourfile.txt') + this_storage.delete_file(filepath) + cleanup_storage(this_storage, tmpdir, ['dir1', 'dir2']) # with a fake remote file storage tmpdir, this_storage = get_tmp_filestorage(fake_remote=True) @@ -77,20 +85,38 @@ class TestWorkbench(object): with this_storage.get_file(filepath, 'w') as our_file: our_file.write('Our file') - filename = self.workbench_manager.localized_file( - this_workbench, this_storage, filepath) + filename = this_workbench.localized_file(this_storage, filepath) assert filename == os.path.join( - this_workbench, 'ourfile.txt') - + this_workbench.dir, 'ourfile.txt') + # fake remote file storage, filename_if_copying set - filename = self.workbench_manager.localized_file( - this_workbench, this_storage, filepath, 'thisfile') + filename = this_workbench.localized_file( + this_storage, filepath, 'thisfile') assert filename == os.path.join( - this_workbench, 'thisfile.txt') + this_workbench.dir, 'thisfile.txt') # fake remote file storage, filename_if_copying set, # keep_extension_if_copying set to false - filename = self.workbench_manager.localized_file( - this_workbench, this_storage, filepath, 'thisfile.text', False) + filename = this_workbench.localized_file( + this_storage, filepath, 'thisfile.text', False) assert filename == os.path.join( - this_workbench, 'thisfile.text') + this_workbench.dir, 'thisfile.text') + + this_storage.delete_file(filepath) + cleanup_storage(this_storage, tmpdir, ['dir1', 'dir2']) + this_workbench.destroy() + + def test_workbench_decorator(self): + """Test @get_workbench decorator and automatic cleanup""" + # The decorator needs mg_globals.workbench_manager + setup_globals(workbench_manager=self.workbench_manager) + + @get_workbench + def create_it(workbench=None): + # workbench dir exists? + assert os.path.isdir(workbench.dir) + return workbench.dir + + benchdir = create_it() + # workbench dir has been cleaned up automatically? + assert not os.path.isdir(benchdir)