def test_05_03_http_image_zipfile(self):
        # Make a zipfile using files accessed from the web
        def alter_fn(module):
            self.assertTrue(isinstance(module, C.CreateWebPage))
            module.wants_zip_file.value = True
            module.zipfile_name.value = ZIPFILE_NAME
            module.directory_choice.dir_choice = C.ABSOLUTE_FOLDER_NAME
            module.directory_choice.custom_path = cpprefs.get_default_image_directory()

        url_root = "http://cellprofiler.org/svnmirror/ExampleImages/ExampleSBSImages/"
        url_query = "?r=11710"
        filenames = [(url_root, fn + url_query) for fn in
                     ("Channel1-01-A-01.tif", "Channel2-01-A-01.tif",
                      "Channel1-02-A-02.tif", "Channel2-02-A-02.tif")]
        #
        # Make sure URLs are accessible
        #
        try:
            for filename in filenames:
                URLopener().open("".join(filename)).close()
        except IOError, e:
            def bad_url(e=e):
                raise e

            unittest.expectedFailure(bad_url)()

  def test_put_file_twice_no_tic(self):
    self.postFile()
    self.commit()
    self.postFile()
    self.tic()

    document_list = self.portal.portal_catalog(reference=self.key)

    self.assertEqual(2, len(document_list))
    expectedFailure(self.assertEqual)(sorted(['archived', 'published']),
        sorted(q.getValidationState() for q in document_list))

def test_load_using_bioformats_url():
    url = "https://github.com/CellProfiler/python-bioformats/raw/1.0.5/bioformats/tests/Channel1-01-A-01.tif"
    try:
        fd = future.moves.urllib.request.urlopen(url)
        if fd.code < 200 or fd.code >= 300:
            raise OSError("Http error %d" % fd.code)
    except OSError as e:
        def bad_url(e=e):
            raise e
        unittest.expectedFailure(bad_url)()

    data = bioformats.formatreader.load_using_bioformats_url(url, rescale=False)
    assert data.shape == (640, 640)

 def test_03_03_load_using_bioformats_url(self):
     url = "https://github.com/CellProfiler/python-bioformats/raw/1.0.5/bioformats/tests/Channel1-01-A-01.tif"
     try:
         fd = urlopen(url)
         if fd.code < 200 or fd.code >= 300:
             raise OSError("Http error %d" % fd.code)
     except OSError as e:
         def bad_url(e=e):
             raise e
         unittest.expectedFailure(bad_url)()
     
     data = F.load_using_bioformats_url(url, rescale=False)
     self.assertSequenceEqual(data.shape, (640, 640))

 def test_01_05_load_pipeline(self):
     import cellprofiler.pipeline as cpp
     import os
     def callback(caller, event):
         self.assertFalse(isinstance(event, cpp.LoadExceptionEvent))
     pipeline = cpp.Pipeline()
     pipeline.add_listener(callback)
     try:
         fd = urlopen(self.fly_url)
     except IOError, e:
         def bad_url(e=e):
             raise e
         unittest.expectedFailure(bad_url)()

    def test_02_01_run_headless(self):
        output_directory = tempfile.mkdtemp()
        temp_directory = os.path.join(output_directory, "temp")
        os.mkdir(temp_directory)
        try:
            #
            # Run with a .cp file
            #
            fly_pipe = "http://cellprofiler.org/ExampleFlyImages/ExampleFlyURL.cppipe"
            urllib.URLopener().open(fly_pipe).close()
            measurements_file = os.path.join(output_directory, "Measurements.h5")
            done_file = os.path.join(output_directory, "Done.txt")
            self.run_cellprofiler(
                "-c",
                "-r",
                "-o",
                output_directory,
                "-p",
                fly_pipe,
                "-d",
                done_file,
                "-t",
                temp_directory,
                "-f",
                "1",
                "-l",
                "1",
                measurements_file,
            )
            import cellprofiler.preferences as cpprefs

            self.assertTrue(os.path.exists(measurements_file))
            self.assertTrue(os.path.exists(done_file))
            #
            # Re-run using the measurements file.
            #
            m2_file = os.path.join(output_directory, "M2.h5")
            self.run_cellprofiler(
                "-c", "-r", "-o", output_directory, "-f", "1", "-l", "1", "-p", measurements_file, m2_file
            )
            self.assertTrue(os.path.exists(m2_file))
        except IOError, e:
            if e.args[0] != "http error":
                raise e

            def bad_url(e=e):
                raise e

            unittest.expectedFailure(bad_url)()

    def test_ResourceTags(self):
        # Host tags might not be present when using a v2 service
        if self.is_v2:
            return unittest.expectedFailure(self)

        topo = Topology()
        h1 = topo.source(host_name)
        h1.resource_tags.add('host1')
        
        h2 = topo.source(host_name)
        h2.resource_tags.add('host2')
        
        h = h1.union({h2})
        h.print()
        h = h.map(RemoveDup())

        h2sink = h2.for_each(lambda x : None, name='SinkOnHost1')
        h2sink.resource_tags.add('host1')

        beacon = Source(topo, "spl.utility::Beacon",
            'tuple<uint64 seq>',
            params = {'period': 0.02, 'iterations':100},
            name = 'BeaconOnHost1')
        beacon.seq = beacon.output('IterationCount()')
        beacon.resource_tags.add('host2')

        h2sinkNotags = h2.for_each(lambda x : None, name='SinkNoTags')
        self.assertFalse(h2sinkNotags.resource_tags)
        
        self.tester = Tester(topo)
        self.tester.tuple_count(h, 2)
        self.tester.local_check = self.check_placements
     
        sr = self.tester.test(self.test_ctxtype, self.test_config)

def create_test(input_file, output_file, thing):
    first_line = open(input_file, 'r').readline()
    def do_test(self):
        self.check_file(input_file, output_file, thing)
    if first_line.startswith('dnl fail') and thing == 'parser':
        return unittest.expectedFailure(do_test)
    return do_test

def main():
    test_cases = list_delegation_tests(FIXTURE_ROOT)
    
    # Generate the 'constraint-solving-without-injection' tests.
    # These replace the TargetSolverNoInjection tests we used to have in test.py.
    constraint_solved_tests = {
            "eq": {
                #"Concolic::Solver::ErrorsReadingSolution": "0",    # Error by default, don't neet to test explicitly.
                #"Concolic::Solver::ConstraintsSolvedAsUNSAT": "0", # Error by default, don't neet to test explicitly.
                #"Concolic::Solver::ConstraintsNotSolved": "0",     # Error by default, don't neet to test explicitly.
                "Concolic::Solver::ConstraintsSolved": "1"
            }
        }
    
    for t in test_cases:
        test_name = 'test_constraint-solving-without-injection_%s' % t['name']
        file_name = "%s%s" % (FIXTURE_ROOT, t['fn'])
        test = concolic.test_generator(_artemis_runner_no_injections, file_name, test_name, test_dict=constraint_solved_tests, internal_test=None)
        setattr(EventDelegation, test_name, test)
    
    # Generate the tests which check for the assertions included in the test suite.
    for t in test_cases:
        test_name = 'test_%s' % t['name']
        file_name = "%s%s" % (FIXTURE_ROOT, t['fn'])
        
        test = concolic.test_generator(_artemis_runner_full, file_name, test_name, test_dict=t['test'], internal_test=t['i_test'])
        
        if t['expected_failure']:
            test = unittest.expectedFailure(test)
        
        setattr(EventDelegation, test_name, test)
    
    unittest.main(buffer=True, catchbreak=True)

def mark_tests_as_expected_failure(failing_tests):
    """
    Flag tests as expectedFailure. This should only run during the
    testsuite.
    """
    django_version = django.VERSION[:2]
    for test_name, versions in six.iteritems(failing_tests):
        if not versions or not isinstance(versions, (list, tuple)):
            # skip None, empty, or invalid
            continue
        if not isinstance(versions[0], (list, tuple)):
            # Ensure list of versions
            versions = [versions]
        if all(map(lambda v: v[:2] != django_version, versions)):
            continue
        test_case_name, _, method_name = test_name.rpartition('.')
        try:
            test_case = import_string(test_case_name)
        except ImproperlyConfigured:
            # Django tests might not be available during
            # testing of client code
            continue
        method = getattr(test_case, method_name)
        method = expectedFailure(method)
        setattr(test_case, method_name, method)

 def f(func):
     rc = subprocess.call(['ip', 'link', 'add', 'dev', 'erspan99', 'type', 'erspan', 'seq', 'key', '30', 'local', '192.168.1.4', 'remote', '192.168.1.1', 'erspan_ver', '1', 'erspan', '123'])
     if rc == 0:
         subprocess.call(['ip', 'link', 'del', 'erspan99'])
         return func
     else:
         return unittest.expectedFailure(func)

def maybe_download_tesst_image( file_name):
    '''Download the given TestImages file if not in the directory

    file_name - name of file to fetch

    Image will be downloaded if not present to CP_EXAMPLEIMAGES directory.
    '''
    local_path = os.path.join(testimages_directory(), file_name)
    if not os.path.exists(local_path):
        url = testimages_url() + "/" + file_name
        try:
            URLopener().retrieve(url, local_path)
        except IOError, e:
            # This raises the "expected failure" exception.
            def bad_url(e=e):
                raise e
            unittest.expectedFailure(bad_url)()

def implements(commands, expectedFailure=False):  # noqa: N803
    def _test(self):
        with self.run_bpd() as client:
            response = client.send_command('commands')
        self._assert_ok(response)
        implemented = response.data['command']
        self.assertEqual(commands.intersection(implemented), commands)
    return unittest.expectedFailure(_test) if expectedFailure else _test

def _addTests():
    for path in glob.glob(os.path.join(drawBotScriptDir, "*.py")):
        scriptName = os.path.splitext(os.path.basename(path))[0]
        for ext in testExt:
            testMethodName = "test_%s_%s" % (ext, scriptName)
            testMethod = makeTestCase(path, ext, testMethodName in ignoreDeprecationWarnings)
            testMethod.__name__ = testMethodName
            if testMethodName in expectedFailures:
                testMethod = unittest.expectedFailure(testMethod)
            setattr(DrawBotTest, testMethodName, testMethod)

def cxOracle_py3_bug(func):
    """
    There's a bug in Django/cx_Oracle with respect to string handling under
    Python 3 (essentially, they treat Python 3 strings as Python 2 strings
    rather than unicode). This makes some tests here fail under Python 3, so
    we mark them as expected failures until someone fixes them in #23843.
    """
    from unittest import expectedFailure
    from django.db import connection
    return expectedFailure(func) if connection.vendor == 'oracle' else func

def expectedFailureIfAppengine(func):
    try:
        from google import appengine
    except ImportError:
        return func
    else:
        import os

        if os.environ.get('SERVER_SOFTWARE', None) is None:
            return func

        return unittest.expectedFailure(func)

def cxOracle_513_py3_bug(func):
    """
    cx_Oracle versions up to and including 5.1.3 have a bug with respect to
    string handling under Python3 (essentially, they treat Python3 strings
    as Python2 strings rather than unicode). This makes some tests here
    fail under Python 3 -- so we mark them as expected failures.

    See  https://code.djangoproject.com/ticket/23843, in particular comment 6,
    which points to https://bitbucket.org/anthony_tuininga/cx_oracle/issue/6/
    """
    from unittest import expectedFailure
    from django.db import connection

    if connection.vendor == 'oracle' and six.PY3 and connection.Database.version <= '5.1.3':
        return expectedFailure(func)
    else:
        return func

def _addExampleTests():
    allExamples = _collectExamples([
        DrawBotDrawingTool,
        DrawBotDrawingTool._formattedStringClass,
        DrawBotDrawingTool._bezierPathClass,
        DrawBotDrawingTool._imageClass
    ])

    for exampleName, source in allExamples.items():
        testMethodName = "test_%s" % exampleName
        testMethod = _makeTestCase(exampleName, source, doSaveImage=testMethodName not in dontSaveImage,
                allowFuzzyImageComparison=testMethodName in allowFuzzyImageComparison)
        testMethod.__name__ = testMethodName
        if testMethodName in skip:
            continue
        if testMethodName in expectedFailures:
            testMethod = unittest.expectedFailure(testMethod)
        setattr(ExampleTester, testMethodName, testMethod)

  def catalog(self, reference_tree, kw, check_search_text=True,
      check_select_expression=True, expected_failure=False):
    reference_param_dict = self._catalog._queryResults(query_table='foo', **kw)
    query = self._catalog.buildQuery(kw)
    assertEqual = self.assertEqual
    if expected_failure:
      assertEqual = unittest.expectedFailure(assertEqual)

    assertEqual(reference_tree, query)
    search_text = query.asSearchTextExpression(self._catalog)
    if check_search_text:
      # XXX: sould "keyword" be always used for search text searches ?
      search_text_param_dict = self._catalog._queryResults(query_table='foo', keyword=search_text)
      if not check_select_expression:
        search_text_param_dict.pop('select_expression')
        reference_param_dict.pop('select_expression')
      assertEqual(reference_param_dict, search_text_param_dict,
          'Query: %r\nSearchText: %r\nReference: %r\nSecond rendering: %r' % \
                       (query, search_text, reference_param_dict, search_text_param_dict))

def expected_failure_if_suit(func):
    if 'suit' in settings.INSTALLED_APPS:
        return unittest.expectedFailure(func)
    return func