portage-2.2
==================================
-* Portage now requires >=python-2.4, but doesn't need pycrypto anymore if
- >=python-2.5 is installed and openssl supports the rmd160 hash.
* The python namespace for portage has been sanitized, all portage related code
is now contained within the portage namespace. External script should be updated
accordingly, though links exist for backward compability.
portage-2.1.3
==================================
+* Portage now requires >=python-2.4, but doesn't need pycrypto anymore if
+ >=python-2.5 is installed and openssl supports the rmd160 hash.
* The "save_summary" and "echo" elog modules are now enabled by default. Setting
PORTAGE_ELOG_SYSTEM in make.conf will override this, so if you don't want elog
set PORTAGE_ELOG_SYSTEM="" in make.conf
# failed to select any nodes. It is reset whenever nodes are
# successfully selected.
prefer_asap = True
+
+ # By default, try to avoid selecting root nodes whenever possible. This
+ # helps ensure that the maximimum possible number of soft dependencies
+ # have been removed from the graph before their parent nodes have
+ # selected. This is especially important when those dependencies are
+ # going to be rebuilt by revdep-rebuild or `emerge -e system` after the
+ # CHOST has been changed (like when building a stage3 from a stage2).
+ accept_root_node = False
+
+ # State of prefer_asap and accept_root_node flags for successive
+ # iterations that loosen the criteria for node selection.
+ #
+ # iteration prefer_asap accept_root_node
+ # 1 True False
+ # 2 False False
+ # 3 False True
+ #
+ # If no nodes are selected on the 3rd iteration, it is due to
+ # unresolved blockers or circular dependencies.
+
while not mygraph.empty():
selected_nodes = None
if prefer_asap and asap_nodes:
# found a non-root node
selected_nodes = [node]
break
- if not selected_nodes:
+ if not selected_nodes and \
+ (accept_root_node or ignore_priority is None):
# settle for a root node
selected_nodes = [nodes[0]]
if not selected_nodes:
for ignore_priority in xrange(DepPriority.SOFT,
DepPriority.MEDIUM_SOFT + 1):
for node in nodes:
+ if not accept_root_node and \
+ not mygraph.parent_nodes(node):
+ continue
selected_nodes = set()
if gather_deps(ignore_priority,
mergeable_nodes, selected_nodes, node):
prefer_asap = False
continue
+ if not selected_nodes and not accept_root_node:
+ # Maybe there are only root nodes left, so accept them
+ # for the next iteration.
+ accept_root_node = True
+ continue
+
if selected_nodes and ignore_priority > DepPriority.SOFT:
# Try to merge ignored medium deps as soon as possible.
for node in selected_nodes:
sys.exit(1)
# At this point, we've succeeded in selecting one or more nodes, so
- # it's now safe to reset the prefer_asap to it's default state.
+ # it's now safe to reset the prefer_asap and accept_root_node flags
+ # to their default states.
prefer_asap = True
+ accept_root_node = False
for node in selected_nodes:
if node[-1] != "nomerge":
# Blacklist vars that could interfere with portage internals.
for blacklisted in "CATEGORY", "PKGUSE", "PORTAGE_CONFIGROOT", \
- "ROOT", "EPREFIX", "EROOT":
+ "ROOT", "SANDBOX_DENY", "SANDBOX_PREDICT", "SANDBOX_READ", \
+ "SANDBOX_WRITE", "EPREFIX", "EROOT":
for cfg in self.lookuplist:
cfg.pop(blacklisted, None)
del blacklisted, cfg
# Distributed under the terms of the GNU General Public License v2
# $Id$
-
+import os
+import re
from portage.dep import dep_getslot, dep_getkey, match_from_list
from portage.locks import unlockfile
from portage.output import red
from portage.util import writemsg
-
from portage import dep_expand
+from portage.versions import catsplit
-import os, re
class dbapi(object):
def __init__(self):
return
def cpv_all(self):
+ """Return all CPVs in the db
+ Args:
+ None
+ Returns:
+ A list of Strings, 1 per CPV
+
+ This function relies on a subclass implementing cp_all, this is why the hasattr is there
+ """
+
+ if not hasattr(self, "cp_all"):
+ raise NotImplementedError
cpv_list = []
for cp in self.cp_all():
cpv_list.extend(self.cp_list(cp))
return cpv_list
+ def cp_all(self):
+ """ Implement this in a child class
+ Args
+ None
+ Returns:
+ A list of strings 1 per CP in the datastore
+ """
+ return NotImplementedError
+
def aux_get(self, mycpv, mylist):
- "stub code for returning auxiliary db information, such as SLOT, DEPEND, etc."
- 'input: "sys-apps/foo-1.0",["SLOT","DEPEND","HOMEPAGE"]'
- 'return: ["0",">=sys-libs/bar-1.0","http://www.foo.com"] or [] if mycpv not found'
+ """Return the metadata keys in mylist for mycpv
+ Args:
+ mycpv - "sys-apps/foo-1.0"
+ mylist - ["SLOT","DEPEND","HOMEPAGE"]
+ Returns:
+ a list of results, in order of keys in mylist, such as:
+ ["0",">=sys-libs/bar-1.0","http://www.foo.com"] or [] if mycpv not found'
+ """
+ raise NotImplementedError
+
+ def aux_update(self, cpv, metadata_updates):
+ """
+ Args:
+ cpv - "sys-apps/foo-1.0"
+ metadata_updates = { key : newvalue }
+ Returns:
+ None
+ """
raise NotImplementedError
def match(self, origdep, use_cache=1):
+ """Given a dependency, try to find packages that match
+ Args:
+ origdep - Depend atom
+ use_cache - Boolean indicating if we should use the cache or not
+ NOTE: Do we ever not want the cache?
+ Returns:
+ a list of packages that match origdep
+ """
mydep = dep_expand(origdep, mydb=self, settings=self.settings)
mykey = dep_getkey(mydep)
mylist = match_from_list(mydep, self.cp_list(mykey, use_cache=use_cache))
return mylist
def invalidentry(self, mypath):
- if re.search("portage_lockfile$", mypath):
+ if mypath.endswith('portage_lockfile'):
if not os.environ.has_key("PORTAGE_MASTER_PID"):
writemsg("Lockfile removed: %s\n" % mypath, 1)
unlockfile((mypath, None, None))
else:
# Nothing we can do about it. We're probably sandboxed.
pass
- elif re.search(".*/-MERGING-(.*)", mypath):
+ elif '/-MERGING-' in mypath:
if os.path.exists(mypath):
writemsg(red("INCOMPLETE MERGE:")+" "+mypath+"\n", noiselevel=-1)
else:
def update_ents(self, updates, onProgress=None):
"""
- Update metadata of all packages for packages moves.
+ Update metadata of all packages for package moves.
@param updates: A list of move commands
@type updates: List
@param onProgress: A progress callback function
onProgress(maxval, i+1)
def move_slot_ent(self, mylist):
+ """This function takes a sequence:
+ Args:
+ mylist: a sequence of (package, originalslot, newslot)
+ Returns:
+ The number of slotmoves this function did
+ """
pkg = mylist[1]
origslot = mylist[2]
newslot = mylist[3]
moves = 0
if not origmatches:
return moves
- from portage.versions import catsplit
for mycpv in origmatches:
slot = self.aux_get(mycpv, ["SLOT"])[0]
if slot != origslot:
from portage.dbapi import dbapi
-
from portage import cpv_getkey
class fakedbapi(dbapi):
- "This is a dbapi to use for the emptytree function. It's empty, but things can be added to it."
+ """A fake dbapi that allows consumers to inject/remove packages to/from it
+ portage.settings is required to maintain the dbAPI.
+ """
def __init__(self, settings=None):
self.cpvdict = {}
self.cpdict = {}
def aux_update(self, cpv, values):
self._clear_cache()
self.cpvdict[cpv].update(values)
+
+class testdbapi(object):
+ """A dbapi instance with completely fake functions to get by hitting disk
+ TODO(antarus):
+ This class really needs to be rewritten to have better stubs; but these work for now.
+ The dbapi classes themselves need unit tests...and that will be a lot of work.
+ """
+
+ def __init__(self):
+ self.cpvs = {}
+ def f(*args, **kwargs):
+ return True
+ fake_api = dir(dbapi)
+ for call in fake_api:
+ if not hasattr(self, call):
+ setattr(self, call, f)
\ No newline at end of file
for root, dirs, files in os.walk(filename):
if 'CVS' in dirs:
dirs.remove('CVS')
- files = filter(files, str.startswith('.'))
- files = filter(files, str.endswith('~'))
+ files = [f for f in files if not f.startswith('.')]
+ files = [f for f in files if not f.endswith('~')]
for f in files:
- yield f
+ yield os.path.join(root, f)
else:
yield filename
"""
raise NotImplementedError("Please override in a subclass")
+class EnvLoader(DataLoader):
+ """ Class to access data in the environment """
+ def __init__(self, validator):
+ DataLoader.__init__(self, validator)
+
+ def load(self):
+ return os.environ
+
+class TestTextLoader(DataLoader):
+ """ You give it some data, it 'loads' it for you, no filesystem access
+ """
+ def __init__(self, validator):
+ DataLoader.__init__(self, validator)
+ self.data = {}
+ self.errors = {}
+
+ def setData(self, text):
+ """Explicitly set the data field
+ Args:
+ text - a dict of data typical of Loaders
+ Returns:
+ None
+ """
+ if isinstance(text, dict):
+ self.data = text
+ else:
+ raise ValueError("setData requires a dict argument")
+
+ def setErrors(self, errors):
+ self.errors = errors
+
+ def load(self):
+ return (self.data, self.errors)
+
class FileLoader(DataLoader):
""" Class to access data in files """
{'key':['foo1','foo2','foo3']}
"""
- def __init__(self, filename, validator):
+ def __init__(self, filename, validator=None, valuevalidator=None):
FileLoader.__init__(self, filename, validator)
+ f = valuevalidator
+ if f is None:
+ # if they pass in no validator, just make a fake one
+ # that always returns true
+ def validate(key):
+ return True
+ f = validate
+ self._valueValidate = f
def lineParser(self, line, line_num, data, errors):
line = line.strip()
if not len(line): # skip empty lines
return
split = line.split()
- if len(split) < 2:
+ if len(split) < 1:
errors.setdefault(self.fname, []).append(
"Malformed data at line: %s, data: %s"
% (line_num + 1, line))
value = split[1:]
if not self._validate(key):
errors.setdefault(self.fname, []).append(
- "Validation failed at line: %s, data %s"
+ "Key validation failed at line: %s, data %s"
% (line_num + 1, key))
return
+ if not self._valueValidate(value):
+ errors.setdefault(self.fname, []).append(
+ "Value validation failed at line: %s, data %s"
+ % (line_num + 1, value))
+ return
if key in data:
data[key].append(value)
else:
'foo':'bar'}
"""
- def __init__(self, filename, validator):
+ def __init__(self, filename, validator, valuevalidator=None):
FileLoader.__init__(self, filename, validator)
+ f = valuevalidator
+ if f is None:
+ # if they pass in no validator, just make a fake one
+ # that always returns true
+ def validate(key):
+ return True
+ f = validate
+ self._valueValidate = f
+
def lineParser(self, line, line_num, data, errors):
line = line.strip()
return
if not self._validate(key):
errors.setdefault(self.fname, []).append(
- "Validation failed at line: %s, data %s"
+ "Key validation failed at line: %s, data %s"
% (line_num + 1, key))
return
+ if not self._valueValidate(value):
+ errors.setdefault(self.fname, []).append(
+ "Value validation failed at line: %s, data %s"
+ % (line_num + 1, value))
+ return
if key in data:
data[key].append(value)
else:
data[key] = value
-
# Distributed under the terms of the GNU General Public License v2
# $Id$
+import os
+import re
from portage.const import INCREMENTALS, PROFILE_PATH, NEWS_LIB_PATH
-from portage.util import ensure_dirs, apply_permissions
+from portage.util import ensure_dirs, apply_permissions, normalize_path
from portage.data import portage_gid
from portage.locks import lockfile, unlockfile, lockdir, unlockdir
from portage.exception import FileNotFound
-import os, re
class NewsManager(object):
"""
"""
- TIMESTAMP_FILE = "news-timestamp"
+ TIMESTAMP_FILE = 'news-timestamp'
def __init__(self, portdb, vardb, NEWS_PATH, UNREAD_PATH, LANGUAGE_ID='en'):
self.NEWS_PATH = NEWS_PATH
self.portdb = portdb
portdir = portdb.porttree_root
- profiles_base = os.path.join(portdir, "profiles") + os.path.sep
- from portage.util import normalize_path
+ profiles_base = os.path.join(portdir, 'profiles') + os.path.sep
profile_path = normalize_path(
os.path.realpath(portdb.mysettings.profile_path))
if profile_path.startswith(profiles_base):
updates.append(item)
del path
- path = os.path.join(self.UNREAD_PATH, "news-"+repoid+".unread")
+ path = os.path.join(self.UNREAD_PATH, 'news-' + repoid + '.unread')
try:
unread_lock = lockfile(path)
if not os.path.exists(path):
open(path, "w")
# Ensure correct perms on the unread file.
apply_permissions( filename=path,
- uid=int(self.config["PORTAGE_INST_UID"]), gid=portage_gid, mode=0664)
+ uid=int(self.config['PORTAGE_INST_UID']), gid=portage_gid, mode=0664)
# Make sure we have the correct permissions when created
- unread_file = open(path, "a")
+ unread_file = open(path, 'a')
for item in updates:
unread_file.write(item.name + "\n")
unlockfile(unread_lock)
# Touch the timestamp file
- f = open(timestamp_file, "w")
+ f = open(timestamp_file, 'w')
f.close()
def getUnreadItems(self, repoid, update=False):
if update:
self.updateItems(repoid)
- unreadfile = os.path.join(self.UNREAD_PATH, "news-"+repoid+".unread")
+ unreadfile = os.path.join(self.UNREAD_PATH, 'news-' + repoid + '.unread')
try:
try:
unread_lock = lockfile(unreadfile)
# Set correct permissions on the news-repoid.unread file
apply_permissions(filename=unreadfile,
- uid=int(self.config["PORTAGE_INST_UID"]), gid=portage_gid, mode=0664)
+ uid=int(self.config['PORTAGE_INST_UID']), gid=portage_gid, mode=0664)
if os.path.exists(unreadfile):
unread = open(unreadfile).readlines()
if not len(self.restrictions):
return True # no restrictions to match means everyone should see it
- kwargs = { 'vardb' : vardb,
- 'config' : config,
- 'profile' : profile }
+ kwargs = \
+ { 'vardb' : vardb,
+ 'config' : config,
+ 'profile' : profile }
for restriction in self.restrictions:
if restriction.checkRestriction(**kwargs):
for line in lines:
#Optimization to ignore regex matchines on lines that
#will never match
- if not line.startswith("D"):
+ if not line.startswith('D'):
continue
restricts = { _installedRE : DisplayInstalledRestriction,
_profileRE : DisplayProfileRestriction,
"""
def checkRestriction(self, **kwargs):
- raise NotImplementedError("Derived class should over-ride this method")
+ raise NotImplementedError('Derived class should over-ride this method')
class DisplayProfileRestriction(DisplayRestriction):
"""
self.keyword = keyword
def checkRestriction(self, **kwargs):
- if kwargs['config']["ARCH"] == self.keyword:
+ if kwargs['config']['ARCH'] == self.keyword:
return True
return False
# $Id$
import os
+from portage import flatten
+from portage.dep import isvalidatom, match_from_list, \
+ best_match_to_list, dep_getkey, use_reduce, paren_reduce
+from portage.exception import InvalidAtom
OPERATIONS = ["merge", "unmerge"]
DEFAULT_SETS = ["world", "system", "everything", "security"] \
+["package_"+x for x in ["mask", "unmask", "use", "keywords"]]
del x
+class PackageSet(object):
+ # Set this to operations that are supported by your subclass. While
+ # technically there is no difference between "merge" and "unmerge" regarding
+ # package sets, the latter doesn't make sense for some sets like "system"
+ # or "security" and therefore isn't supported by them.
+ _operations = ["merge"]
+ _atommap = {}
+ description = "generic package set"
+
+ def __init__(self, name):
+ self.name = name
+ self._atoms = set()
+ self._atommap = {}
+ self._loaded = False
+ self._loading = False
+
+ def __contains__(self, atom):
+ return atom in self.getAtoms()
+
+ def __iter__(self):
+ for x in self.getAtoms():
+ yield x
+
+ def supportsOperation(self, op):
+ if not op in OPERATIONS:
+ raise ValueError(op)
+ return op in self._operations
+
+ def getAtoms(self):
+ if not (self._loaded or self._loading):
+ self._loading = True
+ self.load()
+ self._loaded = True
+ self._loading = False
+ return self._atoms
+
+ def _setAtoms(self, atoms):
+ atoms = map(str.strip, atoms)
+ for a in atoms[:]:
+ if a == "":
+ atoms.remove(a)
+ elif not isvalidatom(a):
+ raise InvalidAtom(a)
+ self._atoms = set(atoms)
+ self._updateAtomMap()
+
+ def load(self):
+ # This method must be overwritten by subclasses
+ # Editable sets should use the value of self._mtime to determine if they
+ # need to reload themselves
+ raise NotImplementedError()
+
+ def containsCPV(self, cpv):
+ for a in self.getAtoms():
+ if match_from_list(a, [cpv]):
+ return True
+ return False
+
+ def getMetadata(self, key):
+ if hasattr(self, key.lower()):
+ return getattr(self, key.lower())
+ else:
+ return ""
+
+ def _updateAtomMap(self, atoms=None):
+ """Update self._atommap for specific atoms or all atoms."""
+ if not atoms:
+ self._atommap.clear()
+ atoms = self._atoms
+ for a in atoms:
+ cp = dep_getkey(a)
+ self._atommap.setdefault(cp, set())
+ self._atommap[cp].add(a)
+
+ # Not sure if this one should really be in PackageSet
+ def findAtomForPackage(self, cpv, metadata):
+ """Return the best match for a given package from the arguments, or
+ None if there are no matches. This matches virtual arguments against
+ the PROVIDE metadata. This can raise an InvalidDependString exception
+ if an error occurs while parsing PROVIDE."""
+ cpv_slot = "%s:%s" % (cpv, metadata["SLOT"])
+ cp = dep_getkey(cpv)
+ self.getAtoms() # make sure the atoms are loaded
+ atoms = self._atommap.get(cp)
+ if atoms:
+ best_match = best_match_to_list(cpv_slot, atoms)
+ if best_match:
+ return best_match
+ if not metadata["PROVIDE"]:
+ return None
+ provides = flatten(use_reduce(paren_reduce(metadata["PROVIDE"]),
+ uselist=metadata["USE"].split()))
+ for provide in provides:
+ provided_cp = dep_getkey(provide)
+ atoms = self._atommap.get(provided_cp)
+ if atoms:
+ atoms = list(atoms)
+ transformed_atoms = [atom.replace(provided_cp, cp) for atom in atoms]
+ best_match = best_match_to_list(cpv_slot, transformed_atoms)
+ if best_match:
+ return atoms[transformed_atoms.index(best_match)]
+ return None
+
+class EditablePackageSet(PackageSet):
+
+ def update(self, atoms):
+ self.getAtoms()
+ self._atoms.update(atoms)
+ self._updateAtomMap(atoms=atoms)
+ self.write()
+
+ def add(self, atom):
+ self.update([atom])
+
+ def replace(self, atoms):
+ self._setAtoms(atoms)
+ self.write()
+
+ def remove(self, atom):
+ self.getAtoms()
+ self._atoms.discard(atom)
+ self._updateAtomMap()
+ self.write()
+
+ def removePackageAtoms(self, cp):
+ for a in list(self.getAtoms()):
+ if dep_getkey(a) == cp:
+ self.remove(a)
+ self.write()
+
+ def write(self):
+ # This method must be overwritten in subclasses that should be editable
+ raise NotImplementedError()
+
+
+class InternalPackageSet(EditablePackageSet):
+ def __init__(self, initial_atoms=None):
+ super(InternalPackageSet, self).__init__("")
+ if initial_atoms != None:
+ self.update(initial_atoms)
+
+ def clear(self):
+ self._atoms.clear()
+ self._updateAtomMap()
+
+ def load(self):
+ pass
+
+ def write(self):
+ pass
+
+
+
def make_default_sets(configroot, root, profile_paths, settings=None,
vdbapi=None, portdbapi=None):
from portage.sets.files import StaticFileSet, ConfigFileSet
# $Id$
from portage.versions import catsplit
-from portage import flatten
-from portage.dep import isvalidatom, match_from_list, \
- best_match_to_list, dep_getkey, use_reduce, paren_reduce
-from portage.exception import InvalidAtom
-from portage.sets import OPERATIONS
-
-class PackageSet(object):
- # Set this to operations that are supported by your subclass. While
- # technically there is no difference between "merge" and "unmerge" regarding
- # package sets, the latter doesn't make sense for some sets like "system"
- # or "security" and therefore isn't supported by them.
- _operations = ["merge"]
- _atommap = {}
- description = "generic package set"
-
- def __init__(self, name):
- self.name = name
- self._atoms = set()
- self._atommap = {}
- self._loaded = False
- self._loading = False
-
- def __contains__(self, atom):
- return atom in self.getAtoms()
-
- def __iter__(self):
- for x in self.getAtoms():
- yield x
-
- def supportsOperation(self, op):
- if not op in OPERATIONS:
- raise ValueError(op)
- return op in self._operations
-
- def getAtoms(self):
- if not (self._loaded or self._loading):
- self._loading = True
- self.load()
- self._loaded = True
- self._loading = False
- return self._atoms
-
- def _setAtoms(self, atoms):
- atoms = map(str.strip, atoms)
- for a in atoms[:]:
- if a == "":
- atoms.remove(a)
- elif not isvalidatom(a):
- raise InvalidAtom(a)
- self._atoms = set(atoms)
- self._updateAtomMap()
-
- def load(self):
- # This method must be overwritten by subclasses
- # Editable sets should use the value of self._mtime to determine if they
- # need to reload themselves
- raise NotImplementedError()
-
- def containsCPV(self, cpv):
- for a in self.getAtoms():
- if match_from_list(a, [cpv]):
- return True
- return False
-
- def getMetadata(self, key):
- if hasattr(self, key.lower()):
- return getattr(self, key.lower())
- else:
- return ""
-
- def _updateAtomMap(self, atoms=None):
- """Update self._atommap for specific atoms or all atoms."""
- if not atoms:
- self._atommap.clear()
- atoms = self._atoms
- for a in atoms:
- cp = dep_getkey(a)
- self._atommap.setdefault(cp, set())
- self._atommap[cp].add(a)
-
- # Not sure if this one should really be in PackageSet
- def findAtomForPackage(self, cpv, metadata):
- """Return the best match for a given package from the arguments, or
- None if there are no matches. This matches virtual arguments against
- the PROVIDE metadata. This can raise an InvalidDependString exception
- if an error occurs while parsing PROVIDE."""
- cpv_slot = "%s:%s" % (cpv, metadata["SLOT"])
- cp = dep_getkey(cpv)
- self.getAtoms() # make sure the atoms are loaded
- atoms = self._atommap.get(cp)
- if atoms:
- best_match = best_match_to_list(cpv_slot, atoms)
- if best_match:
- return best_match
- if not metadata["PROVIDE"]:
- return None
- provides = flatten(use_reduce(paren_reduce(metadata["PROVIDE"]),
- uselist=metadata["USE"].split()))
- for provide in provides:
- provided_cp = dep_getkey(provide)
- atoms = self._atommap.get(provided_cp)
- if atoms:
- atoms = list(atoms)
- transformed_atoms = [atom.replace(provided_cp, cp) for atom in atoms]
- best_match = best_match_to_list(cpv_slot, transformed_atoms)
- if best_match:
- return atoms[transformed_atoms.index(best_match)]
- return None
-
-class EditablePackageSet(PackageSet):
-
- def update(self, atoms):
- self.getAtoms()
- self._atoms.update(atoms)
- self._updateAtomMap(atoms=atoms)
- self.write()
-
- def add(self, atom):
- self.update([atom])
-
- def replace(self, atoms):
- self._setAtoms(atoms)
- self.write()
-
- def remove(self, atom):
- self.getAtoms()
- self._atoms.discard(atom)
- self._updateAtomMap()
- self.write()
-
- def removePackageAtoms(self, cp):
- for a in list(self.getAtoms()):
- if dep_getkey(a) == cp:
- self.remove(a)
- self.write()
-
- def write(self):
- # This method must be overwritten in subclasses that should be editable
- raise NotImplementedError()
-
-
-class InternalPackageSet(EditablePackageSet):
- def __init__(self, initial_atoms=None):
- super(InternalPackageSet, self).__init__("")
- if initial_atoms != None:
- self.update(initial_atoms)
-
- def clear(self):
- self._atoms.clear()
- self._updateAtomMap()
-
- def load(self):
- pass
-
- def write(self):
- pass
+from portage.sets import PackageSet
class EverythingSet(PackageSet):
_operations = ["merge", "unmerge"]
from portage.const import PRIVATE_PATH
from portage.locks import lockfile, unlockfile
from portage import portage_gid
-from portage.sets.dbapi import PackageSet, EditablePackageSet
-from portage.env.config import ConfigLoaderKlass, GenericFile
+from portage.sets import PackageSet, EditablePackageSet
from portage.env.loaders import ItemFileLoader, KeyListFileLoader
-from portage.env.validators import PackagesFileValidator, ValidAtomValidator
+from portage.env.validators import ValidAtomValidator
class StaticFileSet(EditablePackageSet):
_operations = ["merge", "unmerge"]
- def __init__(self, name, filename, loader=None):
+ def __init__(self, name, filename):
super(StaticFileSet, self).__init__(name)
self._filename = filename
self._mtime = None
self.description = "Package set loaded from file %s" % self._filename
- if loader is None:
- self.loader = ConfigLoaderKlass(ItemFileLoader(filename=self._filename,
- validator=PackagesFileValidator))
+ self.loader = ItemFileLoader(self._filename, ValidAtomValidator)
+
metadata = grabfile(self._filename + ".metadata")
key = None
value = []
except (OSError, IOError):
mtime = None
if (not self._loaded or self._mtime != mtime):
- self.loader.load()
- self._setAtoms(self.loader.keys())
+ try:
+ data, errors = self.loader.load()
+ except EnvironmentError, e:
+ if e.errno != errno.ENOENT:
+ raise
+ del e
+ self._setAtoms(data.keys())
self._mtime = mtime
class ConfigFileSet(PackageSet):
- def __init__(self, name, filename, loader=None, validator=None):
+ def __init__(self, name, filename):
super(ConfigFileSet, self).__init__(name)
self._filename = filename
self.description = "Package set generated from %s" % self._filename
- if loader is None:
- self.loader = GenericFile(filename=self._filename)
- else:
- self.loader = loader(filename, validator)
+ self.loader = KeyListFileLoader(self._filename, ValidAtomValidator)
def load(self):
- self.loader.load()
- self._setAtoms(self.loader.keys())
+ data, errors = self.loader.load()
+ self._setAtoms(data.keys())
class WorldSet(StaticFileSet):
description = "Set of packages that were directly installed by the user"
import os
from portage.util import grabfile_package, stack_lists
-from portage.sets.dbapi import PackageSet
+from portage.sets import PackageSet
class PackagesSystemSet(PackageSet):
_operations = ["merge"]
import portage.glsa as glsa
from portage.util import grabfile, write_atomic
from portage.const import CACHE_PATH
-from portage.sets.dbapi import PackageSet
+from portage.sets import PackageSet
class SecuritySet(PackageSet):
_operations = ["merge"]
import subprocess
import os
-from portage.sets.dbapi import PackageSet
+from portage.sets import PackageSet
class CommandOutputSet(PackageSet):
"""This class creates a PackageSet from the output of a shell command.
import unittest
def main():
- testDirs = ["bin", "dep", "ebuild",
- "env/config", "util", "versions", "xpak", "sets/shell"]
+
+ TEST_FILE = '__test__'
suite = unittest.TestSuite()
basedir = os.path.dirname(__file__)
+ testDirs = []
+
+ # the os.walk help mentions relative paths as being quirky
+ # I was tired of adding dirs to the list, so now we add __test__
+ # to each dir we want tested.
+ for root, dirs, files in os.walk(os.getcwd()):
+ if ".svn" in dirs:
+ dirs.remove('.svn')
+ if TEST_FILE in files:
+ testDirs.append(root)
+
for mydir in testDirs:
suite.addTests(getTests(os.path.join(basedir, mydir), basedir) )
return TextTestRunner(verbosity=2).run(suite)
# Distributed under the terms of the GNU General Public License v2
# $Id: test_varExpand.py 5596 2007-01-12 08:08:53Z antarus $
-from portage.tests import TestCase, TestLoader
+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
class NewsItemTestCase(TestCase):
-
- self.fakeItem = """
+ """These tests suck: they use your running config instead of making their own"""
+ fakeItem = """
Title: YourSQL Upgrades from 4.0 to 4.1
Author: Ciaran McCreesh <ciaranm@gentoo.org>
Content-Type: text/plain
The revdep-rebuild tool is provided by app-portage/gentoolkit.
"""
+ def setUp(self):
+ self.profile = "/usr/portage/profiles/default-linux/x86/2007.0/"
+ self.keywords = "x86"
+ # Use fake/test dbapi to avoid slow tests
+ self.vardb = testdbapi()
+ # self.vardb.inject_cpv('sys-apps/portage-2.0', { 'SLOT' : 0 })
+ # Consumers only use ARCH, so avoid portage.settings by using a dict
+ self.settings = { 'ARCH' : 'x86' }
+
+ def testDisplayIfProfile(self):
+ tmpItem = self.fakeItem[:].replace("#Display-If-Profile:", "Display-If-Profile: %s" %
+ self.profile)
- from portage import settings
- import time
-
- def testDisplayIfProfile():
- from portage.const import PROFILE_PATH
- tmpItem = self.fakeItem.replace("#Display-If-Profile:", "Display-If-Profile: %s" %
- os.readlink( PROFILE_PATH ) )
-
- item = _processItem(tmpItem)
- self.assertTrue( item.isRelevant( os.readlink( PROFILE_PATH ) ),
- msg="Expected %s to be relevant, but it was not!" % tmpItem )
-
- def testDisplayIfInstalled():
- tmpItem = self.fakeItem.replace("#Display-If-Installed:", "Display-If-Profile: %s" %
- "sys-apps/portage" )
-
- item = _processItem(tmpItem)
- self.assertTrue( item.isRelevant( portage.settings ),
- msg="Expected %s to be relevant, but it was not!" % tmpItem )
-
+ try:
+ item = self._processItem(tmpItem)
+ self.assertTrue(item.isRelevant(self.vardb, self.settings, self.profile),
+ msg="Expected %s to be relevant, but it was not!" % tmpItem)
+ finally:
+ os.unlink(item.path)
- def testDisplayIfKeyword():
- from portage import settings
- tmpItem = self.fakeItem.replace("#Display-If-Keyword:", "Display-If-Keyword: %s" %
- settings["ACCEPT_KEYWORDS"].split()[0] )
+ def testDisplayIfInstalled(self):
+ tmpItem = self.fakeItem[:].replace("#Display-If-Installed:", "Display-If-Installed: %s" %
+ "sys-apps/portage")
- item = _processItem(tmpItem)
- self.assertTrue( item.isRelevant( os.readlink( PROFILE_PATH ) ),
- msg="Expected %s to be relevant, but it was not!" % tmpItem )
-
+ try:
+ item = self._processItem(tmpItem)
+ self.assertTrue(item.isRelevant(self.vardb, self.settings, self.profile),
+ msg="Expected %s to be relevant, but it was not!" % tmpItem)
+ finally:
+ os.unlink(item.path)
- def _processItem( self, item ):
+ def testDisplayIfKeyword(self):
+ tmpItem = self.fakeItem[:].replace("#Display-If-Keyword:", "Display-If-Keyword: %s" %
+ self.keywords)
- path = os.path.join(settings["PORTAGE_TMPDIR"], str(time.time()))
- f = open(path)
+ try:
+ item = self._processItem(tmpItem)
+ self.assertTrue(item.isRelevant(self.vardb, self.settings, self.profile),
+ msg="Expected %s to be relevant, but it was not!" % tmpItem)
+ finally:
+ os.unlink(item.path)
+
+ def _processItem(self, item):
+ filename = None
+ fd, filename = mkstemp()
+ f = os.fdopen(fd, 'wb')
f.write(item)
f.close
try:
- return NewsItem( path, 0 )
+ return NewsItem(filename, 0)
except TypeError:
- self.fail("Error while processing news item %s" % path )
+ self.fail("Error while processing news item %s" % filename)
--- /dev/null
+# testConfigFileSet.py -- Portage Unit Testing Functionality
+# Copyright 2007 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+# $Id: testShell.py 7363 2007-07-22 23:21:14Z zmedico $
+
+import tempfile, os
+
+from portage.tests import TestCase, test_cps
+from portage.sets.files import ConfigFileSet
+
+class ConfigFileSetTestCase(TestCase):
+ """Simple Test Case for ConfigFileSet"""
+
+ def setUp(self):
+ fd, self.testfile = tempfile.mkstemp(suffix=".testdata", prefix=self.__class__.__name__, text=True)
+ for i in range(0, len(test_cps)):
+ atom = test_cps[i]
+ if i % 2 == 0:
+ os.write(fd, atom+" abc def"+"\n")
+ else:
+ os.write(fd, atom+"\n")
+ os.close(fd)
+
+ def tearDown(self):
+# os.unlink(self.testfile)
+ pass
+
+ def testConfigStaticFileSet(self):
+ s = ConfigFileSet('test', self.testfile)
+ s.load()
+ self.assertEqual(set(test_cps), s.getAtoms())
+
--- /dev/null
+# testStaticFileSet.py -- Portage Unit Testing Functionality
+# Copyright 2007 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+# $Id: testShell.py 7363 2007-07-22 23:21:14Z zmedico $
+
+import tempfile, 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"""
+
+ def setUp(self):
+ fd, self.testfile = tempfile.mkstemp(suffix=".testdata", prefix=self.__class__.__name__, text=True)
+ os.write(fd, "\n".join(test_cps))
+ os.close(fd)
+
+ def tearDown(self):
+ os.unlink(self.testfile)
+
+ def testSampleStaticFileSet(self):
+ s = StaticFileSet('test', self.testfile)
+ s.load()
+ self.assertEqual(set(test_cps), s.getAtoms())
+
projects / portage.git / commitdiff