def testEscape(self):
		self.assertEqual("", model.escape(""))
		self.assertEqual("hello", model.escape("hello"))
		self.assertEqual("%20", model.escape(" "))

		self.assertEqual("file%3a%2f%2ffoo%7ebar",
				model.escape("file://foo~bar"))
		self.assertEqual("file%3a%2f%2ffoo%25bar",
				model.escape("file://foo%bar"))

		self.assertEqual("file:##foo%7ebar",
				model._pretty_escape("file://foo~bar"))
		self.assertEqual("file:##foo%25bar",
				model._pretty_escape("file://foo%bar"))

	def testTimes(self):
		iface_cache = self.config.iface_cache
		with tempfile.TemporaryFile() as stream:
			stream.write(data.thomas_key)
			stream.seek(0)
			gpg.import_key(stream)

		upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
		cached = os.path.join(upstream_dir, model.escape('http://foo'))

		with open(cached, 'wb') as stream:
			stream.write(data.foo_signed_xml)

		signed = iface_cache._get_signature_date('http://foo')
		assert signed == None

		trust.trust_db.trust_key(
			'92429807C9853C0744A68B9AAE07828059A53CC1')

		signed = iface_cache._get_signature_date('http://foo')
		self.assertEqual(1154850229, signed)

		with open(cached, 'w+b') as stream:
			stream.seek(0)
			stream.write(b'Hello')

		# When the signature is invalid, we just return None.
		# This is because versions < 0.22 used to corrupt the signatue
		# by adding an attribute to the XML
		signed = iface_cache._get_signature_date('http://foo')
		assert signed == None

	def get_icon_path(self, iface):
		"""Get the path of a cached icon for an interface.
		@param iface: interface whose icon we want
		@return: the path of the cached icon, or None if not cached.
		@rtype: str"""
		return basedir.load_first_cache(config_site, 'interface_icons',
						 escape(iface.uri))

			def get_inputs():
				for sel in sels.selections.values():
					logger.info("Checking %s", sel.feed)

					if sel.feed.startswith('distribution:'):
						# If the package has changed version, we'll detect that below
						# with get_unavailable_selections.
						pass
					elif os.path.isabs(sel.feed):
						# Local feed
						yield sel.feed
					else:
						# Cached feed
						cached = basedir.load_first_cache(namespaces.config_site, 'interfaces', model.escape(sel.feed))
						if cached:
							yield cached
						else:
							raise IOError("Input %s missing; update" % sel.feed)

					# Per-feed configuration
					yield basedir.load_first_config(namespaces.config_site, namespaces.config_prog,
									   'interfaces', model._pretty_escape(sel.interface))

				# Global configuration
				yield basedir.load_first_config(namespaces.config_site, namespaces.config_prog, 'global')

	def cache_iface(self, name, data):
		cached_ifaces = basedir.save_cache_path('0install.net',
							'interfaces')

		f = open(os.path.join(cached_ifaces, model.escape(name)), 'w')
		f.write(data)
		f.close()

def update_user_overrides(interface):
	"""Update an interface with user-supplied information.
	Sets preferred stability and updates extra_feeds.
	@param interface: the interface object to update
	@type interface: L{model.Interface}
	"""
	user = basedir.load_first_config(config_site, config_prog,
					   'interfaces', model._pretty_escape(interface.uri))
	if user is None:
		# For files saved by 0launch < 0.49
		user = basedir.load_first_config(config_site, config_prog,
						   'user_overrides', escape(interface.uri))
	if not user:
		return

	try:
		root = qdom.parse(file(user))
	except Exception as ex:
		warn(_("Error reading '%(user)s': %(exception)s"), {'user': user, 'exception': ex})
		raise

	stability_policy = root.getAttribute('stability-policy')
	if stability_policy:
		interface.set_stability_policy(stability_levels[str(stability_policy)])

	for item in root.childNodes:
		if item.uri != XMLNS_IFACE: continue
		if item.name == 'feed':
			feed_src = item.getAttribute('src')
			if not feed_src:
				raise InvalidInterface(_('Missing "src" attribute in <feed>'))
			interface.extra_feeds.append(Feed(feed_src, item.getAttribute('arch'), True, langs = item.getAttribute('langs')))

	def testXMLupdate(self):
		iface_cache = self.config.iface_cache
		trust.trust_db.trust_key(
			'92429807C9853C0744A68B9AAE07828059A53CC1')
		with tempfile.TemporaryFile() as stream:
			stream.write(data.thomas_key)
			stream.seek(0)
			gpg.import_key(stream)

		iface = iface_cache.get_interface('http://foo')
		with tempfile.TemporaryFile() as src:
			src.write(data.foo_signed_xml)
			src.seek(0)
			pending = PendingFeed(iface.uri, src)
			assert iface_cache.update_feed_if_trusted(iface.uri, pending.sigs, pending.new_xml)

		iface_cache.__init__()
		feed = iface_cache.get_feed('http://foo')
		assert feed.last_modified == 1154850229

		# mtimes are unreliable because copying often changes them -
		# check that we extract the time from the signature when upgrading
		upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
		cached = os.path.join(upstream_dir, model.escape(feed.url))
		os.utime(cached, None)

		iface_cache.__init__()
		feed = iface_cache.get_feed('http://foo')
		assert feed.last_modified > 1154850229

		with tempfile.TemporaryFile() as src:
			src.write(data.new_foo_signed_xml)
			src.seek(0)

			pending = PendingFeed(feed.url, src)

			old_stdout = sys.stdout
			sys.stdout = StringIO()
			try:
				assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml, dry_run = True)
			finally:
				sys.stdout = old_stdout

			assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml)

		# Can't 'update' to an older copy
		with tempfile.TemporaryFile() as src:
			src.write(data.foo_signed_xml)
			src.seek(0)
			try:
				pending = PendingFeed(feed.url, src)
				assert iface_cache.update_feed_if_trusted(feed.url, pending.sigs, pending.new_xml)

				assert 0
			except model.SafeException:
				pass

	def _import_new_feed(self, feed_url, new_xml, modified_time):
		"""Write new_xml into the cache.
		@param feed_url: the URL for the feed being updated
		@param new_xml: the data to write
		@param modified_time: when new_xml was modified
		@raises ReplayAttack: if the new mtime is older than the current one
		"""
		assert modified_time
		assert isinstance(new_xml, bytes), repr(new_xml)

		upstream_dir = basedir.save_cache_path(config_site, 'interfaces')
		cached = os.path.join(upstream_dir, escape(feed_url))

		old_modified = None
		if os.path.exists(cached):
			with open(cached, 'rb') as stream:
				old_xml = stream.read()
			if old_xml == new_xml:
				logger.debug(_("No change"))
				# Update in-memory copy, in case someone else updated the disk copy
				self.get_feed(feed_url, force = True)
				return
			old_modified = int(os.stat(cached).st_mtime)

		# Do we need to write this temporary file now?
		try:
			with open(cached + '.new', 'wb') as stream:
				stream.write(new_xml)
			os.utime(cached + '.new', (modified_time, modified_time))
			new_mtime = reader.check_readable(feed_url, cached + '.new')
			assert new_mtime == modified_time

			old_modified = self._get_signature_date(feed_url) or old_modified

			if old_modified:
				if new_mtime < old_modified:
					raise ReplayAttack(_("New feed's modification time is "
						"before old version!\nInterface: %(iface)s\nOld time: %(old_time)s\nNew time: %(new_time)s\n"
						"Refusing update.")
						% {'iface': feed_url, 'old_time': _pretty_time(old_modified), 'new_time': _pretty_time(new_mtime)})
				if new_mtime == old_modified:
					# You used to have to update the modification time manually.
					# Now it comes from the signature, this check isn't useful
					# and often causes problems when the stored format changes
					# (e.g., when we stopped writing last-modified attributes)
					pass
					#raise SafeException("Interface has changed, but modification time "
					#		    "hasn't! Refusing update.")
		except:
			os.unlink(cached + '.new')
			raise

		portable_rename(cached + '.new', cached)
		logger.debug(_("Saved as %s") % cached)

		self.get_feed(feed_url, force = True)

 def delete(self):
     if not os.path.isabs(self.uri):
         cached_iface = basedir.load_first_cache(namespaces.config_site, "interfaces", model.escape(self.uri))
         if cached_iface:
             # print "Delete", cached_iface
             os.unlink(cached_iface)
     user_overrides = basedir.load_first_config(
         namespaces.config_site, namespaces.config_prog, "user_overrides", model.escape(self.uri)
     )
     if user_overrides:
         # print "Delete", user_overrides
         os.unlink(user_overrides)

		def download_and_add_icon():
			stream = dl.tempfile
			yield dl.downloaded
			try:
				tasks.check(dl.downloaded)
				if dl.unmodified: return
				stream.seek(0)

				import shutil
				icons_cache = basedir.save_cache_path(config_site, 'interface_icons')
				icon_file = file(os.path.join(icons_cache, escape(interface.uri)), 'w')
				shutil.copyfileobj(stream, icon_file)
			except Exception as ex:
				self.handler.report_error(ex)

def load_feed_from_cache(url, selections_ok = False):
	"""Load a feed. If the feed is remote, load from the cache. If local, load it directly.
	@return: the feed, or None if it's remote and not cached."""
	try:
		if os.path.isabs(url):
			debug(_("Loading local feed file '%s'"), url)
			return load_feed(url, local = True, selections_ok = selections_ok)
		else:
			cached = basedir.load_first_cache(config_site, 'interfaces', escape(url))
			if cached:
				debug(_("Loading cached information for %(interface)s from %(cached)s"), {'interface': url, 'cached': cached})
				return load_feed(cached, local = False)
			else:
				return None
	except InvalidInterface, ex:
		ex.feed_url = url
		raise

	def delete(self):
		if not os.path.isabs(self.uri):
			cached_iface = basedir.load_first_cache(namespaces.config_site,
					'interfaces', model.escape(self.uri))
			if cached_iface:
				if SAFE_MODE:
					print("Delete", cached_iface)
				else:
					os.unlink(cached_iface)
		user_overrides = basedir.load_first_config(namespaces.config_site,
					namespaces.config_prog,
					'interfaces', model._pretty_escape(self.uri))
		if user_overrides:
			if SAFE_MODE:
				print("Delete", user_overrides)
			else:
				os.unlink(user_overrides)

def update_user_overrides(interface, known_site_feeds=frozenset()):
    """Update an interface with user-supplied information.
	Sets preferred stability and updates extra_feeds.
	@param interface: the interface object to update
	@type interface: L{model.Interface}
	@param known_site_feeds: feeds to ignore (for backwards compatibility)
	"""
    user = basedir.load_first_config(config_site, config_prog, "interfaces", model._pretty_escape(interface.uri))
    if user is None:
        # For files saved by 0launch < 0.49
        user = basedir.load_first_config(config_site, config_prog, "user_overrides", escape(interface.uri))
    if not user:
        return

    try:
        with open(user, "rb") as stream:
            root = qdom.parse(stream)
    except Exception as ex:
        logger.warn(_("Error reading '%(user)s': %(exception)s"), {"user": user, "exception": ex})
        raise

    stability_policy = root.getAttribute("stability-policy")
    if stability_policy:
        interface.set_stability_policy(stability_levels[str(stability_policy)])

    for item in root.childNodes:
        if item.uri != XMLNS_IFACE:
            continue
        if item.name == "feed":
            feed_src = item.getAttribute("src")
            if not feed_src:
                raise InvalidInterface(_('Missing "src" attribute in <feed>'))
                # (note: 0install 1.9..1.12 used a different scheme and the "site-package" attribute;
                # we deliberately use a different attribute name to avoid confusion)
            if item.getAttribute("is-site-package"):
                # Site packages are detected earlier. This test isn't completely reliable,
                # since older versions will remove the attribute when saving the config
                # (hence the next test).
                continue
            if feed_src in known_site_feeds:
                continue
            interface.extra_feeds.append(
                Feed(feed_src, item.getAttribute("arch"), True, langs=item.getAttribute("langs"))
            )

    def import_feed(self, url, contents):
        """contents can be a path or an Element."""
        iface_cache = self.config.iface_cache
        iface_cache.get_interface(url)

        if isinstance(contents, qdom.Element):
            feed = model.ZeroInstallFeed(contents)
        else:
            feed = reader.load_feed(contents)

        iface_cache._feeds[url] = feed

        xml = qdom.to_UTF8(feed.feed_element)
        upstream_dir = basedir.save_cache_path(namespaces.config_site, "interfaces")
        cached = os.path.join(upstream_dir, model.escape(url))
        with open(cached, "wb") as stream:
            stream.write(xml)

        return feed

		def download_and_add_icon():
			stream = dl.tempfile
			try:
				yield dl.downloaded
				tasks.check(dl.downloaded)
				if dl.unmodified: return
				stream.seek(0)

				import shutil, tempfile
				icons_cache = basedir.save_cache_path(config_site, 'interface_icons')

				tmp_file = tempfile.NamedTemporaryFile(dir = icons_cache, delete = False)
				shutil.copyfileobj(stream, tmp_file)
				tmp_file.close()

				icon_file = os.path.join(icons_cache, escape(interface.uri))
				portable_rename(tmp_file.name, icon_file)
			finally:
				stream.close()

	def get_cached_signatures(self, uri):
		"""Verify the cached interface using GPG.
		Only new-style XML-signed interfaces retain their signatures in the cache.
		@param uri: the feed to check
		@type uri: str
		@return: a list of signatures, or None
		@rtype: [L{gpg.Signature}] or None
		@since: 0.25"""
		import gpg
		if os.path.isabs(uri):
			old_iface = uri
		else:
			old_iface = basedir.load_first_cache(config_site, 'interfaces', escape(uri))
			if old_iface is None:
				return None
		try:
			return gpg.check_stream(file(old_iface))[1]
		except SafeException, ex:
			debug(_("No signatures (old-style interface): %s") % ex)
			return None

def update_user_overrides(interface):
	"""Update an interface with user-supplied information.
	Sets preferred stability and updates extra_feeds.
	@param interface: the interface object to update
	@type interface: L{model.Interface}
	"""
	user = basedir.load_first_config(config_site, config_prog,
					   'interfaces', model._pretty_escape(interface.uri))
	if user is None:
		# For files saved by 0launch < 0.49
		user = basedir.load_first_config(config_site, config_prog,
						   'user_overrides', escape(interface.uri))
	if not user:
		return

	try:
		root = qdom.parse(file(user))
	except Exception, ex:
		warn(_("Error reading '%(user)s': %(exception)s"), {'user': user, 'exception': ex})
		raise

def update_user_feed_overrides(feed):
	"""Update a feed with user-supplied information.
	Sets last_checked and user_stability ratings.
	@param feed: feed to update
	@since 0.49
	"""
	user = basedir.load_first_config(config_site, config_prog,
					   'feeds', model._pretty_escape(feed.url))
	if user is None:
		# For files saved by 0launch < 0.49
		user = basedir.load_first_config(config_site, config_prog,
						   'user_overrides', escape(feed.url))
	if not user:
		return

	try:
		root = qdom.parse(file(user))
	except Exception, ex:
		warn(_("Error reading '%(user)s': %(exception)s"), {'user': user, 'exception': ex})
		raise

def update_user_feed_overrides(feed):
    """Update a feed with user-supplied information.
	Sets last_checked and user_stability ratings.
	@param feed: feed to update
	@since 0.49
	"""
    user = basedir.load_first_config(config_site, config_prog, "feeds", model._pretty_escape(feed.url))
    if user is None:
        # For files saved by 0launch < 0.49
        user = basedir.load_first_config(config_site, config_prog, "user_overrides", escape(feed.url))
    if not user:
        return

    try:
        with open(user, "rb") as stream:
            root = qdom.parse(stream)
    except Exception as ex:
        logger.warn(_("Error reading '%(user)s': %(exception)s"), {"user": user, "exception": ex})
        raise

    last_checked = root.getAttribute("last-checked")
    if last_checked:
        feed.last_checked = int(last_checked)

    for item in root.childNodes:
        if item.uri != XMLNS_IFACE:
            continue
        if item.name == "implementation":
            id = item.getAttribute("id")
            assert id is not None
            impl = feed.implementations.get(id, None)
            if not impl:
                logger.debug(
                    _("Ignoring user-override for unknown implementation %(id)s in %(interface)s"),
                    {"id": id, "interface": feed},
                )
                continue

            user_stability = item.getAttribute("user-stability")
            if user_stability:
                impl.user_stability = stability_levels[str(user_stability)]

def load_feed_from_cache(url):
    """Load a feed. If the feed is remote, load from the cache. If local, load it directly.
	@type url: str
	@return: the feed, or None if it's remote and not cached.
	@rtype: L{ZeroInstallFeed} | None"""
    try:
        if os.path.isabs(url):
            logger.debug(_("Loading local feed file '%s'"), url)
            return load_feed(url, local=True)
        else:
            cached = basedir.load_first_cache(config_site, "interfaces", escape(url))
            if cached:
                logger.debug(
                    _("Loading cached information for %(interface)s from %(cached)s"),
                    {"interface": url, "cached": cached},
                )
                return load_feed(cached, local=False)
            else:
                return None
    except InvalidInterface as ex:
        ex.feed_url = url
        raise