我们从Python开源项目中，提取了以下49个代码示例，用于说明如何使用fixtures.TempDir()。
def test_cannot_listen(self): """ When the program is run with an argument and a listen address specified with a port that we can't listen on (e.g. port 1), a CannotListenError is expected to be logged and the program should stop. """ temp_dir = self.useFixture(TempDir()) yield main(reactor, raw_args=[ temp_dir.path, '--listen', ':1', # A port we can't listen on ]) # Expect a 'certs' directory to be created self.assertThat(os.path.isdir(temp_dir.join('certs')), Equals(True)) # Expect a default certificate to be created self.assertThat(os.path.isfile(temp_dir.join('default.pem')), Equals(True)) # Expect to be unable to listen flush_logged_errors(CannotListenError)
def setUp(self): super(SkipFileWrites, self).setUp() self.temp_path = self.useFixture(fixtures.TempDir()).path self.root_dir = os.path.abspath(os.path.curdir) self.git_dir = os.path.join(self.root_dir, ".git") if not os.path.exists(self.git_dir): self.skipTest("%s is missing; skipping git-related checks" % self.git_dir) return self.filename = os.path.join(self.temp_path, self.filename) self.option_dict = dict() if self.option_key is not None: self.option_dict[self.option_key] = ('setup.cfg', self.option_value) self.useFixture( fixtures.EnvironmentVariable(self.env_key, self.env_value))
def _setUp(self): super(RealPolicyFixture, self)._setUp() self.policy_dir = self.useFixture(fixtures.TempDir()) self.policy_file = os.path.join(self.policy_dir.path, 'policy.yaml') # Load the fake_policy data and add the missing default rules. policy_rules = yaml.safe_load(fake_policy.policy_data) self.add_missing_default_rules(policy_rules) with open(self.policy_file, 'w') as f: yaml.safe_dump(policy_rules, f) policy_opts.set_defaults(CONF) self.useFixture( ConfPatcher(policy_dirs=[], policy_file=self.policy_file, group='oslo_policy')) deckhand.policy.reset() deckhand.policy.init() self.addCleanup(deckhand.policy.reset) if self.verify: self._install_policy_verification_hook()
def setUp(self): super(HookCfnInitTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-cfn-init/install.d/hook-cfn-init.py') self.fake_tool_path = self.relative_path( __file__, 'config-tool-fake.py') self.metadata_dir = self.useFixture(fixtures.TempDir()) # use the temp dir to store the fake config tool state too self.test_state_path = self.metadata_dir.join('test_state.json') self.env = os.environ.copy() self.env.update({ 'HEAT_CFN_INIT_LAST_METADATA_DIR': self.metadata_dir.join(), 'HEAT_CFN_INIT_CMD': self.fake_tool_path, 'TEST_STATE_PATH': self.test_state_path, })
def setUp(self): super(HeatConfigTest, self).setUp() self.fake_hook_path = self.relative_path(__file__, 'hook-fake.py') self.heat_config_path = self.relative_path( __file__, '..', 'heat-config/os-refresh-config/configure.d/55-heat-config') self.hooks_dir = self.useFixture(fixtures.TempDir()) self.deployed_dir = self.useFixture(fixtures.TempDir()) with open(self.fake_hook_path) as f: fake_hook = f.read() for hook in self.fake_hooks: hook_name = self.hooks_dir.join(hook) with open(hook_name, 'w') as f: os.utime(hook_name, None) f.write(fake_hook) f.flush() os.chmod(hook_name, 0o755) self.env = os.environ.copy()
def setUp(self): super(HookJsonFileConfigTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-json-file/install.d/hook-json-file.py') conf_dir = self.useFixture(fixtures.TempDir()).join() self.conf = tempfile.NamedTemporaryFile( mode='w', dir=conf_dir, delete=False).name os.unlink(self.conf) self.env = os.environ.copy() self.data = { 'id': 'test_json_file', 'name': 'fake_resource_name', 'group': 'json-file', 'config': { self.conf: { 'command': 'foo' } } }
def setUp(self): super(HookHieraTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-hiera/install.d/hook-hiera.py') self.hieradata_dir = self.useFixture(fixtures.TempDir()).join() conf_dir = self.useFixture(fixtures.TempDir()).join() self.conf = tempfile.NamedTemporaryFile( dir=conf_dir, mode='w', delete=False).name os.unlink(self.conf) self.env = os.environ.copy() self.env.update({ 'HEAT_HIERA_CONFIG': self.conf, 'HEAT_PUPPET_HIERA_DATADIR': self.hieradata_dir, })
def setUp(self): super(HeatConfigDockerComposeORCTest, self).setUp() self.fake_hook_path = self.relative_path(__file__, 'hook-fake.py') self.heat_config_docker_compose_path = self.relative_path( __file__, '..', 'heat-config-docker-compose/os-refresh-config/configure.d/' '50-heat-config-docker-compose') self.docker_compose_dir = self.useFixture(fixtures.TempDir()) with open(self.fake_hook_path) as f: fake_hook = f.read() for hook in self.fake_hooks: hook_name = self.docker_compose_dir.join(hook) with open(hook_name, 'w') as f: os.utime(hook_name, None) f.write(fake_hook) f.flush() os.chmod(hook_name, 0o755)
def setUp(self): super(HeatConfigKubeletORCTest, self).setUp() self.fake_hook_path = self.relative_path(__file__, 'hook-fake.py') self.heat_config_kubelet_path = self.relative_path( __file__, '..', 'heat-config-kubelet/os-refresh-config/configure.d/' '50-heat-config-kubelet') self.manifests_dir = self.useFixture(fixtures.TempDir()) with open(self.fake_hook_path) as f: fake_hook = f.read() for hook in self.fake_hooks: hook_name = self.manifests_dir.join(hook) with open(hook_name, 'w') as f: os.utime(hook_name, None) f.write(fake_hook) f.flush() os.chmod(hook_name, 0o755)
def setUp(self): super(HookAnsibleTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-ansible/install.d/hook-ansible.py') self.fake_tool_path = self.relative_path( __file__, 'config-tool-fake.py') self.working_dir = self.useFixture(fixtures.TempDir()) self.outputs_dir = self.useFixture(fixtures.TempDir()) self.test_state_path = self.outputs_dir.join('test_state.json') self.test_inventory = "localhost test_var=123," self.env = os.environ.copy() self.env.update({ 'HEAT_ANSIBLE_WORKING': self.working_dir.join(), 'HEAT_ANSIBLE_OUTPUTS': self.outputs_dir.join(), 'HEAT_ANSIBLE_CMD': self.fake_tool_path, 'TEST_STATE_PATH': self.test_state_path })
def setUp(self): super(HookSaltTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-salt/install.d/hook-salt.py') self.working_dir = self.useFixture(fixtures.TempDir()) self.minion_config_dir = self.useFixture(fixtures.TempDir()) self.minion_cach_dir = self.useFixture(fixtures.TempDir()) self.minion_conf = self.minion_config_dir.join("minion") self.env = os.environ.copy() self.env.update({ 'HEAT_SALT_WORKING': self.working_dir.join(), 'SALT_MINION_CONFIG': self.minion_conf }) with open(self.minion_conf, "w+") as conf_file: conf_file.write("cachedir: %s\n" % self.minion_cach_dir.join()) conf_file.write("log_level: DEBUG\n")
def _setUp(self): tmpdir = self.useFixture(fixtures.TempDir()).path package_dirs = {} for pkg_name in self.packages: pkg_path = os.path.join(tmpdir, pkg_name) package_dirs[pkg_name] = pkg_path os.mkdir(pkg_path) for cf in ['setup.py', 'setup.cfg']: if cf in self.packages[pkg_name]: contents = self.packages[pkg_name].pop(cf) else: contents = self.defaults[cf].format(pkg_name=pkg_name) self._writeFile(pkg_path, cf, contents) for cf in self.packages[pkg_name]: self._writeFile(pkg_path, cf, self.packages[pkg_name][cf]) self.useFixture(TestRepo(pkg_path)).commit() self.addCleanup(delattr, self, 'package_dirs') self.package_dirs = package_dirs return package_dirs
def test_main_handle_ini_options(self): # Test that bandit handles cmdline args from a bandit.yaml file temp_directory = self.useFixture(fixtures.TempDir()).path os.chdir(temp_directory) with open('bandit.yaml', 'wt') as fd: fd.write(bandit_config_content) with mock.patch('bandit.cli.main._get_options_from_ini' ) as mock_get_opts: mock_get_opts.return_value = {"exclude": "/tmp", "skips": "skip_test", "tests": "some_test"} with mock.patch('bandit.cli.main.LOG.error') as err_mock: # SystemExit with code 2 when test not found in profile self.assertRaisesRegex(SystemExit, '2', bandit.main) self.assertEqual(str(err_mock.call_args[0][0]), 'Unknown test found in profile: some_test')
def test_main_git_command_failure(self): # Test that bandit does not run when the Git command fails repo_directory = self.useFixture(fixtures.TempDir()).path git_repo = git.Repo.init(repo_directory) git_repo.index.commit('Initial Commit') os.chdir(repo_directory) additional_content = 'additional_file.py' with open(additional_content, 'wt') as fd: fd.write(self.temp_file_contents) git_repo.index.add([additional_content]) git_repo.index.commit('Additional Content') with mock.patch('git.Repo.commit') as mock_git_repo_commit: mock_git_repo_commit.side_effect = git.exc.GitCommandError( 'commit', '') # assert the system exits with code 2 self.assertRaisesRegex(SystemExit, '2', baseline.main)
def test_main_subprocess_error(self): # Test that bandit handles a CalledProcessError when attempting to run # bandit baseline via a subprocess repo_directory = self.useFixture(fixtures.TempDir()).path git_repo = git.Repo.init(repo_directory) git_repo.index.commit('Initial Commit') os.chdir(repo_directory) additional_content = 'additional_file.py' with open(additional_content, 'wt') as fd: fd.write(self.temp_file_contents) git_repo.index.add([additional_content]) git_repo.index.commit('Additional Content') with mock.patch('subprocess.check_output') as mock_check_output: mock_bandit_cmd = 'bandit_mock -b temp_file.txt' mock_check_output.side_effect = ( subprocess.CalledProcessError('3', mock_bandit_cmd) ) # assert the system exits with code 3 (returned from # CalledProcessError) self.assertRaisesRegex(SystemExit, '3', baseline.main)
def test_initialize_git_command_failure(self): # Test that bandit does not run when the Git command fails repo_directory = self.useFixture(fixtures.TempDir()).path git_repo = git.Repo.init(repo_directory) git_repo.index.commit('Initial Commit') os.chdir(repo_directory) additional_content = 'additional_file.py' with open(additional_content, 'wt') as fd: fd.write(self.temp_file_contents) git_repo.index.add([additional_content]) git_repo.index.commit('Additional Content') with mock.patch('git.Repo') as mock_git_repo: mock_git_repo.side_effect = git.exc.GitCommandNotFound('clone', '') return_value = baseline.initialize() # assert bandit did not run due to git command failure self.assertEqual((None, None, None), return_value)
def test_initialize_existing_report_file(self): # Test that bandit does not run when the output file exists (and the # provided output format does not match the default format) when # calling the initialize method repo_directory = self.useFixture(fixtures.TempDir()).path git_repo = git.Repo.init(repo_directory) git_repo.index.commit('Initial Commit') os.chdir(repo_directory) # create an existing version of output report file existing_report = "{}.{}".format(baseline.report_basename, 'txt') with open(existing_report, 'wt') as fd: fd.write(self.temp_file_contents) return_value = baseline.initialize() # assert bandit did not run due to existing report file self.assertEqual((None, None, None), return_value)
def test_initialize_existing_temp_file(self): # Test that bandit does not run when the temporary output file exists # when calling the initialize method repo_directory = self.useFixture(fixtures.TempDir()).path git_repo = git.Repo.init(repo_directory) git_repo.index.commit('Initial Commit') os.chdir(repo_directory) # create an existing version of temporary output file existing_temp_file = baseline.baseline_tmp_file with open(existing_temp_file, 'wt') as fd: fd.write(self.temp_file_contents) return_value = baseline.initialize() # assert bandit did not run due to existing temporary report file self.assertEqual((None, None, None), return_value)
def test_logs_on_no_permissions(self): workdir = self.useFixture(fixtures.TempDir()) fake_logger = self.useFixture(fixtures.FakeLogger()) bad_path = os.path.join(workdir.path, 'path_not_readable') with open(bad_path, 'w') as f: f.write("# You can't read me") os.chmod(bad_path, 0) result = _build_doubles.extract_schemas_from_file(bad_path) self.assertIsNone(result) self.assertThat( fake_logger.output, Contains('Extracting schemas from %s' % bad_path)) self.assertThat( fake_logger.output, Contains('Cannot extract schemas: Permission denied'))
def test_logs_on_syntax_error(self): workdir = self.useFixture(fixtures.TempDir()) fake_logger = self.useFixture(fixtures.FakeLogger()) bad_path = os.path.join(workdir.path, 'foo.py') with open(bad_path, 'w') as f: f.write("not valid pyton") result = _build_doubles.extract_schemas_from_file(bad_path) self.assertIsNone(result) self.assertThat( fake_logger.output, Contains('Extracting schemas from %s' % bad_path)) self.assertThat( fake_logger.output, Contains('Cannot extract schemas: invalid syntax (foo.py, line 1)') )
def setUp(self): super(SampleConfig, self).setUp() tmpdir = self.useFixture(fixtures.TempDir()).path self._path = os.path.join(tmpdir, "config.yml") with open(self._path, 'w') as f: f.write("""device: /dev/ttyUSB0 logfile: test.log mqtt: server: 10.42.0.3 names: "ec:01": "Outside" "65:00": "Rain" "33:00": "Wind" "a9:04": "Freezer" "8c:00": "Refrigerator" "ce:08": "Arwen Room" "07:05": "Office" "e3:02": "Bomb Shelter" "de:01": "Subaru" "8e:01": "Cold Frame" "55:09": "Bed Room" "e9:04": "Garage" """)
def test_missing_ca_certificate(self): """ If no CA certificate is found in the service account directory, ``https_policy_from_config`` raises ``ValueError``. """ t = FilePath(self.useFixture(TempDir()).join(b"")) serviceaccount = t.child(b"serviceaccount") serviceaccount.makedirs() serviceaccount.child(b"ca.crt").setContent(b"not a cert pem") serviceaccount.child(b"token").setContent(b"token") self.patch( os, "environ", { b"KUBERNETES_SERVICE_HOST": b"example.invalid.", b"KUBERNETES_SERVICE_PORT": b"443", }, ) config = KubeConfig.from_service_account(path=serviceaccount.path) self.assertThat( lambda: https_policy_from_config(config), raises(ValueError("No certificate authority certificate found.")), )
def test_get_dev(self): tempdir = self.useFixture(fixtures.TempDir()).path l = loop.LoopMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', _fake_trycmd_losetup_works)) self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', _fake_noop)) # No error logged, device consumed self.assertTrue(l.get_dev()) self.assertTrue(l.linked) self.assertEqual('', l.error) self.assertEqual('/dev/loop0', l.device) # Free l.unget_dev() self.assertFalse(l.linked) self.assertEqual('', l.error) self.assertIsNone(l.device)
def test_inner_get_dev_fails(self): tempdir = self.useFixture(fixtures.TempDir()).path l = loop.LoopMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', _fake_trycmd_losetup_fails)) # No error logged, device consumed self.assertFalse(l._inner_get_dev()) self.assertFalse(l.linked) self.assertNotEqual('', l.error) self.assertIsNone(l.device) # Free l.unget_dev() self.assertFalse(l.linked) self.assertIsNone(l.device)
def test_nbd_not_loaded(self): tempdir = self.useFixture(fixtures.TempDir()).path n = nbd.NbdMount(self.file, tempdir) # Fake out os.path.exists def fake_exists(path): if path.startswith('/sys/block/nbd'): return False return ORIG_EXISTS(path) self.useFixture(fixtures.MonkeyPatch('os.path.exists', fake_exists)) # This should fail, as we don't have the module "loaded" # TODO(mikal): work out how to force english as the gettext language # so that the error check always passes self.assertIsNone(n._allocate_nbd()) self.assertEqual('nbd unavailable: module not loaded', n.error)
def test_inner_get_dev_works(self): tempdir = self.useFixture(fixtures.TempDir()).path n = nbd.NbdMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('random.shuffle', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('os.path.exists', self.fake_exists_one)) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', self.fake_trycmd_creates_pid)) self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', _fake_noop)) # No error logged, device consumed self.assertTrue(n._inner_get_dev()) self.assertTrue(n.linked) self.assertEqual('', n.error) self.assertEqual('/dev/nbd0', n.device) # Free n.unget_dev() self.assertFalse(n.linked) self.assertEqual('', n.error) self.assertIsNone(n.device)
def test_get_dev_timeout(self): # Always fail to get a device def fake_get_dev_fails(self): return False self.stubs.Set(nbd.NbdMount, '_inner_get_dev', fake_get_dev_fails) tempdir = self.useFixture(fixtures.TempDir()).path n = nbd.NbdMount(self.file, tempdir) self.useFixture(fixtures.MonkeyPatch('random.shuffle', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('time.sleep', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('nova.utils.execute', _fake_noop)) self.useFixture(fixtures.MonkeyPatch('os.path.exists', self.fake_exists_one)) self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', self.fake_trycmd_creates_pid)) self.useFixture(fixtures.MonkeyPatch(('nova.virt.disk.mount.api.' 'MAX_DEVICE_WAIT'), -10)) # No error logged, device consumed self.assertFalse(n.get_dev())
def test_do_mount_need_to_specify_fs_type(self): # NOTE(mikal): Bug 1094373 saw a regression where we failed to # communicate a failed mount properly. def fake_trycmd(*args, **kwargs): return '', 'broken' self.useFixture(fixtures.MonkeyPatch('nova.utils.trycmd', fake_trycmd)) imgfile = tempfile.NamedTemporaryFile() self.addCleanup(imgfile.close) tempdir = self.useFixture(fixtures.TempDir()).path mount = nbd.NbdMount(imgfile.name, tempdir) def fake_returns_true(*args, **kwargs): return True mount.get_dev = fake_returns_true mount.map_dev = fake_returns_true self.assertFalse(mount.do_mount())
def test_smoke(self): output = os.path.join(self.useFixture(TempDir()).path, 'output') stdin = io.BytesIO() stdout = io.StringIO() writer = StreamResultToBytes(stdin) writer.startTestRun() writer.status( 'foo', 'success', set(['tag']), file_name='fred', file_bytes=b'abcdefg', eof=True, mime_type='text/plain') writer.stopTestRun() stdin.seek(0) _to_disk.to_disk(['-d', output], stdin=stdin, stdout=stdout) self.expectThat( os.path.join(output, 'foo/test.json'), FileContains( '{"details": ["fred"], "id": "foo", "start": null, ' '"status": "success", "stop": null, "tags": ["tag"]}')) self.expectThat( os.path.join(output, 'foo/fred'), FileContains('abcdefg'))
def test_storage_dir_provided(self): """ When the program is run with an argument, it should start up and run. The program is expected to fail because it is unable to connect to Marathon. This test takes a while because we have to let txacme go through it's initial sync (registration + issuing of 0 certificates) before things can be halted. """ temp_dir = self.useFixture(TempDir()) yield main(reactor, raw_args=[ temp_dir.path, '--acme', LETSENCRYPT_STAGING_DIRECTORY.asText(), '--marathon', 'http://localhost:28080' # An address we can't reach ]) # Expect a 'certs' directory to be created self.assertThat(os.path.isdir(temp_dir.join('certs')), Equals(True)) # Expect a default certificate to be created self.assertThat(os.path.isfile(temp_dir.join('default.pem')), Equals(True)) # Expect to be unable to connect to Marathon flush_logged_errors(ConnectionRefusedError)
def setUp(self): super(WorkerThreadTest, self).setUp() self.image = FAKE_IMAGE.copy() # NOTE(jeffrey4l): use a real, temporary dir self.image['path'] = self.useFixture(fixtures.TempDir()).path
def setUp(self): super(GitLogsTest, self).setUp() self.temp_path = self.useFixture(fixtures.TempDir()).path self.root_dir = os.path.abspath(os.path.curdir) self.git_dir = os.path.join(self.root_dir, ".git") self.useFixture( fixtures.EnvironmentVariable('SKIP_GENERATE_AUTHORS')) self.useFixture( fixtures.EnvironmentVariable('SKIP_WRITE_GIT_CHANGELOG'))
def _setUp(self): path = self.useFixture(fixtures.TempDir()).path virtualenv.create_environment(path, clear=True) python = os.path.join(path, 'bin', 'python') command = [python] + PIP_CMD + [ '-U', PIPVERSION, 'wheel'] if self._install_pbr: command.append(PBRVERSION) self.useFixture(base.CapturedSubprocess( 'mkvenv-' + self._reason, command)) self.addCleanup(delattr, self, 'path') self.addCleanup(delattr, self, 'python') self.path = path self.python = python return path, python
def setUp(self): super(GetMacByPciAddressTestCase, self).setUp() self.pci_address = '0000:07:00.1' self.if_name = 'enp7s0f1' self.tmpdir = self.useFixture(fixtures.TempDir()) self.fake_file = os.path.join(self.tmpdir.path, "address") with open(self.fake_file, "w") as f: f.write("a0:36:9f:72:00:00\n")
def setUp(self): super(HookApplyConfigTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-apply-config/install.d/hook-apply-config.py') self.metadata_dir = self.useFixture(fixtures.TempDir()) self.templates_dir = self.useFixture(fixtures.TempDir()) tmp_dir = tempfile.NamedTemporaryFile(mode='w', delete=False).name os.unlink(tmp_dir) self.tmp_file = os.path.basename(tmp_dir) self.out_dir = self.templates_dir.join('tmp') self.metadata = self.metadata_dir.join(self.tmp_file) self.env = os.environ.copy() self.env.update({ 'OS_CONFIG_FILES': self.metadata, 'OS_CONFIG_APPLIER_TEMPLATES': self.templates_dir.join(), }) # our fake metadata file with open(self.metadata, "w+") as md: md.write(json.dumps({'foo': 'bar'})) # This is our fake template root we use to verify os-apply-config # works as expected os.mkdir(self.out_dir) with open(os.path.join(self.out_dir, self.tmp_file), "w+") as template: template.write("foo={{foo}}")
def setUp(self): super(HookScriptTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-script/install.d/hook-script.py') self.fake_tool_path = self.relative_path( __file__, 'config-tool-fake.py') with open(self.fake_tool_path) as f: self.fake_tool_contents = f.read() self.data = { 'id': '1234', 'group': 'script', 'inputs': [ {'name': 'foo', 'value': 'bar'}, {'name': 'another', 'value': 'input'}, {'name': 'a_dict', 'value': '{"key": "value"}'}, {'name': 'a_list', 'value': '["v1", 12]'}, ], 'outputs': [ {'name': 'first_output'}, {'name': 'second_output'} ], 'config': self.fake_tool_contents } self.working_dir = self.useFixture(fixtures.TempDir()) self.outputs_dir = self.useFixture(fixtures.TempDir()) self.test_state_path = self.outputs_dir.join('test_state.json') self.env = os.environ.copy() self.env.update({ 'HEAT_SCRIPT_WORKING': self.working_dir.join(), 'HEAT_SCRIPT_OUTPUTS': self.outputs_dir.join(), 'TEST_STATE_PATH': self.test_state_path, })
def setUp(self): super(HookDockerCmdTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-docker-cmd/install.d/hook-docker-cmd.py') self.cleanup_path = self.relative_path( __file__, '..', 'heat-config-docker-cmd/', 'os-refresh-config/configure.d/50-heat-config-docker-cmd') self.fake_tool_path = self.relative_path( __file__, 'config-tool-fake.py') self.working_dir = self.useFixture(fixtures.TempDir()) self.outputs_dir = self.useFixture(fixtures.TempDir()) self.test_state_path = self.outputs_dir.join('test_state.json') self.env = os.environ.copy() self.env.update({ 'HEAT_DOCKER_CMD': self.fake_tool_path, 'TEST_STATE_PATH': self.test_state_path, })
def setUp(self): super(HeatConfigNotifyTest, self).setUp() self.deployed_dir = self.useFixture(fixtures.TempDir()) hcn.init_logging = mock.MagicMock() if six.PY2: self.stdin = io.BytesIO() else: self.stdin = io.StringIO()
def setUp(self): super(HookPuppetTest, self).setUp() self.hook_path = self.relative_path( __file__, '..', 'heat-config-puppet/install.d/hook-puppet.py') self.fake_tool_path = self.relative_path( __file__, 'config-tool-fake.py') self.working_dir = self.useFixture(fixtures.TempDir()) self.outputs_dir = self.useFixture(fixtures.TempDir()) self.log_dir = self.useFixture(fixtures.TempDir()) self.hiera_datadir = self.useFixture(fixtures.TempDir()) self.test_state_path = self.outputs_dir.join('test_state.json') self.env = os.environ.copy() self.env.update({ 'HEAT_PUPPET_WORKING': self.working_dir.join(), 'HEAT_PUPPET_OUTPUTS': self.outputs_dir.join(), 'HEAT_PUPPET_LOGDIR': self.log_dir.join(), 'HEAT_PUPPET_HIERA_DATADIR': self.hiera_datadir.join(), 'HEAT_PUPPET_CMD': self.fake_tool_path, 'TEST_STATE_PATH': self.test_state_path, })
def _setUp(self): path = self.useFixture(fixtures.TempDir()).path virtualenv.create_environment(path, clear=True) python = os.path.join(path, 'bin', 'python') command = [python] + self.pip_cmd + ['-U'] if self.modules and len(self.modules) > 0: command.extend(self.modules) self.useFixture(base.CapturedSubprocess( 'mkvenv-' + self._reason, command)) self.addCleanup(delattr, self, 'path') self.addCleanup(delattr, self, 'python') self.path = path self.python = python return path, python
def test_pip_versions(self): pkgs = { 'test_markers': {'requirements.txt': textwrap.dedent("""\ pkg_a; python_version=='1.2' pkg_b; python_version!='1.2' """)}, 'pkg_a': {}, 'pkg_b': {}, } pkg_dirs = self.useFixture( test_packaging.CreatePackages(pkgs)).package_dirs temp_dir = self.useFixture(fixtures.TempDir()).path repo_dir = os.path.join(temp_dir, 'repo') venv = self.useFixture(test_packaging.Venv('markers')) bin_python = venv.python os.mkdir(repo_dir) for module in self.modules: self._run_cmd( bin_python, ['-m', 'pip', 'install', '--upgrade', module], cwd=venv.path, allow_fail=False) for pkg in pkg_dirs: self._run_cmd( bin_python, ['setup.py', 'sdist', '-d', repo_dir], cwd=pkg_dirs[pkg], allow_fail=False) self._run_cmd( bin_python, ['-m', 'pip', 'install', '--no-index', '-f', repo_dir, 'test_markers'], cwd=venv.path, allow_fail=False) self.assertIn('pkg-b', self._run_cmd( bin_python, ['-m', 'pip', 'freeze'], cwd=venv.path, allow_fail=False)[0])
def setUp(self): super(TestFetch, self).setUp() # Creating temporaty directory for repos self.tmp_path = self.useFixture(fixtures.TempDir()).path self.conf['repositories']['path'] = self.tmp_path if self.clone_side_effect: fixture_clone = fixtures.MockPatch('git.Repo.clone_from') self.mock_clone = self.useFixture(fixture_clone).mock self.mock_clone.side_effect = self.clone_side_effect else: self.repo = mock.Mock() self.repo.git.checkout.side_effect = self.checkout_side_effect fixture_clone = fixtures.MockPatch('git.Repo.clone_from', return_value=self.repo) self.mock_clone = self.useFixture(fixture_clone).mock
def test_render_dockerfile(self): tmp_dir = self.useFixture(fixtures.TempDir()).path fname = os.path.join(tmp_dir, 'Dockerfile.j2') with open(fname, 'w') as f: f.write(self.source) if not self.exception: res = build.render_dockerfile(fname, 'name', self.config) self.assertEqual(res, self.result) else: exc = self.assertRaises(ValueError, build.render_dockerfile, fname, 'name', self.config) self.assertEqual(exc.args[0], self.exception.args[0])
def _prepare_policy(self): policy_dir = self.useFixture(fixtures.TempDir()) policy_file = os.path.join(policy_dir.path, 'policy.yaml') # load the fake_policy data and add the missing default rules. policy_rules = jsonutils.loads('{}') self.add_missing_default_rules(policy_rules) with open(policy_file, 'w') as f: jsonutils.dump(policy_rules, f) BaseTestCase.conf_override(policy_file=policy_file, group='oslo_policy') BaseTestCase.conf_override(policy_dirs=[], group='oslo_policy')
def setUp(self): super().setUp() self.tmpdir = self.useFixture(fixtures.TempDir()).path self.src_mbox = self.useFixture( MaildirFixture(self.tmpdir, 'source-mailbox') ) # self.msg = self.src_mbox.make_message() self.dest_mbox = self.useFixture( MaildirFixture(self.tmpdir, 'destination-mailbox') ) self.client = client.MaildirClient({'maildir': self.tmpdir})
