python类TempDir()的实例源码

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 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 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 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})