def mandatory(test_item):
"""A decorator to mark whole test cases or individual methods as
mandatory. If a mandatory test fails, DataTestRunner will stop
immediately (this is similar to the ``--failfast`` command line
argument behavior)::
@datatest.mandatory
class TestFileFormat(datatest.DataTestCase):
def test_columns(self):
...
"""
test_item.__datatest_mandatory__ = True
return test_item
# The standard unittest.skip decorators are reimplemented to add a
# _wrapped attribute that points to the orignal object so that the
# _sort_key() function can find the proper line number when test_item
# gets wrapped by functools.wraps().
python类skip()的实例源码
def skip(reason):
"""A decorator to unconditionally skip a test:
.. code-block:: python
@datatest.skip('Not finished collecting raw data.')
class TestSumTotals(datatest.DataTestCase):
def test_totals(self):
...
"""
def decorator(test_item):
if not isinstance(test_item, type):
orig_item = test_item # <- Not in unittest.skip()
@functools.wraps(test_item)
def skip_wrapper(*args, **kwargs):
raise unittest.SkipTest(reason)
test_item = skip_wrapper
test_item._wrapped = orig_item # <- Not in unittest.skip()
test_item.__unittest_skip__ = True
test_item.__unittest_skip_why__ = reason
return test_item
return decorator
def _check_unchunked_files(self, d):
self.assertEqual(d['$chunk.ref_contigset_id'], self.INPUT_FILES[1])
self.assertEqual(d['$chunk.ccsset_id'], self.INPUT_FILES[2])
#@unittest.skip("GMAP disabled")
#class TestScatterContigSetGMAP(ContigSetScatterBase,
# pbcommand.testkit.core.PbTestScatterApp):
# DRIVER_BASE = "python -m pbtranscript.tasks.scatter_contigset_gmap"
# INPUT_FILES = [ # identical to TestIcePartial inputs
# op.join(MNT_DATA, "sa3", "nfl.contigset.xml"),
# "/pbi/dept/secondary/siv/references/rat_UCSC/rat_UCSC.referenceset.xml",
# ]
#
# def _check_unchunked_files(self, d):
# self.assertEqual(d['$chunk.reference_id'], self.INPUT_FILES[1])
def test_skipping(self):
class Foo(unittest.TestCase):
def test_skip_me(self):
self.skipTest("skip")
events = []
result = LoggingResult(events)
test = Foo("test_skip_me")
test.run(result)
self.assertEqual(events, ['startTest', 'addSkip', 'stopTest'])
self.assertEqual(result.skipped, [(test, "skip")])
# Try letting setUp skip the test now.
class Foo(unittest.TestCase):
def setUp(self):
self.skipTest("testing")
def test_nothing(self): pass
events = []
result = LoggingResult(events)
test = Foo("test_nothing")
test.run(result)
self.assertEqual(events, ['startTest', 'addSkip', 'stopTest'])
self.assertEqual(result.skipped, [(test, "testing")])
self.assertEqual(result.testsRun, 1)
def test_skip_doesnt_run_setup(self):
class Foo(unittest.TestCase):
wasSetUp = False
wasTornDown = False
def setUp(self):
Foo.wasSetUp = True
def tornDown(self):
Foo.wasTornDown = True
@unittest.skip('testing')
def test_1(self):
pass
result = unittest.TestResult()
test = Foo("test_1")
suite = unittest.TestSuite([test])
suite.run(result)
self.assertEqual(result.skipped, [(test, "testing")])
self.assertFalse(Foo.wasSetUp)
self.assertFalse(Foo.wasTornDown)
def test_decorated_skip(self):
def decorator(func):
def inner(*a):
return func(*a)
return inner
class Foo(unittest.TestCase):
@decorator
@unittest.skip('testing')
def test_1(self):
pass
result = unittest.TestResult()
test = Foo("test_1")
suite = unittest.TestSuite([test])
suite.run(result)
self.assertEqual(result.skipped, [(test, "testing")])
def test_gc(self):
for tp in self._types:
if not isinstance(tp, type):
# If tp is a factory rather than a plain type, skip
continue
class MySource(tp):
pass
class MyObject:
pass
# Create a reference cycle through a memoryview object
b = MySource(tp(b'abc'))
m = self._view(b)
o = MyObject()
b.m = m
b.o = o
wr = weakref.ref(o)
b = m = o = None
# The cycle must be broken
gc.collect()
self.assertTrue(wr() is None, wr())
def test_1(self):
try:
from sys import getrefcount as grc
except ImportError:
return unittest.skip("no sys.getrefcount()")
f = dll._testfunc_callback_i_if
f.restype = ctypes.c_int
f.argtypes = [ctypes.c_int, MyCallback]
def callback(value):
#print "called back with", value
return value
self.assertEqual(grc(callback), 2)
cb = MyCallback(callback)
self.assertTrue(grc(callback) > 2)
result = f(-10, cb)
self.assertEqual(result, -18)
cb = None
gc.collect()
self.assertEqual(grc(callback), 2)
def testSingleDatacenter(self):
"""
Create a single data center and add check if its switch is up
by using manually added hosts. Tests especially the
data center specific addLink method.
"""
# create network
self.createNet(nswitches=0, ndatacenter=1, nhosts=2, ndockers=0)
# setup links
self.net.addLink(self.dc[0], self.h[0])
self.net.addLink(self.h[1], self.dc[0])
# start Mininet network
self.startNet()
# check number of running nodes
self.assertTrue(len(self.getContainernetContainers()) == 0)
self.assertTrue(len(self.net.hosts) == 2)
self.assertTrue(len(self.net.switches) == 1)
# check connectivity by using ping
self.assertTrue(self.net.ping([self.h[0], self.h[1]]) <= 0.0)
# stop Mininet network
self.stopNet()
#@unittest.skip("disabled to test if CI fails because this is the first test.")
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid # noqa
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_
def test_Xlinked_Norine_BMX(self):
"""This gene has no X-linked variant"""
db = 'lgueneau'
ss = samples_selection_factory(db=db,
groups = {'affected': ['09818','09819'], 'not_affected':['09960','09961']})
qs = Variant.objects.using(db).filter(gene_symbol__in=['BMX'], chrom='chrX')
# Using GenotypesFilterXLinked.apply()
variants = variants_collection_factory(db=db, qs=qs)
var = GenotypesFilterXLinked(ss).apply(variants).variants
self.assertEqual(len(var), 0)
# Using FilterCollection.apply()
fc = FiltersCollection([GenotypesFilterXLinked(ss)])
var = fc.apply(initqs=qs, db=db).variants
self.assertEqual(len(var), 0)
############################
# COMPOUND HET #
############################
#@unittest.skip('')
def test_compound_Lucie_FNDC1(self):
"""These 2(3) variants are annotated on different transcripts of the same gene,
and because of the group_by(transcript_id), the compound was not found.
Check here that it is fixed.
"""
ss = samples_selection_factory(db='test',
groups = {'affected': ['101563','101591'], 'not_affected':['101564','101565']})
variants = variants_collection_factory(db='test',
qs=Variant.objects.using('test').filter(gene_symbol='FNDC1'))
## To select them directly:
#variants = variants_collection_factory(
# Variants.objects.using('test').filter(start__in=['159653504','159653634']), db='test')
var = GenotypesFilterCompoundHeterozygous(ss).apply(variants).variants
self.assertEqual(len(var), 3)
############################
# FROM REQUEST #
############################
#@unittest.skip('')
def test_genotypes_Nothing_filter_from_request_with_ss_from_request(self):
"""With a sound grouping passed by url."""
samples = [('samples', 'A=09818,09819'), ('samples', 'B=09960,09961')]
filters = [('filter', 'genotype=nothing')]
request = RequestFactory().get('', samples + filters)
fc = variant_filters_from_request(request, db='test')
self.assertIsInstance(fc, FiltersCollection)
self.assertTrue(len(fc) == len(filters))
gf = fc['genotype']
self.assertEqual(sorted(gf.ss.groups.keys()), ['A','B'])
self.assertEqual(gf.name, 'genotype')
self.assertEqual(gf.val, 'nothing')
self.assertEqual(gf.op, '=')
self.assertIsInstance(gf, GenotypesFilterDoNothing)
#@unittest.skip('')
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
def skip_if_no_uuid_(self):
try:
import uuid
except ImportError:
return self.skipTest("uuid not available in this Python version")
try:
cur = self.conn.cursor()
cur.execute("select typname from pg_type where typname = 'uuid'")
has = cur.fetchone()
finally:
self.conn.rollback()
if has:
return f(self)
else:
return self.skipTest("uuid type not available on the server")
return skip_if_no_uuid_