# Copyright (C) 2013-2015 Codethink Limited # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; version 2 of the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License along # with this program. If not, see . # # =*= License: GPL-2 =*= from gi.repository import OSTree from gi.repository import Gio from gi.repository import GLib import os import logging class OSTreeRepo(object): """Class to wrap the OSTree API.""" G_FILE_QUERY_INFO_NOFOLLOW_SYMLINKS = Gio.FileQueryInfoFlags(1) OSTREE_GIO_FAST_QUERYINFO = ( 'standard::name,' 'standard::type,' 'standard::size,' 'standard::is-symlink,' 'standard::symlink-target,' 'unix::device,' 'unix::inode,' 'unix::mode,' 'unix::uid,' 'unix::gid,' 'unix::rdev') def __init__(self, path, disable_fsync=True, mode='bare'): self.path = path self.repo = self._open_repo(path, disable_fsync, mode) def _open_repo(self, path, disable_fsync=True, mode='bare'): """Create and open and OSTree.Repo, and return it.""" repo_dir = Gio.file_new_for_path(path) repo = OSTree.Repo.new(repo_dir) logging.debug('using %s' % mode) if mode == 'bare': mode = OSTree.RepoMode.BARE elif mode == 'archive_z2': mode = OSTree.RepoMode.ARCHIVE_Z2 else: raise Exception('Mode %s is not supported' % mode) try: repo.open(None) logging.debug('opened') except GLib.GError: if not os.path.exists(path): os.makedirs(path) logging.debug('failed to open, creating') repo.create(mode, None) repo.set_disable_fsync(disable_fsync) return repo def refsdir(self): """Return the abspath to the refs/heads directory in the repo.""" return os.path.join(os.path.abspath(self.path), 'refs/heads') def touch_ref(self, ref): """Update the mtime of a ref file in repo/refs/heads.""" os.utime(os.path.join(self.refsdir(), ref), None) def resolve_rev(self, branch, allow_noent=True): """Return the SHA256 corresponding to 'branch'.""" return self.repo.resolve_rev(branch, allow_noent)[1] def read_commit(self, branch): """Return an OSTree.RepoFile representing a committed tree.""" return self.repo.read_commit(branch, None)[1] def query_info(self, file_object): """Quickly return a Gio.FileInfo for file_object.""" return file_object.query_info(self.OSTREE_GIO_FAST_QUERYINFO, self.G_FILE_QUERY_INFO_NOFOLLOW_SYMLINKS, None) def checkout(self, branch, destdir): """Checkout branch into destdir.""" checkout_path = destdir if not os.path.exists(checkout_path): os.makedirs(checkout_path) checkout = Gio.file_new_for_path(checkout_path) commit = self.read_commit(branch) commit_info = self.query_info(commit) self.repo.checkout_tree(OSTree.RepoCheckoutMode.NONE, OSTree.RepoCheckoutOverwriteMode.UNION_FILES, checkout, commit,commit_info, None) def commit(self, subject, srcdir, branch, body=''): """Commit the contents of 'srcdir' to 'branch'. The subject parameter is the title of the commit message, and the body parameter is the body of the commit message. """ self.repo.prepare_transaction(None) parent = self.resolve_rev(branch) mtree = OSTree.MutableTree() src = Gio.file_new_for_path(srcdir) self.repo.write_directory_to_mtree(src, mtree, None, None) root = self.repo.write_mtree(mtree, None)[1] checksum = self.repo.write_commit(parent, subject, body, None, root, None)[1] self.repo.transaction_set_ref(None, branch, checksum) stats = self.repo.commit_transaction(None) def cat_file(self, ref, path): """Return the file descriptor of path at ref.""" commit = self.read_commit(ref) relative = commit.resolve_relative_path(path) ret, content, etag = relative.load_contents() return content def list_refs(self, ref=None, resolved=False): """Return a list of all refs in the repo.""" if ref: refs = self.repo.list_refs(ref)[1] else: refs = self.repo.list_refs()[1] if not resolved: return refs.keys() return refs def delete_ref(self, ref): """Remove refspec from the repo.""" if not self.list_refs(ref): raise Exception("Failed to delete ref, it doesn't exist") self.repo.set_ref_immediate(None, ref, None, None) def prune(self): """Remove unreachable objects from the repo.""" depth = -1 # no recursion limit return self.repo.prune(OSTree.RepoPruneFlags.REFS_ONLY, depth, None) def add_remote(self, name, url): """Add a remote with a given name and url.""" options_type = GLib.VariantType.new('a{sv}') options_builder = GLib.VariantBuilder.new(options_type) options = options_builder.end() self.repo.remote_add(name, url, options, None) def remove_remote(self, name): """Remove a remote with a given name.""" self.repo.remote_delete(name, None) def get_remote_url(self, name): """Return the URL for a remote.""" return self.repo.remote_get_url(name)[1] def list_remotes(self): """Return a list of all remotes for this repo.""" return self.repo.remote_list() def has_remote(self, name): """Return True if name is a remote for the repo.""" return name in self.list_remotes() def pull(self, refs, remote): """Pull ref from remote into the local repo.""" flags = OSTree.RepoPullFlags.NONE self.repo.pull(remote, refs, flags, None, None)