def test_get_file_and_validate_it(self):
    """Tests get_file from a url, plus extraction and validation.
    """
    dest_dir = self.get_temp_dir()
    orig_dir = self.get_temp_dir()

    text_file_path = os.path.join(orig_dir, 'test.txt')
    zip_file_path = os.path.join(orig_dir, 'test.zip')
    tar_file_path = os.path.join(orig_dir, 'test.tar.gz')

    with open(text_file_path, 'w') as text_file:
      text_file.write('Float like a butterfly, sting like a bee.')

    with tarfile.open(tar_file_path, 'w:gz') as tar_file:
      tar_file.add(text_file_path)

    with zipfile.ZipFile(zip_file_path, 'w') as zip_file:
      zip_file.write(text_file_path)

    origin = urljoin('file://', pathname2url(os.path.abspath(tar_file_path)))

    path = keras.utils.data_utils.get_file('test.txt', origin,
                                           untar=True, cache_subdir=dest_dir)
    filepath = path + '.tar.gz'
    hashval_sha256 = keras.utils.data_utils._hash_file(filepath)
    hashval_md5 = keras.utils.data_utils._hash_file(filepath, algorithm='md5')
    path = keras.utils.data_utils.get_file(
        'test.txt', origin, md5_hash=hashval_md5,
        untar=True, cache_subdir=dest_dir)
    path = keras.utils.data_utils.get_file(
        filepath, origin, file_hash=hashval_sha256,
        extract=True, cache_subdir=dest_dir)
    self.assertTrue(os.path.exists(filepath))
    self.assertTrue(keras.utils.data_utils.validate_file(filepath,
                                                         hashval_sha256))
    self.assertTrue(keras.utils.data_utils.validate_file(filepath, hashval_md5))
    os.remove(filepath)

    origin = urljoin('file://', pathname2url(os.path.abspath(zip_file_path)))

    hashval_sha256 = keras.utils.data_utils._hash_file(zip_file_path)
    hashval_md5 = keras.utils.data_utils._hash_file(zip_file_path,
                                                    algorithm='md5')
    path = keras.utils.data_utils.get_file(
        'test', origin, md5_hash=hashval_md5,
        extract=True, cache_subdir=dest_dir)
    path = keras.utils.data_utils.get_file(
        'test', origin, file_hash=hashval_sha256,
        extract=True, cache_subdir=dest_dir)
    self.assertTrue(os.path.exists(path))
    self.assertTrue(keras.utils.data_utils.validate_file(path, hashval_sha256))
    self.assertTrue(keras.utils.data_utils.validate_file(path, hashval_md5))

    def test_create_with_json_file_uri(self):
        # The contents of env_v2.json must be equivalent to ENVIRONMENT
        path = pkg.resource_filename(
            'mistralclient',
            'tests/unit/resources/env_v2.json'
        )

        # Convert the file path to file URI
        uri = parse.urljoin('file:', request.pathname2url(path))
        data = OrderedDict(
            utils.load_content(
                utils.get_contents_if_file(uri)
            )
        )

        mock = self.mock_http_post(content=data)
        file_input = {'file': uri}
        env = self.environments.create(**file_input)

        self.assertIsNotNone(env)

        expected_data = copy.deepcopy(data)
        expected_data['variables'] = json.dumps(expected_data['variables'])

        mock.assert_called_once_with(URL_TEMPLATE, json.dumps(expected_data))

def path_to_url(path):
    """Convert a system path to a URL."""

    if os.path.sep == '/':
        return path

    return pathname2url(path)

def get_swagger_spec(settings):
    """Return a :class:`bravado_core.spec.Spec` constructed from
    the swagger specs in `pyramid_swagger.schema_directory`. If
    `pyramid_swagger.enable_swagger_spec_validation` is enabled the schema
    will be validated before returning it.

    :param settings: a pyramid registry settings with configuration for
        building a swagger schema
    :type settings: dict
    :rtype: :class:`bravado_core.spec.Spec`
    """
    schema_dir = settings.get('pyramid_swagger.schema_directory', 'api_docs/')
    schema_filename = settings.get('pyramid_swagger.schema_file',
                                   'swagger.json')
    schema_path = os.path.join(schema_dir, schema_filename)
    schema_url = urlparse.urljoin('file:', pathname2url(os.path.abspath(schema_path)))

    handlers = build_http_handlers(None)  # don't need http_client for file:
    file_handler = handlers['file']
    spec_dict = file_handler(schema_url)

    return Spec.from_dict(
        spec_dict,
        config=create_bravado_core_config(settings),
        origin_url=schema_url)

    def test_create_with_json_file_uri(self):
        # The contents of env_v2.json must be equivalent to ENVIRONMENT
        path = pkg.resource_filename(
            'mistralclient',
            'tests/unit/resources/env_v2.json'
        )

        # Convert the file path to file URI
        uri = parse.urljoin('file:', request.pathname2url(path))
        data = collections.OrderedDict(
            utils.load_content(
                utils.get_contents_if_file(uri)
            )
        )

        self.requests_mock.post(self.TEST_URL + URL_TEMPLATE,
                                status_code=201,
                                json=data)
        file_input = {'file': uri}
        env = self.environments.create(**file_input)

        self.assertIsNotNone(env)

        expected_data = copy.deepcopy(data)
        expected_data['variables'] = json.dumps(expected_data['variables'])

        self.assertEqual(expected_data, self.requests_mock.last_request.json())

def test_generate_hashes_with_editable():
    small_fake_package_dir = os.path.join(
        os.path.split(__file__)[0], 'test_data', 'small_fake_package')
    small_fake_package_url = 'file:' + pathname2url(small_fake_package_dir)
    runner = CliRunner()
    with runner.isolated_filesystem():
        with open('requirements.in', 'w') as fp:
            fp.write('-e {}\n'.format(small_fake_package_url))
            fp.write('pytz==2017.2\n')
        out = runner.invoke(
            cli, [
                '--generate-hashes',
                '--index-url', PyPIRepository.DEFAULT_INDEX_URL,
            ],
        )
    expected = (
        '#\n'
        '# This file is autogenerated by pip-compile\n'
        '# To update, run:\n'
        '#\n'
        '#    pip-compile --generate-hashes --output-file requirements.txt requirements.in\n'
        '#\n'
        '-e {}\n'
        'pytz==2017.2 \\\n'
        '    --hash=sha256:d1d6729c85acea5423671382868627129432fba9a89ecbb248d8d1c7a9f01c67 \\\n'
        '    --hash=sha256:f5c056e8f62d45ba8215e5cb8f50dfccb198b4b9fbea8500674f3443e4689589\n'
    ).format(small_fake_package_url)
    assert out.exit_code == 0
    assert expected in out.output

def path_to_file_uri(path):
    """Convert local filesystem path to legal File URIs as described in:
    http://en.wikipedia.org/wiki/File_URI_scheme
    """
    x = pathname2url(os.path.abspath(path))
    if os.name == 'nt':
        x = x.replace('|', ':') # http://bugs.python.org/issue5861
    return 'file:///%s' % x.lstrip('/')

    def _authenticated_server_proxy(self):
        r"""
        Get an XML-RPC proxy object that is authenticated using the users
        username and password.

        EXAMPLES::

            sage: dev.trac._authenticated_server_proxy  # not tested
            Trac username: username
            Trac password:
            Should I store your password in a configuration file for future sessions? (This configuration file might be readable by privileged users on this system.) [yes/No]
            <ServerProxy for trac.sagemath.org/login/xmlrpc>

        TESTS:

        To make sure that doctests do not tamper with the live trac
        server, it is an error to access this property during a
        doctest (The ``dev`` object during doctests is also modified
        to prevent this)::

            sage: from sage.dev.test.config import DoctestConfig
            sage: from sage.dev.test.user_interface import DoctestUserInterface
            sage: from sage.dev.trac_interface import TracInterface
            sage: config = DoctestConfig()
            sage: trac = TracInterface(config['trac'], DoctestUserInterface(config['UI']))
            sage: trac._authenticated_server_proxy
            Traceback (most recent call last):
            ...
            AssertionError: doctest tried to access an authenticated session to trac
        """
        import sage.doctest
        assert not sage.doctest.DOCTEST_MODE, \
            "doctest tried to access an authenticated session to trac"

        self._check_password_timeout()

        if self.__authenticated_server_proxy is None:
            from sage.env import REALM
            realm = self._config.get('realm', REALM)
            from sage.env import TRAC_SERVER_URI
            server = self._config.get('server', TRAC_SERVER_URI)
            url = urljoin(server, pathname2url(os.path.join('login', 'xmlrpc')))
            while True:
                from xmlrpclib import ServerProxy
                from digest_transport import DigestTransport
                from trac_error import TracAuthenticationError
                transport = DigestTransport()
                transport.add_authentication(realm=realm, url=server, username=self._username, password=self._password)
                proxy = ServerProxy(url, transport=transport)
                try:
                    proxy.system.listMethods()
                    break
                except TracAuthenticationError:
                    self._UI.error("Invalid username/password")
                    self.reset_username()
            self.__authenticated_server_proxy = proxy
        self._postpone_password_timeout()
        return self.__authenticated_server_proxy

 def absolutize(self, uri, defrag=1):
     base = urljoin("file:", pathname2url(os.getcwd()))
     result = urljoin("%s/" % base, uri, allow_fragments=not defrag)
     if defrag:
         result = urldefrag(result)[0]
     if not defrag:
         if uri and uri[-1] == "#" and result[-1] != "#":
             result = "%s#" % result
     return URIRef(result)

 def setUp(self):
     """Setup."""
     self.gcold = gc.isenabled()
     gc.collect()
     gc.disable()
     self.graph = Graph(store=self.store)
     if self.path is None:
         self.path = pathname2url(mkdtemp())
     self.graph.open(self.path)
     self.input = Graph()

def test_data_utils(in_tmpdir):
    """Tests get_file from a url, plus extraction and validation.
    """
    dirname = 'data_utils'

    with open('test.txt', 'w') as text_file:
        text_file.write('Float like a butterfly, sting like a bee.')

    with tarfile.open('test.tar.gz', 'w:gz') as tar_file:
        tar_file.add('test.txt')

    with zipfile.ZipFile('test.zip', 'w') as zip_file:
        zip_file.write('test.txt')

    origin = urljoin('file://', pathname2url(os.path.abspath('test.tar.gz')))

    path = get_file(dirname, origin, untar=True)
    filepath = path + '.tar.gz'
    hashval_sha256 = _hash_file(filepath)
    hashval_md5 = _hash_file(filepath, algorithm='md5')
    path = get_file(dirname, origin, md5_hash=hashval_md5, untar=True)
    path = get_file(filepath, origin, file_hash=hashval_sha256, extract=True)
    assert os.path.exists(filepath)
    assert validate_file(filepath, hashval_sha256)
    assert validate_file(filepath, hashval_md5)
    os.remove(filepath)
    os.remove('test.tar.gz')

    origin = urljoin('file://', pathname2url(os.path.abspath('test.zip')))

    hashval_sha256 = _hash_file('test.zip')
    hashval_md5 = _hash_file('test.zip', algorithm='md5')
    path = get_file(dirname, origin, md5_hash=hashval_md5, extract=True)
    path = get_file(dirname, origin, file_hash=hashval_sha256, extract=True)
    assert os.path.exists(path)
    assert validate_file(path, hashval_sha256)
    assert validate_file(path, hashval_md5)

    os.remove(path)
    os.remove('test.txt')
    os.remove('test.zip')

def test_editable_package(tmpdir):
    """ piptools can compile an editable """
    fake_package_dir = os.path.join(os.path.split(__file__)[0], 'test_data', 'small_fake_package')
    fake_package_dir = 'file:' + pathname2url(fake_package_dir)
    runner = CliRunner()
    with runner.isolated_filesystem():
        with open('requirements.in', 'w') as req_in:
            req_in.write('-e ' + fake_package_dir)  # require editable fake package

        out = runner.invoke(cli, ['-n'])

        assert out.exit_code == 0
        assert fake_package_dir in out.output
        assert 'six==1.10.0' in out.output

    def on_confirm_add_directory_dialog(self, action):
        if action == 0:
            for torrent_file in self.selected_torrent_files:
                escaped_uri = u"file:%s" % pathname2url(torrent_file.encode('utf-8'))
                self.perform_start_download_request(escaped_uri,
                                                    self.window().tribler_settings['download_defaults'][
                                                        'anonymity_enabled'],
                                                    self.window().tribler_settings['download_defaults'][
                                                        'safeseeding_enabled'],
                                                    self.tribler_settings['download_defaults']['saveas'], [], 0)

        if self.dialog:
            self.dialog.close_dialog()
            self.dialog = None

    def test_live_downloads(self):
        QTest.mouseClick(window.left_menu_button_home, Qt.LeftButton)
        QTest.mouseClick(window.home_tab_torrents_button, Qt.LeftButton)
        self.screenshot(window, name="home_page_torrents_loading")

        # Start downloading some torrents
        if 'TORRENTS_DIR' in os.environ:
            torrent_dir = os.environ.get('TORRENTS_DIR')
        else:
            torrent_dir = os.path.join(os.path.join(os.path.dirname(__file__), os.pardir), "data", "linux_torrents")
        window.selected_torrent_files = [pathname2url(torrent_file)
                                         for torrent_file in glob.glob(torrent_dir + "/*.torrent")]

        window.on_confirm_add_directory_dialog(0)

        self.go_to_and_wait_for_downloads()
        QTest.qWait(2000)

        with open(output_file, "w") as output:
            output.write("time, upload, download\n")

            def download_refreshed(_):
                line = "%s, %s, %s\n" % (time.time(), window.downloads_page.total_upload/1000,
                                         window.downloads_page.total_download/1000)
                output.write(line)

            window.downloads_page.received_downloads.connect(download_refreshed)
            QTest.qWait(test_timeout)

            # Stop downloads after timeout
            window.downloads_page.received_downloads.disconnect()
            window.downloads_page.stop_loading_downloads()
            QTest.qWait(5000)

        # Plot graph
        data = numpy.genfromtxt(output_file, delimiter=',', skip_header=1,
                                skip_footer=0, names=['time', 'upload', 'download'])
        figure = plot.figure()
        subplot = figure.add_subplot(111)

        subplot.set_title("Live downloads plot")
        subplot.set_xlabel('Time (seconds)')
        subplot.set_ylabel('Speed (kB/s)')

        subplot.plot(data['time'], data['upload'], color='g', label='upload')
        subplot.plot(data['time'], data['download'], color='r', label='download')

        subplot.legend()
        figure.savefig(output_file + '.png', bbox_inches='tight')

def get_contents_if_file(contents_or_file_name):
    """由文件名获取utf8解码的byte字符串
    """
    try:
        if parse.urlparse(contents_or_file_name).scheme:
            definition_url = contents_or_file_name
        else:
            path = os.path.abspath(contents_or_file_name)
            # 'file:///home/jf/scripts/test/my_async.yaml'
            definition_url = parse.urljoin(
                'file:',
                request.pathname2url(path)
            )
        return request.urlopen(definition_url).read().decode('utf8')
    except Exception:
        return contents_or_file_name

    def deploy(self, wgt_file):

        template_content = wgt_file.get_template()
        template_parser = TemplateParser(template_content)

        widget_rel_dir = os.path.join(
            template_parser.get_resource_vendor(),
            template_parser.get_resource_name(),
            template_parser.get_resource_version(),
        )
        widget_dir = os.path.join(self._root_dir, widget_rel_dir)
        template_parser.set_base(pathname2url(widget_rel_dir) + '/')

        self._create_folders(widget_dir)
        wgt_file.extract(widget_dir)

        return template_parser

def get_link_path(target, base):
    """Returns a relative path to a target from a base.

    If base is an existing file, then its parent directory is considered.
    Otherwise, base is assumed to be a directory.

    Rationale: os.path.relpath is not available before Python 2.6
    """
    path =  _get_pathname(target, base)
    # Windows Python 3 pathname2url doesn't accept bytes
    url = pathname2url(path if PY3 else path.encode('UTF-8'))
    if os.path.isabs(path):
        url = 'file:' + url
    # At least Jython seems to use 'C|/Path' and not 'C:/Path'
    if os.sep == '\\' and '|/' in url:
        url = url.replace('|/', ':/', 1)
    return url.replace('%5C', '/').replace('%3A', ':').replace('|', ':')

    def write_item(self, item):
        title = item.get('title', 'Untitled')
        header = """<html lang="en">
<head>
<meta charset="utf-8" />
<title>%s</title>
</head>
<body>
""" % title

        body = self.make_body(item, title)
        closer = """
</body>
</html>
"""

        url = item['location']
        media_guid = hashlib.sha1(url).hexdigest()
        media_ext = '.html' 
        path = 'full/%s%s' % (media_guid, media_ext)
        absolute_path = os.path.join(self.store.basedir, path)
        with codecs.open(absolute_path, 'wb', 'utf-8') as f:
            f.write(header)
            f.write(body)
            f.write(closer)

        item['inline_urls']  = [ urljoin('file://', pathname2url(absolute_path)) ]
        item['inline_metas'] = [ { 'link_url': item['request_url'], 'location': item['location'], 
                'title': title, 'content_type': 'text/html'} ]

        checksum = None
        with open(absolute_path, 'rb') as f:
            checksum = md5sum(f)

        # Compatible with Twisted Deferred results
        results = [
            (True, {'url': url,
                'path': path,
                'checksum': checksum }
            )
        ]

        item = self.item_completed(results, item, self.spiderinfo)
        return item

def validate_swagger_schema(schema_dir, resource_listing):
    """Validate the structure of Swagger schemas against the spec.

    **Valid only for Swagger v1.2 spec**

    Note: It is possible that resource_listing is not present in
    the schema_dir. The path is passed in the call so that ssv
    can fetch the api-declaration files from the path.

    :param resource_listing: Swagger Spec v1.2 resource listing
    :type resource_listing: dict
    :param schema_dir: A path to Swagger spec directory
    :type schema_dir: string
    :raises: :py:class:`swagger_spec_validator.SwaggerValidationError`
    """
    schema_filepath = os.path.join(schema_dir, API_DOCS_FILENAME)
    swagger_spec_validator.validator12.validate_spec(
        resource_listing,
        urlparse.urljoin('file:', pathname2url(os.path.abspath(schema_filepath))),
    )

    def test_file_string_or_uri(self):
        data = '{ "some": "data here"}'
        with tempfile.NamedTemporaryFile(delete=False) as tmp:
            tmp.write(data.encode('utf-8'))
            tmp.close()

            output = _load_file_string_or_uri(tmp.name, 'test')
            self.assertEqual(get_file_json(tmp.name), output)

            uri = urljoin('file:', pathname2url(tmp.name))
            output = _load_file_string_or_uri(uri, 'test')
            self.assertEqual(get_file_json(tmp.name), output)

            os.unlink(tmp.name)

        output = _load_file_string_or_uri(data, 'test')
        self.assertEqual(shell_safe_json_parse(data), output)

        self.assertEqual(None, _load_file_string_or_uri(None, 'test', required=False))
        self.assertRaises(CLIError, _load_file_string_or_uri, None, 'test')