def test_successful_flaky_test(self):
        """
        A successful flaky test is considered successful.
        """

        # We use 'testtools' here to avoid accidentally depending on Twisted
        # TestCase features, thus increasing complexity.
        class SomeTest(testtools.TestCase):

            run_tests_with = retry_flaky()

            @flaky(u'FLOC-XXXX')
            def test_something(self):
                pass

        test = SomeTest('test_something')
        self.assertThat(run_test(test), has_results(tests_run=Equals(1))) 

def test_intermittent_flaky_test_that_errors(self, test_methods):
        """
        Tests marked with 'flaky' are retried if they fail, and marked as
        erroring / failing if they don't reach the minimum number of successes.
        """
        executions = iter(test_methods)

        class SomeTest(testtools.TestCase):
            run_tests_with = retry_flaky()

            @flaky(u'FLOC-XXXX', max_runs=len(test_methods), min_passes=2)
            def test_something(self):
                next(executions)()

        test = SomeTest('test_something')
        self.assertThat(run_test(test), has_results(
            tests_run=Equals(1),
            errors=HasLength(1),
        )) 

def test_intermittent_flaky_subclass(self, test_methods):
        """
        We sometimes subclass test classes in order to test different
        implementations of the same interface. A test within such a subclass
        can be marked as flaky, causing it to retry.
        """
        executions = iter(test_methods)

        class SomeTest(testtools.TestCase):
            run_tests_with = retry_flaky()

            def test_something(self):
                next(executions)()

        class SubclassTest(SomeTest):

            @flaky(u'FLOC-XXXX', max_runs=len(test_methods), min_passes=1)
            def test_something(self):
                super(SubclassTest, self).test_something()

        test = SubclassTest('test_something')
        self.assertThat(run_test(test), has_results(tests_run=Equals(1))) 

def test_flaky_testtools_skipped_test(self, jira_keys, max_runs,
                                          min_passes, reasons):
        """
        If a test is skipped and also marked @flaky, we report it as skipped.
        """
        [min_passes, max_runs] = sorted([min_passes, max_runs])
        observed_reasons = []

        class SkippingTest(testtools.TestCase):
            run_tests_with = retry_flaky()

            @flaky(jira_keys, max_runs, min_passes)
            def test_skip(self):
                observed_reasons.append(reasons.next())
                self.skip(observed_reasons[-1])

        test = SkippingTest('test_skip')
        self.assertThat(run_test(test), only_skips(1, observed_reasons)) 

def test_sends_start_on_start(self, jira_keys):
        """
        Flaky tests send ``startTest`` as soon as they start.

        This means that we see the name of the test in the trial reporter
        (FLOC-3511) and that the recorded duration of the test is more-or-less
        accurate (FLOC-3499)
        """
        log = []

        class FlakyTest(testtools.TestCase):
            run_tests_with = retry_flaky()

            @flaky(jira_keys, 1, 1)
            def test_delayed(self):
                # Get a copy of the event log at this point of the test.
                self.log = list(log)

        test = FlakyTest('test_delayed')
        result = ExtendedTestResult(log)
        test.run(result)
        self.assertThat(test.log, Equals([('startTest', test)])) 

def setUp(self):
        "Run before each test method to initialize test environment."

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def test_failedSaysSo(self):
        """
        A failing test case causes the whole run to report 'FAILED'
        """
        sub_tmpdir = tempfile.mkdtemp(dir=self.tmpdir)
        fh = open(os.path.join(sub_tmpdir, "test_failed.py"), "w")
        fh.write(
            dedent(
                """
            import unittest
            class Failed(unittest.TestCase):
                def test01(self):
                    self.assertTrue(False)
            """.format(
                    os.getpid()
                )
            )
        )
        fh.close()
        os.chdir(sub_tmpdir)
        tests = self.loader.loadTargets("test_failed")
        result = run(tests, self.stream, self.args)
        os.chdir(self.startdir)
        self.assertEqual(result.testsRun, 1)
        self.assertIn("FAILED", self.stream.getvalue()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.temp_homedir = self.useFixture(fixtures.TempHomeDir()).path

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 60)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid, set a default timeout.
            test_timeout = 60
        if test_timeout <= 0:
            test_timeout = 60
        self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def setUp(self):
        """Run before each test method to initialize test environment."""

        super(TestCase, self).setUp()
        self.stubbed_mapped_nics = {}

        def dummy_mapped_nics(nic_mapping=None):
            return self.stubbed_mapped_nics
        if self.stub_mapped_nics:
            self.stub_out('os_net_config.objects.mapped_nics',
                          dummy_mapped_nics)

        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if os.environ.get('OS_STDOUT_CAPTURE') in _TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in _TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger()) 

def tearDown(self):
        super(TestCase, self).tearDown() 

def setUp(self):
        super(TestCase, self).setUp()
        config.register_cli_opts()
        self.addCleanup(mock.patch.stopall)
        self.addCleanup(self.clean_caches) 

def setUp(self):
        super(TestCase, self).setUp()
        self.addCleanup(mock.patch.stopall) 

def mktemp(self):
        """
        Create a temporary path for use in tests.

        Provided for compatibility with Twisted's ``TestCase``.

        :return: Path to non-existent file or directory.
        """
        return self.make_temporary_path().path 

def setUp(self):
        log.msg("--> Begin: %s <--" % (self.id()))
        super(TestCase, self).setUp()
        self.useFixture(_SplitEliotLogs()) 

def assertFailure(self, deferred, exception):
        """
        ``twisted.trial.unittest.TestCase.assertFailure``-alike.

        This is not completely compatible.  ``assert_fails_with`` should be
        preferred for new code.
        """
        return assert_fails_with(deferred, exception) 

def _get_flaky_annotation(case):
    """
    Get the flaky decoration detail from ``case``.

    :param TestCase case: A test case that might have been decorated with
        @flaky.
    :return: ``None`` if not flaky, or a ``pmap`` of the flaky test details.
    """
    # XXX: Alas, there's no public way of doing this:
    # https://bugs.launchpad.net/testtools/+bug/1517867
    method = case._get_test_method()
    return getattr(method, _FLAKY_ATTRIBUTE, None) 

def _run_test(self, case, result):
        """
        Run ``case`` with the ``RunTest`` we are wrapping.

        :param testtools.TestCase case: The test to run.
        :param testtools.TestResult result: The test result to report to.
            Must conform to testtools extended test result interface.
        :return: The modified ``result``.
        """
        run_test = self._run_test_factory(case, *self._args, **self._kwargs)
        return run_test._run_prepared_result(result) 

def _attempt_test(self, case):
        """
        Run 'case' with a temporary result.

        :param testtools.TestCase case: The test to run.

        :return: a tuple of ``(successful, result, details)``, where
            ``successful`` is a boolean indicating whether the test was
            succcessful, ``result`` is a _ResultType indicating what the test
            result was and ``details`` is a dictionary of testtools details.
        """
        tmp_result = testtools.TestResult()
        # XXX: Still using internal API of testtools despite improvements in
        # #165. Will need to do follow-up work on testtools to ensure that
        # RunTest.run(case); RunTest.run(case) is supported.
        case._reset()
        self._run_test(case, tmp_result)
        result_type = _get_result_type(tmp_result)
        details = pmap(case.getDetails())
        if result_type == _ResultType.skip:
            # XXX: Work around a testtools bug where it reports stack traces
            # for skips that aren't passed through its supported
            # SkipException: https://bugs.launchpad.net/testtools/+bug/1518100
            [reason] = list(tmp_result.skip_reasons.keys())
            details = details.discard('traceback').set(
                'reason', text_content(reason))
        return (tmp_result.wasSuccessful(), result_type, details) 

def test_always_erroring_flaky_test(self, jira_keys, max_runs, min_passes):
        """
        A flaky test always errors out is recorded as erroring.
        """
        [min_passes, max_runs] = sorted([min_passes, max_runs])

        executions = repeat(lambda: throw(ValueError('failure')))

        class SomeTest(testtools.TestCase):

            run_tests_with = retry_flaky()

            @flaky(jira_keys, max_runs=max_runs, min_passes=min_passes)
            def test_something(self):
                next(executions)()

        test = SomeTest('test_something')
        result = run_test(test)
        self.expectThat(
            result, has_results(
                errors=HasLength(1),
                tests_run=Equals(1),
            )
        )
        [(_, exception)] = result.errors
        flaky_data = _get_flaky_annotation(test).to_dict()
        flaky_data.update({'runs': max_runs - min_passes + 1, 'passes': 0})
        self.assertThat(
            exception, MatchesAll(
                Contains('ValueError'),
                Contains(pformat(flaky_data)),
            )
        ) 

def create_engine(self):
        func = TestCase._default_engine
        if func is None:
            engine_factory = factory.YaqlFactory(allow_delegates=True)
            TestCase._default_engine = func = engine_factory.create(
                options=self.engine_options)
        return func 

def create_legacy_engine(self):
        func = TestCase._default_legacy_engine
        if func is None:
            engine_factory = legacy.YaqlFactory()
            TestCase._default_legacy_engine = func = engine_factory.create(
                options=self.legacy_engine_options)
        return func 

def setUp(self):
        self._context = None
        self._engine = None
        self._legacy_context = None
        self._legacy_engine = None
        super(TestCase, self).setUp() 

def setUp(self):
        """Run before each test method to initialize test environment."""
        super(TestCase, self).setUp()

        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))
        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        environ_enabled = (lambda var_name:
                           strutils.bool_from_string(os.environ.get(var_name)))
        if environ_enabled('OS_STDOUT_CAPTURE'):
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if environ_enabled('OS_STDERR_CAPTURE'):
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
        if environ_enabled('OS_LOG_CAPTURE'):
            log_format = '%(levelname)s [%(name)s] %(message)s'
            if environ_enabled('OS_DEBUG'):
                level = logging.DEBUG
            else:
                level = logging.INFO
            self.useFixture(fixtures.LoggerFixture(nuke_handlers=False,
                                                   format=log_format,
                                                   level=level))
        # Protect against any case where someone doesn't remember to patch a
        # retry decorated call
        patcher = mock.patch('os_brick.utils._time_sleep')
        patcher.start()
        self.addCleanup(patcher.stop)