def test_rename(self):
        "Histogram: rename"
        import tempfile
        outh5 = tempfile.mktemp()
        code = """
from histogram import histogram, arange
h = histogram('h', [('x', arange(10), 'meter')])
h.I = h.x*h.x
h.setAttribute('name', 'abc')
from histogram.hdf import load, dump
dump(h, %r, '/', 'c')
""" % outh5
        script = tempfile.mktemp()
        open(script, 'w').write(code)
        cmd = 'python %s' % script
        import os
        
        if os.system(cmd):
            raise RuntimeError, "%s failed" % cmd
        
        from histogram.hdf import load
        h = load(outh5, 'abc')

        os.remove(outh5)
        os.remove(script)
        return        

    def test_document_fusion(self):
        # data source and model are in the same content
        alsoProvides(self.portal.REQUEST, ICollectiveDocumentfusionLayer)
        content = api.content.create(self.portal, type='letter',
                           title=u"En réponse...",
                           file=NamedFile(data=open(TEST_LETTER_ODT).read(),
                                          filename=u'letter.odt',
                                          contentType='application/vnd.oasis.opendocument.text'),
                           sender_name="Thomas Desvenain",
                           sender_address="57 Quai du Pré Long",
                           recipient_name="Vincent Fretin",
                           date=datetime.date(2012, 12, 23))

        notify(ObjectModifiedEvent(content))
        generated_stream = content.unrestrictedTraverse('@@getdocumentfusion')()
        self.assertTrue(generated_stream)
        self.assertEqual(self.portal.REQUEST.response['content-type'],
                         'application/pdf')
        generated_path = tempfile.mktemp(suffix='letter.pdf')
        generated_file = open(generated_path, 'w')
        generated_file.write(generated_stream.read())
        generated_file.close()

        txt_path = tempfile.mktemp(suffix='letter.pdf')
        subprocess.call(['pdftotext', generated_path, txt_path])
        txt = open(txt_path).read()
        self.assertIn('Vincent Fretin', txt)
        self.assertIn('57 Quai du Pré Long', txt)
        self.assertIn('2012', txt)
        self.assertIn(u'EN RÉPONSE...', txt)

        os.remove(txt_path)
        os.remove(generated_path)

def test_plot_anat():
    img = _generate_img()

    # Test saving with empty plot
    z_slicer = plot_anat(anat_img=False, display_mode='z')
    filename = tempfile.mktemp(suffix='.png')
    try:
        z_slicer.savefig(filename)
    finally:
        os.remove(filename)

    z_slicer = plot_anat(display_mode='z')
    filename = tempfile.mktemp(suffix='.png')
    try:
        z_slicer.savefig(filename)
    finally:
        os.remove(filename)

    ortho_slicer = plot_anat(img, dim='auto')
    filename = tempfile.mktemp(suffix='.png')
    try:
        ortho_slicer.savefig(filename)
    finally:
        os.remove(filename)

    # Save execution time and memory
    plt.close()

    def test_build(self):
        client = api.Client(url=url, config_path=tempfile.mktemp(), max_size=1024 * 256)  # 256K
        client.register()
        generated = client.build(cleanup=True)
        self.assertTrue(len(generated))

        client = api.Client(url=url, config_path=tempfile.mktemp(), max_size=1024 * 512)  # 512K
        config = client.config()
        client.register()
        generated = client.build(cleanup=True)
        self.assertTrue(len(generated) == 4)

        result = json.loads(urlopen(url + "/api/online/json").read().decode("utf8"))
        result = [farmers for farmers in result["farmers"] if farmers["btc_addr"] == config["payout_address"]]
        last_seen = result[0]["last_seen"]
        result = json.dumps(result, sort_keys=True)
        expected = json.dumps(
            [
                {
                    "height": 4,
                    "btc_addr": config["payout_address"],
                    "last_seen": last_seen,
                    "payout_addr": config["payout_address"],
                }
            ],
            sort_keys=True,
        )

        self.assertEqual(result, expected)  # check that build send height=4 to the online list

 def test_output_compatible_setup_3(self):
     tmpfile = tempfile.mktemp(prefix='avocado_' + __name__)
     tmpfile2 = tempfile.mktemp(prefix='avocado_' + __name__)
     tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__)
     tmpfile3 = tempfile.mktemp(dir=tmpdir)
     os.chdir(basedir)
     cmd_line = ('./scripts/avocado run --job-results-dir %s --sysinfo=off '
                 '--xunit %s --json %s --html %s passtest.py' %
                 (self.tmpdir, tmpfile, tmpfile2, tmpfile3))
     result = process.run(cmd_line, ignore_status=True)
     output = result.stdout + result.stderr
     expected_rc = exit_codes.AVOCADO_ALL_OK
     tmpdir_contents = os.listdir(tmpdir)
     self.assertEqual(len(tmpdir_contents), 5,
                      'Not all resources dir were created: %s' % tmpdir_contents)
     try:
         self.assertEqual(result.exit_status, expected_rc,
                          "Avocado did not return rc %d:\n%s" %
                          (expected_rc, result))
         self.assertNotEqual(output, "", "Output is empty")
         # Check if we are producing valid outputs
         with open(tmpfile2, 'r') as fp:
             json_results = json.load(fp)
             debug_log = json_results['debuglog']
             self.check_output_files(debug_log)
         minidom.parse(tmpfile)
     finally:
         try:
             os.remove(tmpfile)
             os.remove(tmpfile2)
             shutil.rmtree(tmpdir)
         except OSError:
             pass

    def __init__( self, com, debug=0, **params ):

        self.com = com

        self.rec_psf = com.rec().getPsfFile()
        self.lig_psf = com.lig().getPsfFile()

        recCode = com.rec().getPdbCode()
        ligCode = com.lig().getPdbCode()

        self.rec_in = tempfile.mktemp( recCode + ".pdb" )
        self.lig_in = tempfile.mktemp( ligCode + ".pdb" )

        self.lig_out = tempfile.mktemp( "lig_out.pdb" )
        self.rec_out = tempfile.mktemp( "rec_out.pdb" )

        self.inp_template = t.dataRoot() +\
            '/xplor/rb_minimize_complex.inp'

        self.param19 = t.dataRoot() + \
            '/xplor/toppar/param19.pro'

        self.result = None

        Xplorer.__init__( self, self.inp_template, debug=debug, **params )

    def test_fit_to_mesh_file_errorsII(self):
        from anuga.load_mesh.loadASCII import import_mesh_file, export_mesh_file
        import tempfile
        import os

        # create a .tsh file, no user outline
        mesh_file = tempfile.mktemp(".tsh")
        fd = open(mesh_file,'w')
        fd.write("unit testing a bad .tsh file \n")
        fd.close()

        # create a points .csv file
        point_file = tempfile.mktemp(".csv")
        fd = open(point_file,'w')
        fd.write("x,y,elevation, stage \n\
        1.0, 1.0,2.,4 \n\
        1.0, 3.0,4,8 \n\
        3.0,1.0,4.,8 \n")
        fd.close()

        mesh_output_file = "new_triangle.tsh"
        try:
            fit_to_mesh_file(mesh_file, point_file,
                             mesh_output_file, display_errors = False)
        except IOError:
            pass
        else:
            raise Exception('Bad file did not raise error!')
            
        #clean up
        os.remove(mesh_file)
        os.remove(point_file)

def make_video(url, target):
    temp_gif = tempfile.mktemp(suffix=".gif")
    temp_output = tempfile.mktemp(suffix=".mp4")
    try:
        # download file
        subprocess.check_call(["timeout", "-s", "KILL", "5s",
                               "curl", "-o", temp_gif, url])

        ffprobe_output = subprocess.check_output(
            ["timeout", "-s", "KILL", "5s", "ffprobe", "-show_packets", temp_gif])

        frame_count = ffprobe_output.count(b"codec_type=video")
        if frame_count <= 5:
            return False

        # and convert
        subprocess.check_call(["timeout", "-s", "KILL", "30s",
                               "ffmpeg", "-y", "-i", temp_gif, "-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2",
                               "-codec:v", "libx264", "-preset", "medium", "-b:v", "350k", "-an",
                               "-profile:v", "baseline", "-level", "3.0", "-pix_fmt", "yuv420p",
                               "-qmin", "20", "-qmax", "42", temp_output])

        # and move to target
        shutil.copy(temp_output, str(target))
        return True
    except:
        stats.increment(metric_name("error"))
        raise

    finally:
        for temp in temp_gif, temp_output:
            try:
                os.unlink(temp)
            except OSError:
                pass

def test_invert_saveload():
    dclab.PolygonFilter.clear_all_filters()
    ddict = example_data_dict(size=1234, keys=["area_um", "deform"])
    # points of polygon filter
    points = [[np.min(ddict["area_um"]), np.min(ddict["deform"])],
              [np.min(ddict["area_um"]), np.max(ddict["deform"])],
              [np.average(ddict["area_um"]), np.max(ddict["deform"])],
              [np.average(ddict["area_um"]), np.min(ddict["deform"])],
              ]
    filt1 = dclab.PolygonFilter(axes=["area_um", "deform"],
                                points=points,
                                inverted=True)
    name = tempfile.mktemp(prefix="test_dclab_polygon_")
    filt1.save(name)
    filt2 = dclab.PolygonFilter(filename=name)
    assert filt2 == filt1

    filt3 = dclab.PolygonFilter(axes=["area_um", "deform"],
                                points=points,
                                inverted=False)
    try:
        os.remove(name)
    except OSError:
        pass

    name = tempfile.mktemp(prefix="test_dclab_polygon_")
    filt3.save(name)
    filt4 = dclab.PolygonFilter(filename=name)
    assert filt4 == filt3
    try:
        os.remove(name)
    except OSError:
        pass

def setup_for_testing(require_indexes=True):
  """Sets up the stubs for testing.

  Args:
    require_indexes: True if indexes should be required for all indexes.
  """
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import memcache
  from google.appengine.tools import dev_appserver
  from google.appengine.tools import dev_appserver_index
  import urlfetch_test_stub
  before_level = logging.getLogger().getEffectiveLevel()
  try:
    logging.getLogger().setLevel(100)
    root_path = os.path.realpath(os.path.dirname(__file__))
    dev_appserver.SetupStubs(
        TEST_APP_ID,
        root_path=root_path,
        login_url='',
        datastore_path=tempfile.mktemp(suffix='datastore_stub'),
        history_path=tempfile.mktemp(suffix='datastore_history'),
        blobstore_path=tempfile.mktemp(suffix='blobstore_stub'),
        require_indexes=require_indexes,
        clear_datastore=False)
    dev_appserver_index.SetupIndexes(TEST_APP_ID, root_path)
    apiproxy_stub_map.apiproxy._APIProxyStubMap__stub_map['urlfetch'] = \
        urlfetch_test_stub.instance
    # Actually need to flush, even though we've reallocated. Maybe because the
    # memcache stub's cache is at the module level, not the API stub?
    memcache.flush_all()
  finally:
    logging.getLogger().setLevel(before_level)

     def generate_image():
         infile = mktemp(prefix='%s-' % name)
         outfile = mktemp(prefix='%s-' % name)
         try:
             try:
                 f = codecs.open(infile, 'w', 'utf8')
                 try:
                     f.write(content)
                 finally:
                     f.close()
                 cmd = [name, '-a', '-T', type, '-o', outfile, infile]
                 if font:
                     cmd.extend(['-f', font])
                 self.env.log.debug('(%s) command: %r' % (name, cmd))
                 try:
                     proc = Popen(cmd, stderr=PIPE)
                     stderr_value = proc.communicate()[1]
                 except Exception, e:
                     self.env.log.error('(%s) %r' % (name, e))
                     raise ImageGenerationError("Failed to generate diagram. (%s is not found.)" % name)
 
                 if proc.returncode != 0 or not os.path.isfile(outfile):
                     self.env.log.error('(%s) %s' % (name, stderr_value))
                     raise ImageGenerationError("Failed to generate diagram. (rc=%d)" % proc.returncode)
                 f = open(outfile, 'rb')
                 try:
                     data = f.read()
                 finally:
                     f.close()
                 return data
             except ImageGenerationError:
                 raise
             except Exception, e:
                 self.env.log.error('(%s) %r' % (name, e))
                 raise ImageGenerationError("Failed to generate diagram.")

    def run_locally(self, options, handle_files):
        temp_files = []
        for name in handle_files:
            tmp = mktemp(suffix='', dir=TEMP_DIR)
            args = ["kbhs-download", "--handle", name, "-o", tmp]
            subprocess.check_call(args)
            temp_files.append(tmp)

        result_files = list(self._run_locally_internal(options, temp_files))

        if self.cleanup:
            for name in temp_files:
                os.remove(name)

        handles = []
        for name in result_files:
            tmp = mktemp(suffix='.handle', dir=TEMP_DIR)
            args = ["kbhs-upload", "-i", name, "-o", tmp]
            subprocess.check_call(args)
            fh = open(tmp)
            h = json.load(fh)
            fh.close()
            handles.append(h)
            if self.cleanup:
                os.remove(tmp)
                os.remove(name)

        return handles

def get_surface(pdb_file, PDB_TO_XYZR="pdb_to_xyzr", MSMS="msms"):
    """
    Return a Numeric array that represents
    the vertex list of the molecular surface.

    PDB_TO_XYZR --- pdb_to_xyzr executable (arg. to os.system)
    MSMS --- msms executable (arg. to os.system)
    """
    # extract xyz and set radii
    xyz_tmp=tempfile.mktemp()
    PDB_TO_XYZR=PDB_TO_XYZR+" %s > %s"
    make_xyz=PDB_TO_XYZR % (pdb_file, xyz_tmp)
    os.system(make_xyz)
    assert os.path.isfile(xyz_tmp), \
        "Failed to generate XYZR file using command:\n%s" % make_xyz
    # make surface
    surface_tmp=tempfile.mktemp()
    MSMS=MSMS+" -probe_radius 1.5 -if %s -of %s > "+tempfile.mktemp()
    make_surface=MSMS % (xyz_tmp, surface_tmp)
    os.system(make_surface)
    surface_file=surface_tmp+".vert"
    assert os.path.isfile(surface_file), \
        "Failed to generate surface file using command:\n%s" % make_surface
    # read surface vertices from vertex file
    surface=_read_vertex_array(surface_file)
    # clean up tmp files
    # ...this is dangerous
    #os.system("rm "+xyz_tmp)
    #os.system("rm "+surface_tmp+".vert")
    #os.system("rm "+surface_tmp+".face")
    return surface

    def test_rename_sliced_histogram_using_rename_method(self):
        "Histogram: rename()"
        import tempfile
        outh5 = tempfile.mktemp()
        code = """
from histogram import histogram, arange
h = histogram(
  'h', 
  [('x', arange(10), 'meter'),
   ('y', arange(15), 'meter'),
  ]
  )
h1 = h[(2,5), ()].sum('x')
h1.rename('abc')
from histogram.hdf import load, dump
dump(h1, %r, '/', 'c')
""" % outh5
        script = tempfile.mktemp()
        open(script, 'w').write(code)
        cmd = 'python %s' % script
        import os
        
        if os.system(cmd):
            raise RuntimeError, "%s failed" % cmd
        
        from histogram.hdf import load
        try:
            h = load(outh5, 'abc')
        except:
            raise RuntimeError, "failed to load histogram from %s" %(
                outh5,)

        os.remove(outh5)
        os.remove(script)
        return        

def test_kfolds():
    from b4msa.command_line import params, kfolds
    import os
    import sys
    import json
    import tempfile
    output = tempfile.mktemp()
    fname = os.path.dirname(__file__) + '/text.json'
    sys.argv = ['b4msa', '-o', output, '-k', '2', fname, '-s', '2']
    params()
    output2 = tempfile.mktemp()
    sys.argv = ['b4msa', '-m', output, fname, '-o', output2]
    print(output, fname)
    kfolds()
    os.unlink(output)
    a = open(output2).readline()
    os.unlink(output2)
    a = json.loads(a)
    assert 'decision_function' in a
    sys.argv = ['b4msa', '--update-klass', '-m', output, fname, '-o', output2]
    try:
        kfolds()
    except AssertionError:
        return
    assert False

def create_app(name=None, urls=None, py_version='26', app_dir='/opt/apps',
    ve_dir='/opt/ve', app_port=8000):
    """
    Creates a new application container

    :param name: Name of application
    :param urls: Application public URLs (semi-colon separated - i.e. "example.com;anotherexample.com")
    :param py_version: Version of Python to use (default: 26)
    :param app_dir: Root directory for applications (default: /opt/apps)
    :param ve_dir: Root directory for virtualenvs (default: /opt/ve)
    :param app_port: Application port

    """
    if not name or not urls:
        raise RuntimeError('You must specify an name and urls')
    with default_settings():
        # create app directory
        sudo('mkdir -p {0}'.format(os.path.join(app_dir, name)))
        # create supervisor config
        uwsgi_tmp_conf = tempfile.mktemp()
        with open(uwsgi_tmp_conf, 'w') as f:
            f.write(generate_uwsgi_config(app_name=name, app_dir=app_dir, ve_dir=ve_dir, user='www-data', group='www-data'))
        nginx_tmp_conf = tempfile.mktemp()
        # create nginx config
        with open(nginx_tmp_conf, 'w') as f:
            f.write(generate_nginx_config(app_name=name, urls=urls.split(';'),
                app_port=app_port))
        put(uwsgi_tmp_conf, '/etc/supervisor/conf.d/uwsgi-{0}.conf'.format(name), mode=0755, use_sudo=True)
        put(nginx_tmp_conf, '/etc/nginx/conf.d/{0}.conf'.format(name), mode=0755, use_sudo=True)
        # update supervisor
        sudo('supervisorctl update')
        # cleanup
        os.remove(uwsgi_tmp_conf)
        os.remove(nginx_tmp_conf)

    def getPDF(self):
        """
        Fetch PDF and save it locally in a temporary file.
        Tries by order:
        - refereed article
        - refereed article using another machine (set ssh_user & ssh_server)
        - arXiv preprint
        - electronic journal link
        """

        if not self.links:
            return 'failed'

        def filetype(filename):
            return subprocess.Popen('file %s' % filename, shell=True,
                                    stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE).stdout.read()

        #refereed
        if 'article' in self.links:
            url = self.links['article']
            #try locally
            pdf = tempfile.mktemp() + '.pdf'
            urllib.urlretrieve(url, pdf)
            if 'PDF document' in filetype(pdf):
                return pdf

            #try in remote server
            # you need to set SSH public key authentication
            # for this to work!
            elif ssh_user is not None:
                pdf = tempfile.mktemp() + '.pdf'
                cmd = 'ssh %[email protected]%s \"touch toto.pdf; wget -O toto.pdf \\"%s\\"\"' % (ssh_user, ssh_server, url)
                cmd2 = 'scp -q %[email protected]%s:toto.pdf %s' % (ssh_user, ssh_server, pdf)
                subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
                subprocess.Popen(cmd2, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
                if 'PDF document' in filetype(pdf):
                    return pdf

        #arXiv
        if 'preprint' in self.links:
            #arXiv page
            url = self.links['preprint']
            for line in urllib.urlopen(url).readlines():
                if 'dc:identifier' in line:
                    begin = re.search('dc:identifier="', line).end()
                    url = line[begin:-2].replace('&', '&').lower()
                    #get arXiv PDF
                    pdf = tempfile.mktemp() + '.pdf'
                    urllib.urlretrieve(url.replace('abs', 'pdf'), pdf)
                    if 'PDF document' in filetype(pdf):
                        return pdf
                    else:
                        return url

        #electronic journal
        if 'ejournal' in self.links:
            return self.links['ejournal']

        return 'failed'

def hex_to_obj(_hex):
    if len(_hex) == 0 or len(_hex) % 2 != 0:
        raise Exception('Not valid _hex: %s' % _hex)

    tmp_shell_fp = mktemp()
    tmp_obj_fp = mktemp()

    asm = '.byte '
    for i in xrange(0, len(_hex), 2):
        asm += '0x%s,' % _hex[i:i+2]
    asm = asm.rstrip(',')
    asm += '\n'

    f = open(tmp_shell_fp, 'w')
    f.write(asm)
    f.close()

    cmd = 'as --32 %s -o %s' % (tmp_shell_fp, tmp_obj_fp)
    os.system(cmd)

    obj = open(tmp_obj_fp, 'rb').read()

    os.unlink(tmp_shell_fp)
    os.unlink(tmp_obj_fp)

    return obj

def show_longdesc():
    if not HAVE_README:
        logging.error(
            "To check the long description, we need the 'readme' package. "
            "(It is included if you install `zest.releaser[recommended]`)"
        )
        sys.exit(1)

    filename = tempfile.mktemp(".html")
    # Note: for the setup.py call we use _execute_command() from our
    # utils module. This makes sure the python path is set up right.
    longdesc = _execute_command(utils.setup_py("--long-description"))
    warnings = io.StringIO()
    html = render(longdesc, warnings)
    if html is None:
        logging.error("Error generating html. Invalid ReST.")
        rst_filename = tempfile.mktemp(".rst")
        with open(rst_filename, "wb") as rst_file:
            rst_file.write(longdesc.encode("utf-8"))
        warning_text = warnings.getvalue()
        warning_text = warning_text.replace("<string>", rst_filename)
        print(warning_text)
        sys.exit(1)

    if "<html" not in html[:20]:
        # Add a html declaration including utf-8 indicator
        html = HTML_PREFIX + html + HTML_POSTFIX

    with open(filename, "wb") as fh:
        fh.write(html.encode("utf-8"))

    url = "file://" + filename
    logging.info("Opening %s in your webbrowser.", url)
    webbrowser.open(url)

    def test_convert(self):
        tmp = tempfile.mktemp()

        output = romanesco.convert("image", {
            "format": "png.base64",
            "data": self.image
        }, {
            "format": "png",
            "url": "file://" + tmp,
            "mode": "auto"
        })

        value = open(tmp).read()
        os.remove(tmp)
        self.assertEqual(output["format"], "png")
        self.assertEqual(base64.b64encode(value), self.image)

        output = romanesco.convert(
            "image",
            {"format": "png.base64", "data": self.image},
            {"format": "pil"})

        tmp = tempfile.mktemp()

        output = romanesco.convert(
            "image",
            output,
            {"format": "png"})

        io1 = StringIO(base64.b64decode(self.image))
        im1 = Image.open(io1)
        io2 = StringIO(output["data"])
        im2 = Image.open(io2)
        self.assertEqual(compareImages(im1, im2), 0)