self.useFixture(lockutils.ExternalLockFixture())
- self.temp_dir = self.useFixture(fixtures.TempDir()).path
- cfg.CONF.set_override('state_path', self.temp_dir)
+ cfg.CONF.set_override('state_path', self.get_default_temp_dir().path)
self.addCleanup(CONF.reset)
self.setup_rpc_mocks()
self.setup_config()
+ def get_new_temp_dir(self):
+ """Create a new temporary directory.
+
+ :returns fixtures.TempDir
+ """
+ return self.useFixture(fixtures.TempDir())
+
+ def get_default_temp_dir(self):
+ """Create a default temporary directory.
+
+ Returns the same directory during the whole test case.
+
+ :returns fixtures.TempDir
+ """
+ if not hasattr(self, '_temp_dir'):
+ self._temp_dir = self.get_new_temp_dir()
+ return self._temp_dir
+
+ def get_temp_file_path(self, filename, root=None):
+ """Returns an absolute path for a temporary file.
+
+ If root is None, the file is created in default temporary directory. It
+ also creates the directory if it's not initialized yet.
+
+ If root is not None, the file is created inside the directory passed as
+ root= argument.
+
+ :param filename: filename
+ :type filename: string
+ :param root: temporary directory to create a new file in
+ :type root: fixtures.TempDir
+ :returns absolute file path string
+ """
+ root = root or self.get_default_temp_dir()
+ return root.join(filename)
+
def setup_rpc_mocks(self):
# don't actually start RPC listeners when testing
self.useFixture(fixtures.MonkeyPatch(
# under the License.
import eventlet
-import fixtures
from six import moves
def setUp(self):
super(TestAsyncProcess, self).setUp()
- self.test_file_path = self.useFixture(
- fixtures.TempDir()).join("test_async_process.tmp")
+ self.test_file_path = self.get_temp_file_path('test_async_process.tmp')
self.data = [str(x) for x in moves.xrange(4)]
with file(self.test_file_path, 'w') as f:
f.writelines('%s\n' % item for item in self.data)
import copy
import functools
-import fixtures
import mock
import netaddr
from oslo.config import cfg
conf.set_override('ovs_integration_bridge', br_int.br_name)
conf.set_override('external_network_bridge', br_ex.br_name)
- temp_dir = self.useFixture(fixtures.TempDir()).path
- conf.set_override('state_path', temp_dir)
+ temp_dir = self.get_new_temp_dir()
+ get_temp_file_path = functools.partial(self.get_temp_file_path,
+ root=temp_dir)
+ conf.set_override('state_path', temp_dir.path)
conf.set_override('metadata_proxy_socket',
- '%s/metadata_proxy' % temp_dir)
+ get_temp_file_path('metadata_proxy'))
conf.set_override('ha_confs_path',
- '%s/ha_confs' % temp_dir)
+ get_temp_file_path('ha_confs'))
conf.set_override('external_pids',
- '%s/external/pids' % temp_dir)
+ get_temp_file_path('external/pids'))
conf.set_override('host', host)
agent = l3_test_agent.TestL3NATAgent(host, conf)
mock.patch.object(ip_lib, 'send_gratuitous_arp').start()
# See the License for the specific language governing permissions and
# limitations under the License.
-import os
-
-import fixtures
import mock
import webob.exc
_nec_ini = NEC_PLUGIN_INI
def _set_nec_ini(self):
- self.nec_ini_file = self.useFixture(fixtures.TempDir()).join("nec.ini")
+ self.nec_ini_file = self.get_temp_file_path('nec.ini')
with open(self.nec_ini_file, 'w') as f:
f.write(self._nec_ini)
if 'config_files' in test_lib.test_config.keys():
test_lib.test_config['config_files'].append(self.nec_ini_file)
else:
test_lib.test_config['config_files'] = [self.nec_ini_file]
+ self.addCleanup(self._clean_nec_ini)
def _clean_nec_ini(self):
test_lib.test_config['config_files'].remove(self.nec_ini_file)
- os.remove(self.nec_ini_file)
self.nec_ini_file = None
def patch_remote_calls(self):
def setup_nec_plugin_base(self):
self._set_nec_ini()
- self.addCleanup(self._clean_nec_ini)
self.patch_remote_calls()
def setUp(self, plugin=None, ext_mgr=None):
self._set_nec_ini()
- self.addCleanup(self._clean_nec_ini)
plugin = plugin or self._plugin_name
super(NecPluginV2TestCase, self).setUp(plugin, ext_mgr=ext_mgr)
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
-
-import fixtures
import mock
import testtools
def setUp(self):
super(AgentUtilsExecuteTest, self).setUp()
self.root_helper = "echo"
- self.test_file = self.useFixture(
- fixtures.TempDir()).join("test_execute.tmp")
+ self.test_file = self.get_temp_file_path('test_execute.tmp')
open(self.test_file, 'w').close()
self.mock_popen_p = mock.patch("subprocess.Popen.communicate")
self.mock_popen = self.mock_popen_p.start()
import StringIO
import urllib2
-import fixtures
import mock
from oslo.config import cfg
from oslo.serialization import jsonutils
self.addCleanup(policy.reset)
self.context = context.Context('fake', 'fake', is_admin=False)
self.target = {'tenant_id': 'fake'}
- self.tempdir = self.useFixture(fixtures.TempDir())
def test_modified_policy_reloads(self):
- tmpfilename = self.tempdir.join('policy')
+ tmpfilename = self.get_temp_file_path('policy')
action = "example:test"
with open(tmpfilename, "w") as policyfile:
policyfile.write("""{"example:test": ""}""")
def setUp(self):
super(DefaultPolicyTestCase, self).setUp()
- self.tempdir = self.useFixture(fixtures.TempDir())
- tmpfilename = self.tempdir.join('policy.json')
+ tmpfilename = self.get_temp_file_path('policy.json')
self.rules = {
"default": '',
"example:exist": '!',