from portage import os
from portage import dispatch_conf
+from portage import _unicode_decode
from portage.process import find_binary
from portage.const import EPREFIX
import curses
try:
curses.setupterm()
- sys.stdout.write(curses.tigetstr("clear"))
+ sys.stdout.write(_unicode_decode(curses.tigetstr("clear")))
sys.stdout.flush()
return
except curses.error:
for x in lookup_owners:
if not search_for_multiple and os.path.isdir(x):
search_for_multiple = True
- relative_paths.append(x[len(root):])
+ relative_paths.append(x[len(root)-1:])
owners = set()
for pkg, relative_path in \
for x in lookup_owners:
if not search_for_multiple and os.path.isdir(x):
search_for_multiple = True
- relative_paths.append(x[len(myroot):])
+ relative_paths.append(x[len(myroot)-1:])
owners = set()
for pkg, relative_path in \
Raises an EnvironmentError from openpty() if it fails.
"""
- import array, pty, termios
+ import array, fcntl, pty, select, termios
test_string = 2 * "blah blah blah\n"
test_string = _unicode_decode(test_string,
encoding='utf_8', errors='strict')
# may raise EnvironmentError
master_fd, slave_fd = pty.openpty()
- master_file = os.fdopen(master_fd, 'rb')
- slave_file = os.fdopen(slave_fd, 'wb')
+ # Non-blocking mode is required for Darwin kernel.
+ fcntl.fcntl(master_fd, fcntl.F_SETFL,
+ fcntl.fcntl(master_fd, fcntl.F_GETFL) | os.O_NONBLOCK)
# Disable post-processing of output since otherwise weird
# things like \n -> \r\n transformations may occur.
termios.tcsetattr(slave_fd, termios.TCSANOW, mode)
# Simulate a subprocess writing some data to the
- # slave end of the pipe, and then exiting.
- slave_file.write(_unicode_encode(test_string,
- encoding='utf_8', errors='strict'))
- slave_file.close()
+ # slave end of the pipe, and then exiting. Do a
+ # real fork here since otherwise os.close(slave_fd)
+ # would block on some platforms such as Darwin.
+ pids = process.spawn_bash(_unicode_encode("echo -n '%s'" % test_string,
+ encoding='utf_8', errors='strict'), env=os.environ,
+ fd_pipes={0:sys.stdin.fileno(), 1:slave_fd, 2:slave_fd},
+ returnpid=True)
+ if isinstance(pids, int):
+ os.close(master_fd)
+ os.close(slave_fd)
+ raise EnvironmentError('spawn failed')
+ os.close(slave_fd)
+ master_file = os.fdopen(master_fd, 'rb')
eof = False
data = []
+ iwtd = [master_file]
+ owtd = []
+ ewtd = []
while not eof:
+ events = select.select(iwtd, owtd, ewtd)
+ if not events[0]:
+ eof = True
+ break
+
buf = array.array('B')
try:
buf.fromfile(master_file, 1024)
data.append(_unicode_decode(buf.tostring(),
encoding='utf_8', errors='strict'))
+ os.waitpid(pids[0], 0)
master_file.close()
return test_string == ''.join(data)
'portage.locks:unlockfile',
'portage.output:colorize',
'portage.util:cmp_sort_key,writemsg',
- 'portage.versions:catpkgsplit,vercmp',
+ 'portage.versions:catsplit,catpkgsplit,vercmp',
)
from portage import os
"""
if self._categories is not None:
return self._categories
- categories = set()
- cat_pattern = re.compile(r'(.*)/.*')
- for cp in self.cp_all():
- categories.add(cat_pattern.match(cp).group(1))
- self._categories = tuple(sorted(categories))
+ self._categories = tuple(sorted(set(catsplit(x)[0] \
+ for x in self.cp_all())))
return self._categories
def close_caches(self):
# Distributed under the terms of the GNU General Public License v2
# $Id$
+import tempfile
+
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 = os.path.join(os.path.dirname(os.path.dirname(
+basepath = os.path.join(os.path.dirname(os.path.dirname(
os.path.abspath(__file__))),
- "..", "..", "..", "bin", "ebuild-helpers")
-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")
-os.environ["PF"] = "portage-tests-0.09-r1"
-os.environ["PATH"] = bindir + ":" + os.environ["PATH"]
-os.environ["PORTAGE_BIN_PATH"] = PORTAGE_BIN_PATH
+ "..", "..", "..")
+bindir = os.path.join(basepath, "bin")
+pymdir = os.path.join(basepath, "pym")
+basedir = None
+env = None
def binTestsCleanup():
+ global basedir
+ if basedir is None:
+ return
if os.access(basedir, os.W_OK):
shutil.rmtree(basedir)
+ basedir = None
+
def binTestsInit():
binTestsCleanup()
- os.mkdir(basedir)
- os.mkdir(os.environ["D"])
- os.mkdir(os.environ["T"])
- os.mkdir(os.environ["S"])
- os.chdir(os.environ["S"])
+ global basedir, env
+ basedir = tempfile.mkdtemp()
+ env = os.environ.copy()
+ env["D"] = os.path.join(basedir, "image")
+ env["T"] = os.path.join(basedir, "temp")
+ env["S"] = os.path.join(basedir, "workdir")
+ env["PF"] = "portage-tests-0.09-r1"
+ env["PATH"] = bindir + ":" + env["PATH"]
+ env["PORTAGE_BIN_PATH"] = bindir
+ env["PORTAGE_PYM_PATH"] = pymdir
+ os.mkdir(env["D"])
+ os.mkdir(env["T"])
+ os.mkdir(env["S"])
+ os.chdir(env["S"])
class BinTestCase(TestCase):
def __init__(self, methodName):
def _exists_in_D(path):
# Note: do not use os.path.join() here, we assume D to end in /
- return os.access(os.environ["D"] + path, os.W_OK)
+ return os.access(env["D"] + path, os.W_OK)
def exists_in_D(path):
if not _exists_in_D(path):
raise TestCase.failureException
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
+ global env
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)
+ spawn([func] + args.split(), env=env, fd_pipes=fd_pipes)
f.close()
def create_portage_wrapper(bin):
return portage_func(*newargs)
return derived_func
-for bin in os.listdir(bindir):
+for bin in os.listdir(os.path.join(bindir, "ebuild-helpers")):
if bin.startswith("do") or \
bin.startswith("new") or \
bin.startswith("prep") or \
bin in ["ecompress","ecompressdir","fowners","fperms"]:
- globals()[bin] = create_portage_wrapper(bin)
+ globals()[bin] = create_portage_wrapper(
+ os.path.join(bindir, "ebuild-helpers", bin))
# Distributed under the terms of the GNU General Public License v2
# $Id$
-from portage.tests.bin.setup_env import *
+from portage.tests.bin.setup_env import BinTestCase, dobin, xexists_in_D
class DoBin(BinTestCase):
def testDoBin(self):
# Distributed under the terms of the GNU General Public License v2
# $Id$
-from portage.tests.bin.setup_env import *
+from portage.tests.bin.setup_env import BinTestCase, dodir, exists_in_D
class DoDir(BinTestCase):
def testDoDir(self):
projects / portage.git / commitdiff