path: root/subversion/bindings/ctypes-python/csvn/repos.py

#    Licensed to the Apache Software Foundation (ASF) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The ASF licenses this file
#    to you under the Apache License, Version 2.0 (the
#    "License"); you may not use this file except in compliance
#    with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing,
#    software distributed under the License is distributed on an
#    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#    KIND, either express or implied.  See the License for the
#    specific language governing permissions and limitations
#    under the License.


import csvn.core as svn
from csvn.core import *
import csvn.types as _types
from csvn.ext.callback_receiver import CallbackReceiver
from txn import Txn
from auth import User
import os

class RepositoryURI(object):
    """A URI to an object in a Subversion repository, stored internally in
       encoded format.

       When you supply URIs to a RemoteClient, or a transaction"""

    def __init__(self, uri, encoded=True):
        """Create a RepositoryURI object from a URI. If encoded=True, the
           input string may be URI-encoded."""
        pool = Pool()
        if not encoded:
            uri = svn_path_uri_encode(uri, pool)
        self._as_parameter_ = str(svn_path_canonicalize(uri, pool))

    def join(self, uri):
        """Join this URI and the specified relative URI,
           adding a slash if necessary."""
        pool = Pool()
        return RepositoryURI(svn_path_join(self, uri, pool))

    def dirname(self):
        """Get the parent directory of this URI"""
        pool = Pool()
        return RepositoryURI(svn_path_dirname(self, pool))

    def relative_path(self, uri, encoded=True):
        """Convert the supplied URI to a decoded path, relative to me."""
        pool = Pool()
        if not encoded:
            uri = svn_path_uri_encode(uri, pool)
        child_path = svn_path_is_child(self, uri, pool) or uri
        return str(svn_path_uri_decode(child_path, pool))

    def longest_ancestor(self, uri):
        """Get the longest ancestor of this URI and another URI"""
        pool = Pool()
        return RepositoryURI(svn_path_get_longest_ancestor(self, uri, pool))

    def __str__(self):
        """Return the URI as a string"""
        return self._as_parameter_

class RemoteRepository(object):
    """This class represents a connection from the client to a remote
       Subversion repository."""
    # The interface corresponds roughly to the svn_ra API, and an object of
    # this type basically represents the C type 'svn_ra_session_t'.

    def __init__(self, url, user=None):
        """Open a new session to URL with the specified USER.
           USER must be an object that implements the
           'csvn.auth.User' interface."""

        if user is None:
            user = User()

        self.pool = Pool()
        self.iterpool = Pool()
        self.url = RepositoryURI(url)
        self.user = user

        self.client = POINTER(svn_client_ctx_t)()
        svn_client_create_context(byref(self.client), self.pool)

        self.user.setup_auth_baton(pointer(self.client.contents.auth_baton))

        self._as_parameter_ = POINTER(svn_ra_session_t)()
        svn_client_open_ra_session(byref(self._as_parameter_), url,
                                   self.client, self.pool)

        self.client[0].log_msg_func2 = \
            svn_client_get_commit_log2_t(self._log_func_wrapper)
        self.client[0].log_msg_baton2 = c_void_p()
        self._log_func = None

    def close(self):
        """Close this RemoteRepository object, releasing any resources."""
        self.pool.clear()

    def txn(self):
        """Create a transaction"""
        return Txn(self)

    def latest_revnum(self):
        """Get the latest revision number in the repository"""
        revnum = svn_revnum_t()
        svn_ra_get_latest_revnum(self, byref(revnum), self.iterpool)
        self.iterpool.clear()
        return revnum.value

    def check_path(self, path, rev = None, encoded=True):
        """Check the status of PATH@REV. If REV is not specified,
           look at the latest revision in the repository.

        If the path is ...
          ... absent, then we return svn_node_none.
          ... a regular file, then we return svn_node_file.
          ... a directory, then we return svn_node_dir
          ... unknown, then we return svn_node_unknown

        If ENCODED is True, the path may be URI-encoded.
        """

        path = self._relative_path(path, encoded)
        if rev is None:
            rev = self.latest_revnum()
        kind = svn_node_kind_t()
        svn_ra_check_path(self, path, svn_revnum_t(rev), byref(kind),
                          self.iterpool)
        self.iterpool.clear()
        return kind.value

    def list(self, path, rev = SVN_INVALID_REVNUM, fields = SVN_DIRENT_ALL):
        """List the contents of the specified directory PATH@REV. This
           function returns a dictionary, which maps entry names to
           directory entries (svn_dirent_t objects).

           If REV is not specified, we look at the latest revision of the
           repository.

           FIELDS controls what portions of the svn_dirent_t object are
           filled in. To have them completely filled in, just pass in
           SVN_DIRENT_ALL (which is the default); otherwise, pass the
           bitwise OR of all the SVN_DIRENT_ fields you would like to
           have returned to you.
        """
        dirents = _types.Hash(POINTER(svn_dirent_t), None)
        svn_ra_get_dir2(self, dirents.byref(), NULL, NULL, path,
                        rev, fields, dirents.pool)
        self.iterpool.clear()
        # Create a Python dict of svn_dirent_t objects from this Hash of
        # pointers to svn_dirent_t.
        result = {}
        for path, dirent_p in dirents.items():
            result[path] = dirent_p[0]
        return result

    def cat(self, buffer, path, rev = SVN_INVALID_REVNUM):
        """Get PATH@REV and save it to BUFFER. BUFFER must be a Python file
           or a StringIO object.

           If REV is not specified, we look at the latest revision of the
           repository."""
        stream = _types.Stream(buffer)
        svn_ra_get_file(self, path, rev, stream, NULL, NULL, stream.pool)
        self.iterpool.clear()

    def info(self, path, rev = None):
        """Get a pointer to a svn_dirent_t object associated with PATH@REV.
           If PATH does not exist, return None.

           If REV is not specified, we look at the latest revision of the
           file."""
        dirent = POINTER(svn_dirent_t)()
        dirent.pool = Pool()
        if rev is None:
            rev = self.latest_revnum()
        svn_ra_stat(self, path, rev, byref(dirent), dirent.pool)
        self.iterpool.clear()
        return dirent[0]

    def proplist(self, path, rev = SVN_INVALID_REVNUM):
        """Return a dictionary containing the properties on PATH@REV

           If REV is not specified, we look at the latest revision of the
           repository."""

        props = _types.Hash(POINTER(svn_string_t), None,
                   wrapper=_types.SvnStringPtr)
        status = self.check_path(path, rev)
        if status == svn_node_dir:
            svn_ra_get_dir2(self, NULL, NULL, props.byref(), path,
                            rev, 0, props.pool)
        else:
            svn_ra_get_file(self, path, rev, NULL, NULL, props.byref(),
                            props.pool)
        self.iterpool.clear()
        return props

    def propget(self, name, path, rev = SVN_INVALID_REVNUM):
        """Get property NAME from PATH@REV.

           If REV is not specified, we look at the latest revision of the
           repository."""

        return self.proplist(path, rev)[name]

    def log(self, start_rev, end_rev, paths=None, limit=0,
            discover_changed_paths=FALSE, stop_on_copy=FALSE):
        """A generator function which returns information about the revisions
           between START_REV and END_REV. Each return value is a
           csvn.types.LogEntry object which describes a revision.

           For details on what fields are contained in a LogEntry object,
           please see the documentation from csvn.types.LogEntry.

           You can iterate through the log information for several revisions
           using a regular for loop. For example:
             for entry in session.log(start_rev, end_rev):
               print("Revision %d" % entry.revision)
               ...

           ARGUMENTS:

             If PATHS is not None and has one or more elements, then only
             show revisions in which at least one of PATHS was changed (i.e.,
             if file, text or props changed; if dir, props changed or an entry
             was added or deleted). Each PATH should be relative to the current
             session's root.

             If LIMIT is non-zero, only the first LIMIT logs are returned.

             If DISCOVER_CHANGED_PATHS is True, then changed_paths will contain
             a list of paths affected by this revision.

             If STOP_ON_COPY is True, then this function will not cross
             copies while traversing history.

             If START_REV or END_REV is a non-existent revision, we throw
             a SVN_ERR_FS_NO_SUCH_REVISION SubversionException, without
             returning any logs.

        """

        paths = _types.Array(c_char_p, paths is None and [""] or paths)
        return iter(_LogMessageReceiver(self, start_rev, end_rev, paths,
                                        limit, discover_changed_paths, stop_on_copy))


    # Private. Produces a delta editor for the commit, so that the Txn
    # class can commit its changes over the RA layer.
    def _get_commit_editor(self, message, commit_callback, commit_baton, pool):
        editor = POINTER(svn_delta_editor_t)()
        editor_baton = c_void_p()
        svn_ra_get_commit_editor2(self, byref(editor),
            byref(editor_baton), message, commit_callback,
            commit_baton, NULL, FALSE, pool)
        return (editor, editor_baton)

    # Private. Convert a URI to a repository-relative path
    def _relative_path(self, path, encoded=True):
        return self.url.relative_path(path, encoded)

    # Private. Convert a repository-relative copyfrom path into a proper
    # copyfrom URI
    def _abs_copyfrom_path(self, path):
        return self.url.join(RepositoryURI(path, False))

    def revprop_list(self, revnum=None):
        """Returns a hash of the revision properties of REVNUM. If REVNUM is
        not provided, it defaults to the head revision."""
        rev = svn_opt_revision_t()
        if revnum is not None:
            rev.kind = svn_opt_revision_number
            rev.value.number = revnum
        else:
            rev.kind = svn_opt_revision_head

        props = _types.Hash(POINTER(svn_string_t), None,
                   wrapper=_types.SvnStringPtr)

        set_rev = svn_revnum_t()

        svn_client_revprop_list(props.byref(),
                        self.url,
                        byref(rev),
                        byref(set_rev),
                        self.client,
                        props.pool)

        self.iterpool.clear()

        return props

    def revprop_get(self, propname, revnum=None):
        """Returns the value of PROPNAME at REVNUM. If REVNUM is not
        provided, it defaults to the head revision."""
        return self.revprop_list(revnum)[propname]

    def revprop_set(self, propname, propval=NULL, revnum=None, force=False):
        """Set PROPNAME to PROPVAL for REVNUM. If REVNUM is not given, it
        defaults to the head revision. Returns the actual revision number
        effected.

        If PROPVAL is not provided, the property will be deleted.

        If FORCE is True (False by default), newlines will be allowed in the
        author property.

        Be careful, this is a lossy operation."""
        rev = svn_opt_revision_t()
        if revnum is not None:
            rev.kind = svn_opt_revision_number
            rev.value.number = revnum
        else:
            rev.kind = svn_opt_revision_head

        set_rev = svn_revnum_t()

        svn_client_revprop_set(propname,
                svn_string_create(propval, self.iterpool), self.url,
                byref(rev), byref(set_rev), force, self.client,
                self.iterpool)

        try:
            return set_rev.value
        finally:
            self.iterpool.clear()


    def set_log_func(self, log_func):
        """Register a callback to get a log message for commit and
        commit-like operations. LOG_FUNC should take an array as an argument,
        which holds the files to be committed. It should return a list of the
        form [LOG, FILE] where LOG is a log message and FILE is the temporary
        file, if one was created instead of a log message. If LOG is None,
        the operation will be canceled and FILE will be treated as the
        temporary file holding the temporary commit message."""
        self._log_func = log_func

    def _log_func_wrapper(self, log_msg, tmp_file, commit_items, baton, pool):
        log_msg[0].raw = NULL
        tmp_file[0] = NULL

        if self._log_func:
            [log, file] = self._log_func(_types.Array(String, commit_items))

            if log:
                log_msg[0].raw = apr_pstrdup(pool, String(log)).raw
            if file:
                tmp_file[0] = apr_pstrdup(pool, String(file)).raw

    def svnimport(self, path, url=None, nonrecursive=False, no_ignore=True, log_func=None):

        if not url:
            url = self.url

        if log_func:
            self.set_log_func(log_func)

        pool = Pool()
        commit_info = POINTER(svn_commit_info_t)()
        svn_client_import2(byref(commit_info), path, url, nonrecursive,
                           no_ignore, self.client, pool)

        commit_info[0].pool = pool
        return commit_info[0]

class LocalRepository(object):
    """A client which accesses the repository directly. This class
       may allow you to perform some administrative actions which
       cannot be performed remotely (e.g. create repositories,
       dump repositories, etc.)

       Unlike RemoteRepository, the functions in this class do not
       accept URIs, and instead only accept local filesystem
       paths.

       By default, this class does not perform any checks to verify
       permissions, assuming that the specified user has full
       administrative access to the repository. To teach this class
       to enforce an authz policy, you must subclass csvn.auth.User
       and implement the allow_access function.
    """

    def __init__(self, path, create=False, user=None):
        """Open the repository at PATH. If create is True,
           create a new repository.

           If specified, user must be a csvn.auth.User instance.
        """
        if user is None:
            user = User()

        self.pool = Pool()
        self.iterpool = Pool()
        self._as_parameter_ = POINTER(svn_repos_t)()
        self.user = user
        if create:
            svn_repos_create(byref(self._as_parameter_), path,
                             None, None, None, None, self.pool)
        else:
            svn_repos_open(byref(self._as_parameter_), path, self.pool)
        self.fs = _fs(self)

    def __del__(self):
        self.close()

    def close(self):
        """Close this LocalRepository object, releasing any resources. In
           particular, this closes the rep-cache DB."""
        self.pool.clear()

    def latest_revnum(self):
        """Get the latest revision in the repository"""
        return self.fs.latest_revnum()

    def check_path(self, path, rev = None, encoded=False):
        """Check whether the given PATH exists in the specified REV. If REV
           is not specified, look at the latest revision.

        If the path is ...
          ... absent, then we return svn_node_none.
          ... a regular file, then we return svn_node_file.
          ... a directory, then we return svn_node_dir
          ... unknown, then we return svn_node_unknown
        """
        assert(not encoded)
        root = self.fs.root(rev=rev, pool=self.iterpool)
        try:
            return root.check_path(path)
        finally:
            self.iterpool.clear()

    def uuid(self):
        """Return a universally-unique ID for this repository"""
        return self.fs.uuid()

    def set_rev_prop(self, rev, name, value, author=NULL):
        """Set the NAME property to VALUE in the specified
           REV, attribute the change to AUTHOR if provided."""
        rev = svn_revnum_t(rev)
        svn_repos_fs_change_rev_prop2(self, rev, author, name, value,
                                      svn_repos_authz_func_t(),
                                      None, self.iterpool)
        self.iterpool.clear()

    def get_rev_prop(self, rev, name):
        """Returns the value of NAME in REV. If NAME does not exist in REV,
        returns None."""
        rev = svn_revnum_t(rev)
        value = POINTER(svn_string_t)()

        svn_repos_fs_revision_prop(byref(value), self, rev, name,
                                   svn_repos_authz_func_t(), None,
                                   self.iterpool)

        try:
            if value:
                return _types.SvnStringPtr.from_param(value)
            else:
                return None
        finally:
            self.iterpool.clear()

    def txn(self):
        """Open up a new transaction, so that you can commit a change
           to the repository"""
        assert self.user is not None, (
               "If you would like to commit changes to the repository, "
               "you must supply a user object when you initialize "
               "the repository object")
        return Txn(self)

    # Private. Produces a delta editor for the commit, so that the Txn
    # class can commit its changes over the RA layer.
    def _get_commit_editor(self, message, commit_callback, commit_baton, pool):
        editor = POINTER(svn_delta_editor_t)()
        editor_baton = c_void_p()
        svn_repos_get_commit_editor4(byref(editor),
            byref(editor_baton), self, None, "", "",
            self.user.username(), message,
            commit_callback, commit_baton, svn_repos_authz_callback_t(),
            None, pool)
        return (editor, editor_baton)

    def _relative_path(self, path):
        return path

    # Private. Convert a repository-relative copyfrom path into a proper
    # copyfrom URI
    def _abs_copyfrom_path(self, path):
        return path

    def load(self, dumpfile, feedbackfile=None,
            uuid_action=svn_repos_load_uuid_default, parent_dir="",
            use_pre_commit_hook=False, use_post_commit_hook=False,
            cancel_func=None):
        """Read and parse dumpfile-formatted DUMPFILE, reconstructing
        filesystem revisions. Dumpfile should be an open python file object
        or file like object. UUID will be handled according to UUID_ACTION
        which defaults to svn_repos_load_uuid_default.

        If FEEDBACKFILE is provided (in the form of a python file object or
        file like object), feedback will be sent to it.

        If PARENT_DIR is provided, everything loaded from the dump will be
        reparented to PARENT_DIR.

        USE_PRE_COMMIT_HOOK and USE_POST_COMMIT_HOOK are False by default,
        if either is set to True that hook will be used.

        If CANCEL_FUNC is provided, it will be called at various points to
        allow the operation to be cancelled. The cancel baton will be the
        LocalRepository object."""

        if not cancel_func:
            cancel_func = svn_cancel_func_t()

        apr_dump = _types.APRFile(dumpfile)
        stream_dump = svn_stream_from_aprfile2(apr_dump._as_parameter_,
                                               False, self.iterpool)

        if feedbackfile:
            apr_feedback = _types.APRFile(feedbackfile)
            stream_feedback = svn_stream_from_aprfile2(
                                apr_feedback._as_parameter_, False,
                                self.iterpool)
        else:
            stream_feedback = NULL

        svn_repos_load_fs2(self._as_parameter_, stream_dump, stream_feedback,
                           uuid_action, parent_dir, use_pre_commit_hook,
                           use_post_commit_hook, cancel_func,
                           c_void_p(), self.iterpool)

        apr_dump.close()
        if feedbackfile:
            apr_feedback.close()

        self.iterpool.clear()

class _fs(object):
    """NOTE: This is a private class. Don't use it outside of
       this module. Use the Repos class instead.

       This class represents an svn_fs_t object"""

    def __init__(self, repos):
        self.iterpool = Pool()
        self._as_parameter_ = svn_repos_fs(repos)

    def latest_revnum(self):
        """See Repos.latest_revnum"""
        rev = svn_revnum_t()
        svn_fs_youngest_rev(byref(rev), self, self.iterpool)
        self.iterpool.clear()
        return rev.value

    def uuid(self):
        """See Repos.uuid"""
        uuid_buffer = String()
        svn_fs_get_uuid(self, byref(uuid_buffer), self.iterpool)
        uuid_str = str(uuid_buffer)
        self.iterpool.clear()
        return uuid_str

    def root(self, rev = None, txn = None, pool = None,
             iterpool = None):
        """Create a new svn_fs_root_t object from txn or rev.
           If neither txn nor rev or set, this root object will
           point to the latest revision root.

           The svn_fs_root object itself will be allocated in pool.
           If iterpool is supplied, iterpool will be used for any
           temporary allocations. Otherwise, pool will be used for
           temporary allocations."""
        return _fs_root(self, rev, txn, pool, iterpool)

class _fs_root(object):
    """NOTE: This is a private class. Don't use it outside of
       this module. Use the Repos.txn() method instead.

       This class represents an svn_fs_root_t object"""

    def __init__(self, fs, rev = None, txn = None, pool = None,
                 iterpool = None):
        """See _fs.root()"""

        assert(pool)

        self.pool = pool
        self.iterpool = iterpool or pool
        self.fs = fs
        self._as_parameter_ = POINTER(svn_fs_root_t)()

        if txn and rev:
            raise Exception("You can't specify both a txn and a rev")

        if txn:
            svn_fs_txn_root(byref(self._as_parameter_), txn, self.pool)
        else:
            if not rev:
                rev = fs.latest_revnum()
            svn_fs_revision_root(byref(self._as_parameter_), fs, rev, self.pool)

    def check_path(self, path):
        """Check whether the specified path exists in this root.
           See Repos.check_path() for details."""

        kind = svn_node_kind_t()
        svn_fs_check_path(byref(kind), self, path, self.iterpool)

        return kind.value

class LogEntry(object):
    """REVISION, AUTHOR, DATE, and MESSAGE are straightforward, and
       contain what you expect. DATE is a csvn.types.SvnDate object.

       If no information about the paths changed in this revision is
       available, CHANGED_PATHS will be None. Otherwise, CHANGED_PATHS
       will contain a dictionary which maps every path committed
       in REVISION to svn_log_changed_path_t pointers."""

    __slots__ = ['changed_paths', 'revision',
                 'author', 'date', 'message']

class _LogMessageReceiver(CallbackReceiver):

    def collect(self, session, start_rev, end_rev, paths, limit,
                discover_changed_paths, stop_on_copy):
        self.discover_changed_paths = discover_changed_paths
        pool = Pool()
        baton = c_void_p()
        receiver = svn_log_message_receiver_t(self.receive)
        svn_ra_get_log(session, paths, start_rev, end_rev,
                       limit, discover_changed_paths, stop_on_copy, receiver,
                       baton, pool)

    def receive(self, baton, changed_paths, revision, author, date, message, pool):
        entry = LogEntry()

        # Save information about the log entry
        entry.revision = revision
        entry.author = str(author)
        entry.date = _types.SvnDate(date)
        entry.message = str(message)

        if self.discover_changed_paths:
            entry.changed_paths = _types.Hash(POINTER(svn_log_changed_path_t),
              changed_paths, dup = svn_log_changed_path_dup)
        else:
            entry.changed_paths = None

        self.send(entry)