python类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 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_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 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_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())