# 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
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']
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)
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)
projects / mediagoblin.git / blobdiff