class _unicode_func_wrapper(object):
"""
Wraps a function, converts arguments from unicode to bytes,
- and return values to unicode from bytes.
+ and return values to unicode from bytes. Function calls
+ will raise UnicodeEncodeError if an argument fails to be
+ encoded with the required encoding. Return values that
+ are single strings are decoded with errors='replace'. Return
+ values that are lists of strings are decoded with errors='strict'
+ and elements that fail to be decoded are omitted from the returned
+ list.
"""
__slots__ = ('_func', '_encoding')
def __call__(self, *args, **kwargs):
encoding = self._encoding
- wrapped_args = [_unicode_encode(x) for x in args]
+ wrapped_args = [_unicode_encode(x, encoding=encoding, errors='strict')
+ for x in args]
if kwargs:
- wrapped_kwargs = dict((_unicode_encode(k, encoding=encoding),
- _unicode_encode(v, encoding=encoding)) \
+ wrapped_kwargs = dict(
+ (_unicode_encode(k, encoding=encoding, errors='strict'),
+ _unicode_encode(v, encoding=encoding, errors='strict'))
for k, v in kwargs.iteritems())
else:
wrapped_kwargs = {}
if isinstance(rval, (basestring, list, tuple)):
if isinstance(rval, basestring):
- rval = _unicode_decode(rval, encoding=encoding)
+ rval = _unicode_decode(rval,
+ encoding=encoding, errors='replace')
else:
decoded_rval = []
for x in rval:
import os as _os
_os_overrides = {
id(_os.fdopen) : _os.fdopen,
+ id(_os.popen) : _os.popen,
id(_os.read) : _os.read,
id(_os.system) : _os.system,
}
import os
import shutil
-import portage
+from portage import _content_encoding
+from portage import _fs_encoding
+from portage import _unicode_encode
from portage.localization import _
import selinux
-from selinux import is_selinux_enabled, getfilecon, lgetfilecon
+from selinux import is_selinux_enabled
def copyfile(src, dest):
- src = portage._unicode_encode(src)
- dest = portage._unicode_encode(dest)
+ src = _unicode_encode(src, encoding=_fs_encoding, errors='strict')
+ dest = _unicode_encode(dest, encoding=_fs_encoding, errors='strict')
(rc, ctx) = selinux.lgetfilecon(src)
if rc < 0:
raise OSError(_("copyfile: Failed getting context of \"%s\".") % src)
return ctx
def mkdir(target, refdir):
- target = portage._unicode_encode(target)
- refdir = portage._unicode_encode(refdir)
+ target = _unicode_encode(target, encoding=_fs_encoding, errors='strict')
+ refdir = _unicode_encode(refdir, encoding=_fs_encoding, errors='strict')
(rc, ctx) = selinux.getfilecon(refdir)
if rc < 0:
raise OSError(
_("mkdir: Failed getting context of reference directory \"%s\".") \
% refdir)
- setfscreatecon(ctx)
+ selinux.setfscreatecon(ctx)
try:
os.mkdir(target)
finally:
- setfscreatecon()
+ selinux.setfscreatecon()
def rename(src, dest):
- src = portage._unicode_encode(src)
- dest = portage._unicode_encode(dest)
+ src = _unicode_encode(src, encoding=_fs_encoding, errors='strict')
+ dest = _unicode_encode(dest, encoding=_fs_encoding, errors='strict')
(rc, ctx) = selinux.lgetfilecon(src)
if rc < 0:
raise OSError(_("rename: Failed getting context of \"%s\".") % src)
return ":".join(ret)
def setexec(ctx="\n"):
- if isinstance(ctx, unicode):
- ctx = ctx.encode('utf_8', 'replace')
+ ctx = _unicode_encode(ctx, encoding=_content_encoding, errors='strict')
if selinux.setexeccon(ctx) < 0:
raise OSError(_("setexec: Failed setting exec() context \"%s\".") % ctx)
def setfscreate(ctx="\n"):
- ctx = portage._unicode_encode(ctx)
+ ctx = _unicode_encode(ctx,
+ encoding=_content_encoding, errors='strict')
if selinux.setfscreatecon(ctx) < 0:
raise OSError(
_("setfscreate: Failed setting fs create context \"%s\".") % ctx)
def spawn_wrapper(spawn_func, selinux_type):
+ selinux_type = _unicode_encode(selinux_type,
+ encoding=_content_encoding, errors='strict')
+
def wrapper_func(*args, **kwargs):
- selinux_type = portage._unicode_encode(selinux_type)
con = settype(selinux_type)
setexec(con)
try:
return wrapper_func
def symlink(target, link, reflnk):
- target = portage._unicode_encode(target)
- link = portage._unicode_encode(link)
- reflnk = portage._unicode_encode(reflnk)
+ target = _unicode_encode(target, encoding=_fs_encoding, errors='strict')
+ link = _unicode_encode(link, encoding=_fs_encoding, errors='strict')
+ reflnk = _unicode_encode(reflnk, encoding=_fs_encoding, errors='strict')
(rc, ctx) = selinux.lgetfilecon(reflnk)
if rc < 0:
raise OSError(
# Library by Wayne Davison <gentoo@blorf.net>, derived from code
# written by Jeremy Wohl (http://igmus.org)
-from stat import *
import os, sys, commands, shutil
import portage, portage.const
# This puts the results of the merge into mrgconf.
ret = os.system(RCS_MERGE % (archive, mrgconf))
mystat = os.lstat(newconf)
- os.chmod(mrgconf, mystat[ST_MODE])
- os.chown(mrgconf, mystat[ST_UID], mystat[ST_GID])
+ os.chmod(mrgconf, mystat.st_mode)
+ os.chown(mrgconf, mystat.st_uid, mystat.st_gid)
os.rename(archive, archive + '.dist.new')
return ret
# This puts the results of the merge into mrgconf.
ret = os.system(DIFF3_MERGE % (curconf, archive + '.dist', newconf, mrgconf))
mystat = os.lstat(newconf)
- os.chmod(mrgconf, mystat[ST_MODE])
- os.chown(mrgconf, mystat[ST_UID], mystat[ST_GID])
+ os.chmod(mrgconf, mystat.st_mode)
+ os.chown(mrgconf, mystat.st_uid, mystat.st_gid)
return ret
"hardlock_name", "hardlink_is_mine", "hardlink_lockfile", \
"unhardlink_lockfile", "hardlock_cleanup"]
-import errno, os, stat, time
+import errno
+import stat
+import time
+from portage import os
from portage.exception import DirectoryNotFound, FileNotFound, \
InvalidData, TryAgain, OperationNotPermitted, PermissionDenied
from portage.data import portage_gid
import codecs
import logging
-import os
+import os as _os
import re
+from portage import os
+from portage import _content_encoding
+from portage import _fs_encoding
+from portage import _unicode_decode
+from portage import _unicode_encode
from portage.util import apply_secpass_permissions, ensure_dirs, \
grabfile, normalize_path, write_atomic, writemsg_level
from portage.data import portage_gid
news_dir = self._news_dir(repoid)
try:
- news = os.listdir(news_dir)
+ news = _os.listdir(_unicode_encode(news_dir,
+ encoding=_fs_encoding, errors='strict'))
except OSError:
return
updates = []
for itemid in news:
+ try:
+ itemid = _unicode_decode(itemid,
+ encoding=_fs_encoding, errors='strict')
+ except UnicodeDecodeError:
+ itemid = _unicode_decode(itemid,
+ encoding=_fs_encoding, errors='replace')
+ writemsg_level(
+ "!!! Invalid encoding in news item name: '%s'\n" % \
+ itemid, level=logging.ERROR, noiselevel=-1)
+ continue
+
if itemid in skip:
continue
filename = os.path.join(news_dir, itemid,
itemid + "." + self.language_id + ".txt")
if not os.path.isfile(filename):
continue
- if not isinstance(itemid, unicode):
- itemid = unicode(itemid, encoding='utf_8', errors='replace')
item = NewsItem(filename, itemid)
if not item.isValid():
continue
if unread_lock:
unlockfile(unread_lock)
+_formatRE = re.compile("News-Item-Format:\s*([^\s]*)\s*$")
_installedRE = re.compile("Display-If-Installed:(.*)\n")
_profileRE = re.compile("Display-If-Profile:(.*)\n")
_keywordRE = re.compile("Display-If-Keyword:(.*)\n")
"display if arch: x86" and so forth.
Creation of a news item involves passing in the path to the particular news item.
-
"""
def __init__(self, path, name):
and a vardb so we can look at installed packages).
Each restriction will pluck out the items that are required for it to match
or raise a ValueError exception if the required object is not present.
+
+ Restrictions of the form Display-X are OR'd with like-restrictions;
+ otherwise restrictions are AND'd. any_match is the ORing and
+ all_match is the ANDing.
"""
if not self._parsed:
self.parse()
if not len(self.restrictions):
- return True # no restrictions to match means everyone should see it
+ return True
kwargs = \
{ 'vardb' : vardb,
'config' : config,
'profile' : profile }
- for restriction in self.restrictions:
- if restriction.checkRestriction(**kwargs):
- return True
+ all_match = True
+ for values in self.restrictions.itervalues():
+ any_match = False
+ for restriction in values:
+ if restriction.checkRestriction(**kwargs):
+ any_match = True
+ if not any_match:
+ all_match = False
- return False # No restrictions were met; thus we aren't relevant :(
+ return all_match
def isValid(self):
if not self._parsed:
return self._valid
def parse(self):
- lines = codecs.open(self.path, mode='r',
- encoding='utf_8', errors='replace').readlines()
- self.restrictions = []
+ lines = codecs.open(_unicode_encode(self.path,
+ encoding=_fs_encoding, errors='strict'),
+ mode='r', encoding=_content_encoding, errors='replace').readlines()
+ self.restrictions = {}
invalids = []
for i, line in enumerate(lines):
- #Optimization to ignore regex matchines on lines that
- #will never match
+ # Optimization to ignore regex matchines on lines that
+ # will never match
+ format_match = _formatRE.match(line)
+ if format_match is not None and format_match.group(1) != '1.0':
+ invalids.append((i + 1, line.rstrip('\n')))
+ break
if not line.startswith('D'):
continue
restricts = { _installedRE : DisplayInstalledRestriction,
for regex, restriction in restricts.iteritems():
match = regex.match(line)
if match:
- self.restrictions.append(restriction(match.groups()[0].strip()))
- if not self.restrictions[-1].isValid():
+ restrict = restriction(match.groups()[0].strip())
+ if not restrict.isValid():
invalids.append((i + 1, line.rstrip("\n")))
+ else:
+ self.restrictions.setdefault(
+ id(restriction), []).append(restrict)
continue
if invalids:
self._valid = False
import commands
import errno
import formatter
-import os
import re
-import shlex
import sys
import portage
'portage.util:writemsg',
)
+from portage import os
+from portage import _content_encoding
+from portage import _fs_encoding
+from portage import _unicode_encode
from portage.const import COLOR_MAP_FILE, EPREFIX
from portage.exception import CommandNotFound, FileNotFound, \
ParseError, PermissionDenied, PortageException
_styles["PROMPT_CHOICE_DEFAULT"] = ( "green", )
_styles["PROMPT_CHOICE_OTHER"] = ( "red", )
-def _parse_color_map(onerror=None):
+def _parse_color_map(config_root='/', onerror=None):
"""
Parse /etc/portage/color.map and return a dict of error codes.
# that can be called in order adjust the location that color.map
# is read from.
global codes, _styles
- myfile = os.path.join(EPREFIX, COLOR_MAP_FILE)
+ myfile = os.path.join(config_root, COLOR_MAP_FILE)
ansi_code_pattern = re.compile("^[0-9;]*m$")
quotes = '\'"'
def strip_quotes(token):
return token
try:
lineno=0
- for line in codecs.open( myfile, mode='r',
- encoding='utf_8', errors='replace' ):
+ for line in codecs.open(_unicode_encode(myfile,
+ encoding=_fs_encoding, errors='strict'),
+ mode='r', encoding=_content_encoding, errors='replace'):
lineno += 1
commenter_pos = line.find("#")
def _write(self, f, s):
if sys.hexversion < 0x3000000 and isinstance(s, unicode):
# avoid potential UnicodeEncodeError
- s = s.encode('utf_8', 'replace')
+ s = s.encode(_content_encoding, 'replace')
f.write(s)
f.flush()
">" + ((max_bar_width - bar_width) * " ") + "]"
return image
-try:
- _parse_color_map(onerror=lambda e: writemsg("%s\n" % str(e), noiselevel=-1))
-except FileNotFound:
- pass
-except PermissionDenied, e:
- writemsg(_("Permission denied: '%s'\n") % str(e), noiselevel=-1)
- del e
-except PortageException, e:
- writemsg("%s\n" % str(e), noiselevel=-1)
- del e
+_color_map_loaded = False
+
+def _init(config_root='/'):
+ """
+ Load color.map from the given config_root. This is called automatically
+ on first access of the codes or _styles attributes (unless it has already
+ been called for some other reason).
+ """
+
+ global _color_map_loaded, codes, _styles
+ if _color_map_loaded:
+ return
+
+ _color_map_loaded = True
+ codes = object.__getattribute__(codes, '_attr')
+ _styles = object.__getattribute__(_styles, '_attr')
+
+ try:
+ _parse_color_map(config_root=config_root,
+ onerror=lambda e: writemsg("%s\n" % str(e), noiselevel=-1))
+ except FileNotFound:
+ pass
+ except PermissionDenied, e:
+ writemsg(_("Permission denied: '%s'\n") % str(e), noiselevel=-1)
+ del e
+ except PortageException, e:
+ writemsg("%s\n" % str(e), noiselevel=-1)
+ del e
+
+class _LazyInitColorMap(portage.proxy.objectproxy.ObjectProxy):
+
+ __slots__ = ('_attr',)
+
+ def __init__(self, attr):
+ portage.proxy.objectproxy.ObjectProxy.__init__(self)
+ object.__setattr__(self, '_attr', attr)
+
+ def _get_target(self):
+ _init()
+ return object.__getattribute__(self, '_attr')
+
+codes = _LazyInitColorMap(codes)
+_styles = _LazyInitColorMap(_styles)
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import os, shutil, sys
-from os.path import dirname, abspath, join
+from portage import os
+from portage import shutil
from portage.tests import TestCase
from portage.process import spawn
from portage.const import PORTAGE_BIN_PATH
-bindir = join(dirname(dirname(abspath(__file__))),
+bindir = os.path.join(os.path.dirname(os.path.dirname(
+ os.path.abspath(__file__))),
"..", "..", "..", "bin", "ebuild-helpers")
-basedir = join(dirname(dirname(abspath(__file__))), "bin", "root")
+basedir = os.path.join(os.path.dirname(os.path.dirname(
+ os.path.abspath(__file__))), "bin", "root")
os.environ["D"] = os.path.join(basedir, "image")
os.environ["T"] = os.path.join(basedir, "temp")
os.environ["S"] = os.path.join(basedir, "workdir")
def portage_func(func, args, exit_status=0):
# we don't care about the output of the programs,
# just their exit value and the state of $D
- f = open('/dev/null', 'w')
+ f = open('/dev/null', 'wb')
fd_pipes = {0:0,1:f.fileno(),2:f.fileno()}
spawn(func+" "+args, env=os.environ, fd_pipes=fd_pipes)
f.close()
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import errno, os, sys
+import codecs
+import errno
+import sys
+from portage import os
+from portage import _content_encoding
+from portage import _fs_encoding
+from portage import _unicode_encode
from portage.tests import TestCase
class SpawnTestCase(TestCase):
spawn("echo -n '%s'" % test_string, settings, logfile=logfile,
free=1, fd_pipes={0:sys.stdin.fileno(), 1:null_fd, 2:null_fd})
os.close(null_fd)
- f = open(logfile, 'r')
+ f = codecs.open(_unicode_encode(logfile,
+ encoding=_fs_encoding, errors='strict'),
+ mode='r', encoding=_content_encoding, errors='strict')
log_content = f.read()
f.close()
# When logging passes through a pty, this comparison will fail
# Distributed under the terms of the GNU General Public License v2
# $Id$
+from portage import os
from portage.tests import TestCase
from portage.env.config import PackageKeywordsFile
from tempfile import mkstemp
-import os
class PackageKeywordsFileTestCase(TestCase):
f.close()
def NukeFile(self):
- import os
os.unlink(self.fname)
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import os
-
+from portage import os
from portage.env.config import PackageMaskFile
from portage.tests import TestCase, test_cps
from tempfile import mkstemp
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import os
+from portage import os
from portage.tests import TestCase
from portage.env.config import PackageUseFile
from tempfile import mkstemp
+# Copyright 2006-2009 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+# $Id$
+from portage import os
from portage.tests import TestCase
from portage.env.config import PortageModulesFile
from tempfile import mkstemp
from itertools import izip
-import os
class PortageModulesFileTestCase(TestCase):
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import os
+from portage import os
from portage.tests import TestCase
from portage.news import NewsItem
-from portage.const import PROFILE_PATH
from portage.dbapi.virtual import testdbapi
from tempfile import mkstemp
# TODO(antarus) Make newsitem use a loader so we can load using a string instead of a tempfile
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import errno, os, sys
-import fcntl
-import termios
-import portage
-from portage.output import get_term_size, set_term_size
+import sys
+from portage import os
from portage.tests import TestCase
from _emerge.TaskScheduler import TaskScheduler
from _emerge.PipeReader import PipeReader
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import tempfile, os
+import tempfile
+from portage import os
from portage.tests import TestCase, test_cps
from portage.sets.files import ConfigFileSet
# Distributed under the terms of the GNU General Public License v2
# $Id$
-import tempfile, os
+import tempfile
+from portage import os
from portage.tests import TestCase, test_cps
from portage.sets.files import StaticFileSet
-from portage.env.loaders import TestTextLoader
-from portage.env.config import ConfigLoaderKlass
class StaticFileSetTestCase(TestCase):
"""Simple Test Case for StaticFileSet"""
# Distributed under the terms of the GNU General Public License v2
# $Id$
+from portage import os
from portage.tests import TestCase
from portage.util import unique_array
test portage.util.uniqueArray()
"""
- import os
-
tests = [ ( ["a","a","a",os,os,[],[],[]], ['a',os,[]] ),
( [1,1,1,2,3,4,4] , [1,2,3,4]) ]
projects / portage.git / commitdiff