Python unittest 模块，case() 实例源码

def _testCongestion(self):
        # test the behavior in case of congestion
        self.data = b'fill'
        self.cli.setblocking(False)
        try:
            # try to lower the receiver's socket buffer size
            self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16384)
        except OSError:
            pass
        with self.assertRaises(OSError) as cm:
            try:
                # fill the receiver's socket buffer
                while True:
                    self.cli.sendto(self.data, 0, (HOST, self.port))
            finally:
                # signal the receiver we're done
                self.evt.set()
        # sendto() should have failed with ENOBUFS
        self.assertEqual(cm.exception.errno, errno.ENOBUFS)
        # and we should have received a congestion notification through poll
        r, w, x = select.select([self.serv], [], [], 3.0)
        self.assertIn(self.serv, r)

def nose_transform():
    """Custom transform for the nose.tools module."""

    builder = AstroidBuilder(MANAGER)
    stub = AstroidBuilder(MANAGER).string_build('''__all__ = []''')
    unittest_module = builder.module_build(unittest.case)
    case = unittest_module['TestCase']
    all_entries = ['ok_', 'eq_']

    for method_name, method in case.locals.items():
        if method_name.startswith('assert') and '_' not in method_name:
            pep8_name = _pep8(method_name)
            all_entries.append(pep8_name)
            stub[pep8_name] = method[0]

    # Update the __all__ variable, since nose.tools
    # does this manually with .append.
    all_assign = stub['__all__'].parent
    all_object = List(all_entries)
    all_object.parent = all_assign
    all_assign.value = all_object
    return stub

def nose_transform():
    """Custom transform for the nose.tools module."""

    builder = AstroidBuilder(MANAGER)
    stub = AstroidBuilder(MANAGER).string_build('''__all__ = []''')
    unittest_module = builder.module_build(unittest.case)
    case = unittest_module['TestCase']
    all_entries = ['ok_', 'eq_']

    for method_name, method in case.locals.items():
        if method_name.startswith('assert') and '_' not in method_name:
            pep8_name = _pep8(method_name)
            all_entries.append(pep8_name)
            stub[pep8_name] = method[0]

    # Update the __all__ variable, since nose.tools
    # does this manually with .append.
    all_assign = stub['__all__'].parent
    all_object = List(all_entries)
    all_object.parent = all_assign
    all_assign.value = all_object
    return stub

def nose_transform():
    """Custom transform for the nose.tools module."""

    builder = AstroidBuilder(MANAGER)
    stub = AstroidBuilder(MANAGER).string_build('''__all__ = []''')
    unittest_module = builder.module_build(unittest.case)
    case = unittest_module['TestCase']
    all_entries = ['ok_', 'eq_']

    for method_name, method in case.locals.items():
        if method_name.startswith('assert') and '_' not in method_name:
            pep8_name = _pep8(method_name)
            all_entries.append(pep8_name)
            stub[pep8_name] = method[0]

    # Update the __all__ variable, since nose.tools
    # does this manually with .append.
    all_assign = stub['__all__'].parent
    all_object = List(all_entries)
    all_object.parent = all_assign
    all_assign.value = all_object
    return stub

def get_sha(repo_dir):
    try:
        output = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=repo_dir).strip()
        prefix = 'github:apache/'
        local_repo_location = os.environ.get('LOCAL_GIT_REPO')
        if local_repo_location is not None:
            prefix = 'local:{}:'.format(local_repo_location)  # local: slugs take the form 'local:/some/path/to/cassandra/:branch_name_or_sha'
        return "{}{}".format(prefix, output)
    except CalledProcessError as e:
        if re.search('Not a git repository', e.message) is not None:
            # we tried to get a sha, but repo_dir isn't a git repo. No big deal, must just be working from a non-git install.
            return None
        else:
            # git call failed for some unknown reason
            raise


# There are times when we want to know the C* version we're testing against
# before we call Tester.setUp. In the general case, we can't know that -- the
# test method could use any version it wants for self.cluster. However, we can
# get the version from build.xml in the C* repository specified by
# CASSANDRA_VERSION or CASSANDRA_DIR. This should use the same resolution
# strategy as the actual checkout code in Tester.setUp; if it does not, that is
# a bug.

def test_reduce_move(self):
        from operator import add
        # reduce tests may have already triggered this warning
        reset_module_registry(unittest.case)
        with warnings.catch_warnings():
            warnings.filterwarnings("error", "reduce")
            self.assertRaises(DeprecationWarning, reduce, add, range(10))

def test_reduce_move(self):
        from operator import add
        # reduce tests may have already triggered this warning
        reset_module_registry(unittest.case)
        with warnings.catch_warnings():
            warnings.filterwarnings("error", "reduce")
            self.assertRaises(DeprecationWarning, reduce, add, range(10))

def test_reduce_move(self):
        from operator import add
        # reduce tests may have already triggered this warning
        reset_module_registry(unittest.case)
        with warnings.catch_warnings():
            warnings.filterwarnings("error", "reduce")
            self.assertRaises(DeprecationWarning, reduce, add, range(10))

def test_reduce_move(self):
        from operator import add
        # reduce tests may have already triggered this warning
        reset_module_registry(unittest.case)
        with warnings.catch_warnings():
            warnings.filterwarnings("error", "reduce")
            self.assertRaises(DeprecationWarning, reduce, add, range(10))

def did_fail():
    if sys.exc_info() == (None, None, None):
        return False

    exc_class, _, _ = sys.exc_info()
    return not issubclass(exc_class, unittest.case.SkipTest)