def test_mode(self):
     path = pjoin(self.dir, 'mode', 'mode')
     assert osutils.ensure_dirs(path, mode=0o700)
     self.check_dir(path, os.geteuid(), os.getegid(), 0o700)
     # unrestrict it
     osutils.ensure_dirs(path)
     self.check_dir(path, os.geteuid(), os.getegid(), 0o777)

 def mk_tree(self, path, *args, **kwds):
     eclasses = kwds.pop('eclass_cache', None)
     if eclasses is None:
         epath = pjoin(path, 'eclass')
         ensure_dirs(epath)
         eclasses = eclass_cache.cache(epath)
     return repository._UnconfiguredTree(path, eclasses, *args, **kwds)

 def test_licenses(self):
     licenses = ('GPL-2', 'GPL-3+', 'BSD')
     ensure_dirs(pjoin(self.dir, 'licenses'))
     for license in licenses:
         touch(pjoin(self.dir, 'licenses', license))
     repo = self.mk_tree(self.dir)
     self.assertEqual(sorted(repo.licenses), sorted(licenses))

    def nofetch(self):
        """Execute the nofetch phase.

        We need the same prerequisites as setup, so reuse that.
        """
        ensure_dirs(self.env["T"], mode=0o770, gid=portage_gid, minimal=True)
        return setup_mixin.setup(self, "nofetch")

 def test_create_unwritable_subdir(self):
     path = pjoin(self.dir, 'restricted', 'restricted')
     # create the subdirs without 020 first
     assert osutils.ensure_dirs(os.path.dirname(path))
     assert osutils.ensure_dirs(path, mode=0o020)
     self.check_dir(path, os.geteuid(), os.getegid(), 0o020)
     # unrestrict it
     osutils.ensure_dirs(path)
     self.check_dir(path, os.geteuid(), os.getegid(), 0o777)

    def setUp(self):
        TempDirMixin.setUp(self)
        self.pdir = pjoin(self.dir, 'profiles')
        ensure_dirs(self.pdir)

        # silence missing masters warnings
        ensure_dirs(pjoin(self.dir, 'metadata'))
        with open(pjoin(self.dir, 'metadata', 'layout.conf'), 'w') as f:
            f.write('masters =\n')

 def trigger(self, engine, cset):
     pkg = engine.new
     filename = "%s_%s%s.deb" % (pkg.package, pkg.version, self.postfix)
     tmp_path = pjoin(engine.tempdir, filename)
     final_path = pjoin(self.basepath, filename)
     ensure_dirs(tmp_path)
     deb.write(tmp_path, final_path, pkg,
         cset=cset,
         platform=self.platform, maintainer=self.maintainer)

 def test_repo_id(self):
     dir1 = pjoin(self.dir, '1')
     os.mkdir(dir1, 0755)
     repo = self.mk_tree(dir1)
     self.assertEqual(repo.repo_id, '<unlabeled repository %s>' % (dir1,))
     dir2 = pjoin(self.dir, '2')
     osutils.ensure_dirs(pjoin(dir2, 'profiles'))
     open(pjoin(dir2, 'profiles', 'repo_name'), 'w').write('testrepo\n')
     repo = self.mk_tree(dir2)
     self.assertEqual('testrepo', repo.repo_id)

 def test_repo_id(self):
     dir1 = pjoin(self.dir, '1')
     os.mkdir(dir1, 0o755)
     repo = self.mk_tree(dir1)
     self.assertEqual(repo.repo_id, f'<unlabeled repo: {dir1!r}>')
     dir2 = pjoin(self.dir, '2')
     ensure_dirs(pjoin(dir2, 'profiles'))
     with open(pjoin(dir2, 'profiles', 'repo_name'), 'w') as f:
         f.write('testrepo\n')
     repo = self.mk_tree(dir2)
     self.assertEqual('testrepo', repo.repo_id)

    def test_mkdir_failing(self):
        # fail if os.mkdir fails
        with mock.patch('snakeoil.osutils.os.mkdir') as mkdir:
            mkdir.side_effect = OSError(30, 'Read-only file system')
            path = pjoin(self.dir, 'dir')
            assert not osutils.ensure_dirs(path, mode=0o700)

            # force temp perms
            assert not osutils.ensure_dirs(path, mode=0o400)
            mkdir.side_effect = OSError(17, 'File exists')
            assert not osutils.ensure_dirs(path, mode=0o700)

    def test_categories_packages(self):
        ensure_dirs(pjoin(self.dir, 'cat', 'pkg'))
        ensure_dirs(pjoin(self.dir, 'empty', 'empty'))
        ensure_dirs(pjoin(self.dir, 'scripts', 'pkg'))
        ensure_dirs(pjoin(self.dir, 'notcat', 'CVS'))
        touch(pjoin(self.dir, 'cat', 'pkg', 'pkg-3.ebuild'))
        repo = self.mk_tree(self.dir)
        self.assertEqual(
            {'cat': (), 'notcat': (), 'empty': ()}, dict(repo.categories))
        self.assertEqual(
            {'cat': ('pkg',), 'empty': ('empty',), 'notcat': ()},
            dict(repo.packages))
        self.assertEqual(
            {('cat', 'pkg'): ('3',), ('empty', 'empty'): ()},
            dict(repo.versions))

        for x in ("1-scm", "scm", "1-try", "1_beta-scm", "1_beta-try"):
            for rev in ("", "-r1"):
                fp = pjoin(self.dir, 'cat', 'pkg', 'pkg-%s%s.ebuild' %
                    (x, rev))
                open(fp, 'w').close()
                repo = self.mk_tree(self.dir)
                self.assertRaises(ebuild_errors.InvalidCPV,
                    repo.match, atom('cat/pkg'))
                repo = self.mk_tree(self.dir, ignore_paludis_versioning=True)
                self.assertEqual(sorted(x.cpvstr for x in
                    repo.itermatch(atom('cat/pkg'))), ['cat/pkg-3'])
                os.unlink(fp)

 def _cmd_implementation_sanity_check(self, domain):
     pkg = self.pkg
     eapi = pkg.eapi_obj
     if eapi.options.has_required_use:
         use = pkg.use
         for node in pkg.required_use:
             if not node.match(use):
                 print(textwrap.dedent(
                     """
                     REQUIRED_USE requirement wasn't met
                     Failed to match: {}
                     from: {}
                     for USE: {}
                     pkg: {}
                     """.format(node, pkg.required_use, " ".join(use), pkg.cpvstr)
                 ))
                 return False
     if 'pretend' not in pkg.mandatory_phases:
         return True
     commands = None
     if not pkg.built:
         commands = {"request_inherit": partial(inherit_handler, self._eclass_cache)}
     env = expected_ebuild_env(pkg)
     tmpdir = normpath(domain._get_tempspace())
     builddir = pjoin(tmpdir, env["CATEGORY"], env["PF"])
     pkg_tmpdir = normpath(pjoin(builddir, "temp"))
     ensure_dirs(pkg_tmpdir, mode=0770, gid=portage_gid, minimal=True)
     env["ROOT"] = domain.root
     env["T"] = pkg_tmpdir
     try:
         logger.debug("running ebuild pkg_pretend sanity check for %s", pkg.cpvstr)
         start = time.time()
         ret = run_generic_phase(pkg, "pretend", env, userpriv=True, sandbox=True,
                                 fakeroot=False, extra_handlers=commands)
         logger.debug("pkg_pretend sanity check for %s took %2.2f seconds",
             pkg.cpvstr, time.time() - start)
         return ret
     except format.GenericBuildError as e:
         logger.error("pkg_pretend sanity check for %s failed with exception %r"
             % (pkg.cpvstr, e))
         return False
     finally:
         shutil.rmtree(builddir)
         # try to wipe the cat dir; if not empty, ignore it
         try:
             os.rmdir(os.path.dirname(builddir))
         except EnvironmentError as e:
             if e.errno != errno.ENOTEMPTY:
                 raise

 def test_gid(self):
     # abuse the portage group as secondary group
     try:
         portage_gid = grp.getgrnam('portage').gr_gid
     except KeyError:
         pytest.skip('the portage group does not exist')
     if portage_gid not in os.getgroups():
         pytest.skip('you are not in the portage group')
     path = pjoin(self.dir, 'group', 'group')
     assert osutils.ensure_dirs(path, gid=portage_gid)
     self.check_dir(path, os.geteuid(), portage_gid, 0o777)
     assert osutils.ensure_dirs(path)
     self.check_dir(path, os.geteuid(), portage_gid, 0o777)
     assert osutils.ensure_dirs(path, gid=os.getegid())
     self.check_dir(path, os.geteuid(), os.getegid(), 0o777)

    def setup_distfiles(self):
        if not self.verified_files and self.allow_fetching:
            ops = self.domain.pkg_operations(self.pkg,
                observer=self.observer)
            if not ops.fetch():
                raise format.BuildError("failed fetching required distfiles")
            self.verified_files = ops._fetch_op.verified_files

        if self.verified_files:
            try:
                if os.path.exists(self.env["DISTDIR"]):
                    if (os.path.isdir(self.env["DISTDIR"])
                        and not os.path.islink(self.env["DISTDIR"])):
                        shutil.rmtree(self.env["DISTDIR"])
                    else:
                        os.unlink(self.env["DISTDIR"])

            except EnvironmentError as oe:
                raise_from(format.FailedDirectory(
                    self.env["DISTDIR"],
                    "failed removing existing file/dir/link at: exception %s"
                    % oe))

            if not ensure_dirs(self.env["DISTDIR"], mode=0770,
                               gid=portage_gid):
                raise format.FailedDirectory(

    def add_data(self):
        if self.observer is None:
            end = start = lambda x:None
        else:
            start = self.observer.phase_start
            end = self.observer.phase_end
        pkg = self.new_pkg
        final_path = discern_loc(self.repo.base, pkg, self.repo.extension)
        tmp_path = pjoin(os.path.dirname(final_path),
            ".tmp.%i.%s" % (os.getpid(), os.path.basename(final_path)))

        self.tmp_path, self.final_path = tmp_path, final_path

        if not ensure_dirs(os.path.dirname(tmp_path), mode=0755):
            raise repo_interfaces.Failure("failed creating directory %r" %
                os.path.dirname(tmp_path))
        try:
            start("generating tarball: %s" % tmp_path)
            tar.write_set(pkg.contents, tmp_path, compressor='bzip2',
                parallelize=True)
            end("tarball created", True)
            start("writing Xpak")
            # ok... got a tarball.  now add xpak.
            xpak.Xpak.write_xpak(tmp_path, generate_attr_dict(pkg))
            end("wrote Xpak", True)
            # ok... we tagged the xpak on.
            os.chmod(tmp_path, 0644)
        except Exception as e:
            try:
                unlink_if_exists(tmp_path)
            except EnvironmentError as e:
                logger.warning("failed removing %r: %r" % (tmp_path, e))
            raise
        return True

def _write_mtime_cache(mtimes, data, location):
    old_umask = os.umask(0113)
    try:
        f = None
        logger.debug("attempting to update mtime cache at %r", (location,))
        try:
            if not ensure_dirs(os.path.dirname(location),
                gid=portage_gid, mode=0775):
                # bugger, can't update..
                return
            f = AtomicWriteFile(location, gid=portage_gid, perms=0664)
            # invert the data...
            rev_data = {}
            for pkg, ver_dict in data.iteritems():
                for fullver, virtuals in ver_dict.iteritems():
                    for virtual in virtuals:
                        rev_data.setdefault(virtual.category, []).extend(
                            (pkg, fullver, str(virtual)))
            for cat, mtime in mtimes.iteritems():
                if cat in rev_data:
                    f.write("%s\t%i\t%s\n" % (cat, mtime,
                         '\t'.join(rev_data[cat])))
                else:
                    f.write("%s\t%i\n" % (cat, mtime))
            f.close()
            os.chown(location, -1, portage_gid)
        except IOError as e:
            if f is not None:
                f.discard()
            if e.errno != errno.EACCES:
                raise
            logger.warning("unable to update vdb virtuals cache due to "
                "lacking permissions")
    finally:
        os.umask(old_umask)

    def _setup_distfiles(self):
        if not self.verified_files and self.allow_fetching:
            ops = self.domain.pkg_operations(self.pkg, observer=self.observer)
            if not ops.fetch():
                raise format.GenericBuildError("failed fetching required distfiles")
            self.verified_files = ops._fetch_op.verified_files

        if self.verified_files:
            try:
                if os.path.exists(self.env["DISTDIR"]):
                    if (os.path.isdir(self.env["DISTDIR"]) and
                            not os.path.islink(self.env["DISTDIR"])):
                        shutil.rmtree(self.env["DISTDIR"])
                    else:
                        os.unlink(self.env["DISTDIR"])

            except EnvironmentError as e:
                raise format.FailedDirectory(
                    self.env["DISTDIR"],
                    f"failed removing existing file/dir/link: {e}") from e

            if not ensure_dirs(self.env["DISTDIR"], mode=0o770, gid=portage_gid):
                raise format.FailedDirectory(
                    self.env["DISTDIR"],
                    "failed creating distdir symlink directory")

            try:
                for src, dest in [
                        (k, pjoin(self.env["DISTDIR"], v.filename))
                        for (k, v) in self.verified_files.items()]:
                    os.symlink(src, dest)

            except EnvironmentError as e:
                raise format.GenericBuildError(
                    f"Failed symlinking in distfiles for src {src} -> {dest}: {e}") from e

    def _post_download(self, path):
        super()._post_download(path)

        # create tempdir for staging decompression
        if not ensure_dirs(self.tempdir, mode=0o755, uid=self.uid, gid=self.gid):
            raise base.SyncError(
                f'failed creating repo update dir: {self.tempdir!r}')

        exts = {'gz': 'gzip', 'bz2': 'bzip2', 'xz': 'xz'}
        compression = exts[self.uri.rsplit('.', 1)[1]]
        # use tar instead of tarfile so we can easily strip leading path components
        # TODO: programmatically determine how man components to strip?
        cmd = [
            'tar', '--extract', f'--{compression}', '-f', path,
            '--strip-components=1', '--no-same-owner', '-C', self.tempdir
        ]
        with open(os.devnull) as f:
            ret = self._spawn(cmd, pipes={1: f.fileno(), 2: f.fileno()})
        if ret:
            raise base.SyncError('failed to unpack tarball')

        # TODO: verify gpg data if it exists

        # move old repo out of the way and then move new, unpacked repo into place
        try:
            os.rename(self.basedir, self.tempdir_old)
            os.rename(self.tempdir, self.basedir)
        except OSError as e:
            raise base.SyncError(f'failed to update repo: {e.strerror}') from e

        # register old repo removal after it has been successfully replaced
        atexit.register(partial(shutil.rmtree, self.tempdir_old, ignore_errors=True))

    def mk_profiles(self, profiles, base='profiles', arches=None):
        os.mkdir(pjoin(self.dir, 'metadata'))
        # write masters= to suppress logging complaints.
        write_file(pjoin(self.dir, 'metadata', 'layout.conf'), 'w', 'masters=')

        loc = pjoin(self.dir, base)
        os.mkdir(loc)
        for profile in profiles:
            self.assertTrue(ensure_dirs(pjoin(loc, profile)),
                            msg="failed creating profile %r" % profile)
        if arches is None:
            arches = set(val[0] for val in profiles.itervalues())
        write_file(pjoin(loc, 'arch.list'), 'w', "\n".join(arches))
        write_file(pjoin(loc, 'repo_name'), 'w', 'testing')
        write_file(pjoin(loc, 'eapi'), 'w', '5')
        with open(pjoin(loc, 'profiles.desc'), 'w') as fd:
            for profile, vals in profiles.iteritems():
                l = len(vals)
                if l == 1 or not vals[1]:
                    fd.write("%s\t%s\tstable\n" % (vals[0], profile))
                else:
                    fd.write("%s\t%s\t%s\n" % (vals[0], profile, vals[1]))
                if l == 3 and vals[2]:
                    with open(pjoin(loc, profile, 'deprecated'), 'w') as f:
                        f.write("foon\n#dar\n")
                with open(pjoin(loc, profile, 'make.defaults'), 'w') as f:
                    f.write("ARCH=%s\n" % vals[0])
                with open(pjoin(loc, profile, 'eapi'), 'w') as f:
                    f.write('5')

def default_copyfile(obj, mkdirs=False):
    """
    copy a :class:`pkgcore.fs.fs.fsBase` to its stated location.

    :param obj: :class:`pkgcore.fs.fs.fsBase` instance, exempting :class:`fsDir`
    :return: true if success, else an exception is thrown
    :raise EnvironmentError: permission errors

    """

    existent = False
    ensure_perms = get_plugin("fs_ops.ensure_perms")
    if not fs.isfs_obj(obj):
        raise TypeError("obj must be fsBase derivative: %r" % obj)
    elif fs.isdir(obj):
        raise TypeError("obj must not be a fsDir instance: %r" % obj)

    try:
        existing = gen_obj(obj.location)
        if fs.isdir(existing):
            raise CannotOverwrite(obj, existing)
        existent = True
    except OSError as oe:
        # verify the parent dir is there at least
        basefp = os.path.dirname(obj.location)
        if basefp.strip(os.path.sep) and not os.path.exists(basefp):
            if mkdirs:
                if not ensure_dirs(basefp, mode=0750, minimal=True):
                    raise FailedCopy(obj, str(oe))