Commits

Thomas Waldmann committed b2542a3

remove current storage code

  • Participants
  • Parent commits 7f0a567
  • Branches storage-ng

Comments (0)

Files changed (32)

File MoinMoin/storage/__init__.py

-# Copyright: 2008 MoinMoin:ChristopherDenter
-# Copyright: 2008 MoinMoin:JohannesBerg
-# Copyright: 2009-2010 MoinMoin:ThomasWaldmann
-# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
-
-"""
-    MoinMoin - Backends - Storage API Definition.
-
-    The storage API consists of the classes defined in this module. That is:
-    Backend, Item, Revision, NewRevision and StoredRevision.
-
-    A concrete backend implements the abstract methods defined by the API,
-    but also uses concrete methods that have already been defined in this
-    module.
-    A backend is a collection of items. Examples for backends include SQL,
-    mercurial or filesystem. All of those are means to store data.
-
-    Items are the units you store within those backends. You can store content
-    of arbitrary type in an item, e.g. text, images or even films.
-
-    An item itself has revisions and metadata. For instance, you can use that
-    to show a diff between two `versions` of a page, where the page "Foo" is
-    represented by an item and the two versions are represented by two
-    revisions of that item.
-
-    Metadata is data that describes other data. An item has metadata. Each
-    revision has metadata as well. E.g. "Which user created this revision?"
-    would be something stored in the metadata of a revision, while "Who created
-    this page in the first place?" would be answered by looking at the metadata
-    of the first revision. Thus, an item basically is a collection of revisions
-    which contain the content for the item. The last revision represents the most
-    recent contents. A stored item can have metadata or revisions, or both.
-
-    For normal operation, revision data and metadata are immutable as soon as the
-    revision is committed to storage (by calling the commit() method on the item
-    that holds the revision), thus making it a StoredRevision.
-    Item metadata, on the other hand, as infrequently used as it may be, is mutable.
-    Hence, it can only be modified under a read lock.
-"""
-
-
-import os
-import sys
-import shutil
-import time
-
-from MoinMoin import log
-logging = log.getLogger(__name__)
-
-from UserDict import DictMixin
-from MoinMoin.storage.error import RevisionNumberMismatchError, AccessError, \
-                                   BackendError, NoSuchItemError, \
-                                   RevisionAlreadyExistsError, ItemAlreadyExistsError
-
-from MoinMoin.config import SIZE, MTIME, HASH_ALGORITHM
-
-import hashlib
-
-
-class Backend(object):
-    """
-    This class abstracts access to backends. If you want to write a specific
-    backend, say a mercurial backend, you have to implement the methods below.
-    A backend knows of its items and can perform several item related operations
-    such as get_item, create_item, etc.
-    """
-    #
-    # If you need to write a backend it is sufficient
-    # to implement the methods of this class. That
-    # way you don't *have to* implement the other classes
-    # like Item and Revision as well. Though, if you want
-    # to, you can do it as well.
-    # Assuming my_item is instanceof(Item), when you call
-    # my_item.create_revision(42), internally the
-    # _create_revision() method of the item's backend is
-    # invoked and the item passes itself as parameter.
-    #
-    def __init__(self, *args, **kw):
-        """
-        Create the backend instance.
-        """
-        pass
-
-    def close(self):
-        """
-        Close all resources the backend is using.
-        """
-        pass
-
-    def get_item(self, itemname):
-        """
-        Returns item object or raises Exception if that item does not exist.
-
-        When implementing this, don't rely on has_item unless you've overridden it.
-
-        :type itemname: unicode
-        :param itemname: The name of the item we want to get.
-        :rtype: item object
-        :raises NoSuchItemError: No item with name 'itemname' is known to this backend.
-        """
-        raise NotImplementedError()
-
-    def has_item(self, itemname):
-        """
-        Override this method!
-
-        This method is added for convenience. With it you don't need to try get_item
-        and catch an exception that may be thrown if the item doesn't exist yet.
-
-        :type itemname: unicode
-        :param itemname: The name of the item of which we want to know whether it exists.
-        :rtype: bool
-        """
-        try:
-            self.get_item(itemname)
-            return True
-        except NoSuchItemError:
-            return False
-
-    def create_item(self, itemname):
-        """
-        Creates an item with a given itemname. If that item already exists,
-        raise an exception.
-
-        :type itemname: unicode
-        :param itemname: Name of the item we want to create.
-        :rtype: item object
-        :raises ItemAlreadyExistsError: The item you were trying to create already exists.
-        """
-        raise NotImplementedError()
-
-    def iteritems(self):
-        """
-        Iterate over all items.
-
-        May use an index internally to optimize.
-
-        :rtype: iterator of item objects
-        """
-        raise NotImplementedError()
-
-    def iter_items_noindex(self):
-        """
-        Iterate over all items.
-
-        Must not use an index as this method is used to *build* the index.
-
-        :rtype: iterator of item objects
-        """
-        raise NotImplementedError()
-
-    def _get_revision(self, item, revno):
-        """
-        For a given item and revision number, return the corresponding revision
-        of that item.
-        Note: If you pass -1 as revno, this shall return the latest revision of the item.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        :type revno: int
-        :param revno: Indicate which revision is wanted precisely. If revno is
-        -1, return the most recent revision.
-        :rtype: Object of class Revision
-        :raises NoSuchRevisionError: No revision with that revno was found on item.
-        """
-        raise NotImplementedError()
-
-    def _list_revisions(self, item):
-        """
-        For a given item, return a list containing all revision numbers (as ints)
-        of the revisions the item has. The list must be ordered, starting with
-        the oldest revision number.
-        Since we allow to totally destroy certain revisions, list_revisions does
-        not need to return subsequent, but only monotone revision numbers.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        :returns: list of ints (possibly empty)
-        """
-        raise NotImplementedError()
-
-    def _create_revision(self, item, revno):
-        """
-        Takes an item object and creates a new revision. Note that you need to pass
-        a revision number for concurrency reasons. The revno passed must be
-        greater than the revision number of the item's most recent revision.
-        The newly created revision object is returned to the caller.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        :type revno: int
-        :param revno: Indicate which revision we want to create.
-        @precondition: item.get_revision(-1).revno < revno
-        :returns: Object of class Revision.
-        :raises RevisionAlreadyExistsError: Raised if a revision with that number
-        already exists on item.
-        :raises RevisionNumberMismatchError: Raised if precondition is not
-        fulfilled.
-        """
-        raise NotImplementedError()
-
-    def _destroy_revision(self, revision):
-        """
-        Similarly to self._destroy_item. The given revision is completely destroyed.
-        As this is an irreversible action, great care must be taken when performing it.
-
-        In case the revision has already been destroyed by someone else (e.g. another
-        process) this method should just pass silently as the job is already done.
-
-        If the revision cannot be destroyed for technical reasons (e.g. missing permissions
-        on disk), this method shall raise a CouldNotDestroyError.
-
-        Note: Again, backends not capable of really erasing something should at the very
-              least ignore the existence of the revision in question. (The only hint will
-              be the gap in item.list_revisions().
-
-        :type revision: Object of class StoredRevision
-        :param revision: The revision we want to destroy completely.
-        :raises CouldNotDestroyError: Raised in case the revision could not be destroyed.
-        """
-        raise NotImplementedError()
-
-    def _rename_item(self, item, newname):
-        """
-        Renames a given item. Raises Exception if the item you are trying to rename
-        does not exist or if the newname is already chosen by another item.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        :type newname: string
-        :param newname: Name of item after this operation has succeeded.
-        @precondition: self.has_item(newname) == False
-        @postcondition: self.has_item(newname) == True
-        :raises ItemAlreadyExistsError: Raised if an item with name 'newname'
-        already exists.
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _commit_item(self, revision):
-        """
-        Commits the changes that have been done to a given item. That is, after you
-        created a revision on that item and filled it with data you still need to
-        commit() it. You need to pass the revision you want to commit. The item
-        can be looked up by the revision's 'item' property.
-
-        :type revision: Object of class NewRevision.
-        :param revision: The revision we want to commit to  storage.
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _rollback_item(self, revision):
-        """
-        This method is invoked when external events happen that cannot be handled in a
-        sane way and thus the changes that have been made must be rolled back.
-
-        :type revision: Object of class NewRevision.
-        :param revision: The revision we want to roll back.
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _destroy_item(self, item):
-        """
-        Use this method carefully!
-
-        This method attempts to completely *destroy* an item with all its revisions and
-        metadata. After that, it will be impossible to access the item again via the
-        storage API. This is very different from the deletion a user can perform on
-        a wiki item, as such a deletion does not really delete anything from disk but
-        just hides the former existence of the item. Such a deletion is undoable, while
-        having destroyed an item is not.
-
-        In case the item has already been destroyed by someone else (e.g. another process)
-        this method should just pass silently as the job is already done.
-
-        If the item cannot be destroyed for technical reasons (e.g. missing permissions
-        on disk), this method shall raise a CouldNotDestroyError.
-
-        Note: Several backends (in particular those based on VCS) do not, by their nature,
-              support erasing any content that has been put into them at some point.
-              Those backends then need to emulate erasure as best they can. They should at
-              least ignore the former existence of the item completely.
-              A wiki admin must be aware that when using such a backend, he either needs
-              to invoke an erasure (clone old, dirtied backend to new, fresh backend) script
-              from time to time to get rid of the stuff, or not choose a backend of this
-              kind (in case disk space is limited and large items are uploaded).
-
-        :type item: Object of class Item
-        :param item: The item we want to destroy
-        :raises CouldNotDestroyError: Raised in case the revision could not be destroyed.
-        :returns: None
-        """
-        # XXX Should this perhaps return a bool indicating whether erasure was actually performed on disk or something like that?
-        raise NotImplementedError()
-
-    def _change_item_metadata(self, item):
-        """
-        This method is used to acquire a lock on an item. This is necessary to prevent
-        side effects caused by concurrency.
-
-        You need to call this method before altering the metadata of the item.
-        E.g.:   item.change_metadata()  # Invokes this method
-                item["metadata_key"] = "metadata_value"
-                item.publish_metadata()
-
-        As you can see, the lock acquired by this method is released by calling
-        the publish_metadata() method on the item.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        @precondition: item not already locked
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _publish_item_metadata(self, item):
-        """
-        This method tries to release a lock on the given item and put the newly
-        added Metadata of the item to storage.
-
-        You need to call this method after altering the metadata of the item.
-        E.g.:   item.change_metadata()
-                item["metadata_key"] = "metadata_value"
-                item.publish_metadata()  # Invokes this method
-
-        The lock this method releases is acquired by the _change_metadata method.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        :raises AssertionError: item was not locked XXX use more special exception
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _read_revision_data(self, revision, chunksize):
-        """
-        Called to read a given amount of bytes of a revision's data. By default, all
-        data is read.
-
-        :type revision: Object of class StoredRevision.
-        :param revision: The revision on which we want to operate.
-        :type chunksize: int
-        :param chunksize: amount of bytes to be read at a time
-        :returns: string
-        """
-        raise NotImplementedError()
-
-    def _write_revision_data(self, revision, data):
-        """
-        When this method is called, the passed data is written to the revision's data.
-
-        :type revision: Object of class NewRevision.
-        :param revision: The revision on which we want to operate.
-        :type data: str
-        :param data: The data to be written on the revision.
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _get_item_metadata(self, item):
-        """
-        Load metadata for a given item, return dict.
-
-        :type item: Object of class Item.
-        :param item: The Item on which we want to operate.
-        :returns: dict of metadata key / value pairs.
-        """
-        raise NotImplementedError()
-
-    def _get_revision_metadata(self, revision):
-        """
-        Load metadata for a given revision, returns dict.
-
-        :type revision: Object of a subclass of Revision.
-        :param revision: The revision on which we want to operate.
-        :returns: dict of metadata key / value pairs.
-        """
-        raise NotImplementedError()
-
-    def _seek_revision_data(self, revision, position, mode):
-        """
-        Set the revision's cursor on the revision's data.
-
-        :type revision: Object of StoredRevision.
-        :param revision: The revision on which we want to operate.
-        :type position: int
-        :param position: Indicates where to position the cursor
-        :type mode: int
-        :param mode: 0 for 'absolute positioning', 1 to seek 'relatively to the
-        current position', 2 to seek 'relative to the files end'.
-        :returns: None
-        """
-        raise NotImplementedError()
-
-    def _tell_revision_data(self, revision):
-        """
-        Tell the revision's cursor's position on the revision's data.
-
-        :type revision: Object of type StoredRevision.
-        :param revision: The revision on which tell() was invoked.
-        :returns: int indicating the cursor's position.
-        """
-        raise NotImplementedError()
-
-    # item copying
-    def _copy_item_progress(self, verbose, st):
-        if verbose:
-            progress_char = dict(converts='.', skips='s', fails='F')
-            sys.stdout.write(progress_char[st])
-
-    def copy_item(self, item, verbose=False, name=None):
-        def same_revision(rev1, rev2):
-            for k, v in rev1.iteritems():
-                if rev2[k] != v:
-                    return False
-            return True
-
-        if name is None:
-            name = item.name
-
-        status = dict(converts={}, skips={}, fails={})
-        revisions = item.list_revisions()
-
-        try:
-            new_item = self.get_item(name)
-        except NoSuchItemError:
-            new_item = self.create_item(name)
-
-        # This only uses the metadata of the item that we clone.
-        # Arguments for doing this:
-        #   * If old stuff ends up in item after clone, that'd be counter intuitive
-        #   * When caching some data from the latest rev in the item, we don't want the old stuff.
-        new_item.change_metadata()
-        for k, v in item.iteritems():
-            new_item[k] = v
-        new_item.publish_metadata()
-
-        for revno in revisions:
-            revision = item.get_revision(revno)
-
-            try:
-                new_rev = new_item.create_revision(revision.revno)
-            except RevisionAlreadyExistsError:
-                existing_revision = new_item.get_revision(revision.revno)
-                st = same_revision(existing_revision, revision) and 'skips' or 'fails'
-            else:
-                for k, v in revision.iteritems():
-                    new_rev[k] = v
-                shutil.copyfileobj(revision, new_rev)
-                new_item.commit()
-                st = 'converts'
-            try:
-                status[st][name].append(revision.revno)
-            except KeyError:
-                status[st][name] = [revision.revno]
-            self._copy_item_progress(verbose, st)
-
-        return status['converts'], status['skips'], status['fails']
-
-    # cloning support
-    def _clone_before(self, source, verbose):
-        if verbose:
-            # reopen stdout file descriptor with write mode
-            # and 0 as the buffer size (unbuffered)
-            sys.stdout = os.fdopen(os.dup(sys.stdout.fileno()), 'w', 0)
-            sys.stdout.write("[converting %s to %s]: " % (source.__class__.__name__,
-                                                          self.__class__.__name__, ))
-
-    def _clone_after(self, source, verbose):
-        if verbose:
-            sys.stdout.write("\n")
-
-    def clone(self, source, verbose=False, only_these=[]):
-        """
-        Create exact copy of source Backend with all the Items into THIS
-        backend. If you don't want all items, you can give an item name list
-        in only_these.
-
-        Note: this is a generic implementation, you can maybe specialize it to
-              make it faster in your backend implementation (esp. if creating
-              new items is expensive).
-
-        Return a tuple consisting of three dictionaries (Item name:Revision
-        numbers list): converted, skipped and failed Items dictionary.
-        """
-        def item_generator(source, only_these):
-            if only_these:
-                for name in only_these:
-                    try:
-                        yield source.get_item(name)
-                    except NoSuchItemError:
-                        # TODO Find out why this fails sometimes.
-                        #sys.stdout.write("Unable to copy %s\n" % itemname)
-                        pass
-            else:
-                for item in source.iteritems():
-                    yield item
-
-        self._clone_before(source, verbose)
-
-        converts, skips, fails = {}, {}, {}
-        for item in item_generator(source, only_these):
-            c, s, f = self.copy_item(item, verbose)
-            converts.update(c)
-            skips.update(s)
-            fails.update(f)
-
-        self._clone_after(source, verbose)
-        return converts, skips, fails
-
-
-class Item(object, DictMixin):
-    """
-    An item object collects the information of an item (e.g. a page) that is
-    stored in persistent storage. It has metadata and revisions.
-    An item object is just a proxy to the information stored in the backend.
-    It doesn't necessarily live very long.
-
-    Do NOT create instances of this class directly, but use backend.get_item
-    or backend.create_item!
-    """
-    def __init__(self, backend, itemname):
-        """
-        Initialize an item. Memorize the backend to which it belongs.
-
-        :type backend: Object of a subclass of Backend.
-        :param backend: The backend that stores this item.
-        :type itemname: unicode
-        :param itemname: The name representing this item in the backend. Unique
-        within the backend.
-        """
-        self._backend = backend
-        self._name = itemname
-        self._locked = False
-        self._read_accessed = False
-        self._metadata = None  # Will be loaded lazily upon first real access.
-        self._uncommitted_revision = None
-
-    def get_name(self):
-        """
-        name is a read-only property of this class.
-        """
-        return self._name
-
-    name = property(get_name, doc="This is the name of this item. This attribute is read-only.")
-
-    @property
-    def next_revno(self):
-        """
-        The revno of the most recent committed revision + 1.
-        I.e., the next revision's revno.
-        """
-        revs = self.list_revisions()
-        try:
-            return revs[-1] + 1
-        except IndexError:
-            # No revisions yet (empty sequence)
-            return 0
-
-    def __setitem__(self, key, value):
-        """
-        In order to access the item's metadata you can use the well-known dict-like
-        semantics Python's dictionaries offer. If you want to set a value,
-        my_item["key"] = "value" will do the trick. Note that keys must be of the
-        type string (or unicode).
-        Values must be of the type str, unicode or tuple, in which case every element
-        of the tuple must be a string, unicode or tuple object.
-        You must wrap write accesses to metadata in change_metadata/publish_metadata calls.
-        Keys starting with two underscores are reserved and cannot be used.
-
-        :type key: str or unicode
-        :param key: The keyword that is used to look up the corresponding value.
-        :type value: str, unicode, int, long, float, bool, complex or a nested tuple thereof.
-        :param value: The value that is referenced by the keyword `key` in this
-        specific item's metadata dict.
-        """
-        if not self._locked:
-            raise AttributeError("Cannot write to unlocked metadata")
-        if not isinstance(key, (str, unicode)):
-            raise TypeError("Key must be string type")
-        if key.startswith('__'):
-            raise TypeError("Key must not begin with two underscores")
-        check_value_type_is_valid(value)
-        if self._metadata is None:
-            self._metadata = self._backend._get_item_metadata(self)
-        self._metadata[key] = value
-
-    def __delitem__(self, key):
-        """
-        Delete an item metadata key/value pair.
-
-        :type key: str or unicode
-        :param key: Key identifying a unique key/value pair in this item's metadata.
-        @postcondition: self[key] raises KeyError
-        """
-        if not self._locked:
-            raise AttributeError("Cannot write to unlocked metadata")
-        if key.startswith('__'):
-            raise KeyError(key)
-        if self._metadata is None:
-            self._metadata = self._backend._get_item_metadata(self)
-        del self._metadata[key]
-
-    def __getitem__(self, key):
-        """
-        See __setitem__.__doc__ -- You may use my_item["key"] to get the corresponding
-        metadata value. Note however, that the key you pass must be of type str or unicode.
-
-        :type key: str or unicode
-        :param key: The key refering to the value we want to return.
-        :returns: self._metadata[key]
-        """
-        self._read_accessed = True
-        if not isinstance(key, (unicode, str)):
-            raise TypeError("key must be string type")
-        if key.startswith('__'):
-            raise KeyError(key)
-        if self._metadata is None:
-            self._metadata = self._backend._get_item_metadata(self)
-
-        return self._metadata[key]
-
-    def keys(self):
-        """
-        This method returns a list of all metadata keys of this item (i.e., a list of Strings.)
-        That allows using Python's `for mdkey in itemobj: do_something` syntax.
-
-        :returns: list of metadata keys not starting with two leading underscores
-        """
-        if self._metadata is None:
-            self._metadata = self._backend._get_item_metadata(self)
-
-        return [key for key in self._metadata if not key.startswith("__")]
-
-    def change_metadata(self):
-        """
-        @see: Backend._change_item_metadata.__doc__
-        """
-        if self._uncommitted_revision is not None:
-            raise RuntimeError(("You tried to change the metadata of the item %r but there "
-                                "are uncommitted revisions on that item. Commit first.") % (self.name))
-        if self._read_accessed:
-            raise AccessError("Cannot lock after reading metadata")
-
-        self._backend._change_item_metadata(self)
-        self._locked = True
-
-    def publish_metadata(self):
-        """
-        @see: Backend._publish_item_metadata.__doc__
-        """
-        if not self._locked:
-            raise AccessError("cannot publish without change_metadata")
-        self._backend._publish_item_metadata(self)
-        self._read_accessed = False
-        self._locked = False
-
-    def get_revision(self, revno):
-        """
-        @see: Backend._get_revision.__doc__
-        """
-        return self._backend._get_revision(self, revno)
-
-    def list_revisions(self):
-        """
-        @see: Backend._list_revisions.__doc__
-        """
-        return self._backend._list_revisions(self)
-
-    def rename(self, newname):
-        """
-        @see: Backend._rename_item.__doc__
-        """
-        if not isinstance(newname, (str, unicode)):
-            raise TypeError("Item names must have string type, not %s" % (type(newname)))
-
-        self._backend._rename_item(self, newname)
-        self._name = newname
-
-    def commit(self):
-        """
-        @see: Backend._commit_item.__doc__
-        """
-        rev = self._uncommitted_revision
-        assert rev is not None
-        rev[HASH_ALGORITHM] = unicode(rev._rev_hash.hexdigest())
-        rev[SIZE] = rev._size
-        if MTIME not in rev:
-            rev[MTIME] = int(time.time())
-        self._backend._commit_item(rev)
-        self._uncommitted_revision = None
-
-    def rollback(self):
-        """
-        @see: Backend._rollback_item.__doc__
-        """
-        self._backend._rollback_item(self._uncommitted_revision)
-        self._uncommitted_revision = None
-
-    def create_revision(self, revno):
-        """
-        @see: Backend._create_revision.__doc__
-
-        Please note that we do not require the revnos to be subsequent, but they
-        need to be monotonic. I.e., a sequence like 0, 1, 5, 9, 10 is ok, but
-        neither 0, 1, 1, 2, 3 nor 0, 1, 3, 2, 9 are.
-        This is done so as to allow functionality like unserializing a backend
-        whose item's revisions have been subject to destroy().
-        """
-        if self._locked:
-            raise RuntimeError(("You tried to create revision #%d on the item %r, but there "
-                                "is unpublished metadata on that item. Publish first.") % (revno, self.name))
-        current_revno = self.next_revno - 1
-        if current_revno >= revno:
-            raise RevisionNumberMismatchError("You cannot create a revision with revno %s. Your revno must be greater than " % revno + \
-                                              "the item's last revision, which is %s." % current_revno)
-        if self._uncommitted_revision is not None:
-            return self._uncommitted_revision
-        else:
-            self._uncommitted_revision = self._backend._create_revision(self, revno)
-            return self._uncommitted_revision
-
-    def destroy(self):
-        """
-        @see: Backend._destroy_item.__doc__
-        """
-        return self._backend._destroy_item(self)
-
-
-class Revision(object, DictMixin):
-    """
-    This class serves as superclass for StoredRevision and NewRevision.
-    An object of either subclass represents a revision of an item. An item can have
-    several revisions at a time, one being the most recent revision.
-    This is a principle that is similar to the concepts used in Version Control
-    Systems.
-
-    Each revision object has a creation timestamp in the 'timestamp' property
-    that defaults to None for newly created revisions in which case it will be
-    assigned at commit() time. It is writable for use by converter backends, but
-    care must be taken in that case to create monotone timestamps!
-    """
-    def __init__(self, item, revno):
-        """
-        Initialize the revision.
-
-        :type item: Object of class Item.
-        :param item: The item to which this revision belongs.
-        :type revno: int
-        :param revno: The unique number identifying this revision on the item.
-        :type timestamp: int
-        :param timestamp: int representing the UNIX time this revision was
-        created. (UNIX time: seconds since the epoch, i.e. 1st of January 1970, 00:00 UTC)
-        """
-        self._revno = revno
-        self._item = item
-        self._backend = item._backend
-        self._metadata = None
-
-    def _get_item(self):
-        return self._item
-
-    item = property(_get_item)
-
-    def get_revno(self):
-        """
-        Getter for the read-only revno property.
-        """
-        return self._revno
-
-    revno = property(get_revno, doc=("This property stores the revno of the revision object. "
-                                     "Only read-only access is allowed."))
-
-    @property
-    def timestamp(self):
-        """This property returns the creation timestamp of the revision"""
-        return self[MTIME]
-
-    def _load_metadata(self):
-        self._metadata = self._backend._get_revision_metadata(self)
-
-    def __getitem__(self, key):
-        """
-        @see: Item.__getitem__.__doc__
-        """
-        if not isinstance(key, (unicode, str)):
-            raise TypeError("key must be string type")
-        if key.startswith('__'):
-            raise KeyError(key)
-        if self._metadata is None:
-            self._load_metadata()
-
-        return self._metadata[key]
-
-    def keys(self):
-        """
-        @see: Item.keys.__doc__
-        """
-        if self._metadata is None:
-            self._load_metadata()
-
-        return [key for key in self._metadata if not key.startswith("__")]
-
-    def read(self, chunksize=-1):
-        """
-        @see: Backend._read_revision_data.__doc__
-        """
-        return self._backend._read_revision_data(self, chunksize)
-
-    def seek(self, position, mode=0):
-        """
-        @see: StringIO.StringIO().seek.__doc__
-        """
-        self._backend._seek_revision_data(self, position, mode)
-
-    def tell(self):
-        """
-        @see: StringIO.StringIO().tell.__doc__
-        """
-        return self._backend._tell_revision_data(self)
-
-
-class StoredRevision(Revision):
-    """
-    This is the brother of NewRevision. It allows reading data from a revision
-    that has already been stored in storage. It doesn't allow data manipulation
-    and can only be used for information retrieval.
-
-    Do NOT create instances of this class directly, but use item.get_revision or
-    one of the other methods intended for getting stored revisions.
-    """
-    def __init__(self, item, revno):
-        """
-        Initialize the StoredRevision
-        """
-        Revision.__init__(self, item, revno)
-
-    def __setitem__(self, key, value):
-        """
-        Revision metadata cannot be altered, thus, we raise an Exception.
-        """
-        raise AttributeError("Metadata of already existing revisions may not be altered.")
-
-    def __delitem__(self, key):
-        """
-        Revision metadata cannot be altered, thus, we raise an Exception.
-        """
-        raise AttributeError("Metadata of already existing revisions may not be altered.")
-
-    def destroy(self):
-        """
-        @see: Backend._destroy_revision.__doc__
-        """
-        self._backend._destroy_revision(self)
-
-
-class NewRevision(Revision):
-    """
-    This is basically the same as Revision but with mutable metadata and data properties.
-
-    Do NOT create instances of this class directly, but use item.create_revision.
-    """
-    def __init__(self, item, revno):
-        """
-        Initialize the NewRevision
-        """
-        Revision.__init__(self, item, revno)
-        self._metadata = {}
-        # these values need to be kept uptodate to that item.commit() can
-        # use them to update the metadata of the rev before committing it:
-        self._size = 0
-        self._rev_hash = hashlib.new(HASH_ALGORITHM)
-
-    def __setitem__(self, key, value):
-        """
-        Internal method used for dict-like access to the NewRevisions metadata-dict.
-        Keys starting with two underscores are reserved and cannot be used.
-
-        :type key: str or unicode
-        :param key: The keyword that is used to look up the corresponding value.
-        :type value: str, unicode, int, long, float, bool, complex or a nested tuple thereof.
-        :param value: The value that is referenced by the keyword `key` in this
-        specific items metadata-dict.
-        """
-        if not isinstance(key, (str, unicode)):
-            raise TypeError("Key must be string type")
-        if key.startswith('__'):
-            raise TypeError("Key must not begin with two underscores")
-        check_value_type_is_valid(value)
-
-        self._metadata[key] = value
-
-    def __delitem__(self, key):
-        if key.startswith('__'):
-            raise KeyError(key)
-
-        del self._metadata[key]
-
-    def write(self, data):
-        """
-        @see: Backend._write_revision_data.__doc__
-        """
-        self._size += len(data)
-        self._rev_hash.update(data)
-        self._backend._write_revision_data(self, data)
-
-
-# Little helper function:
-def check_value_type_is_valid(value):
-    """
-    For metadata-values, we allow only immutable types, namely:
-    str, unicode, bool, int, long, float, complex and tuple.
-    Since tuples can contain other types, we need to check the types recursively.
-
-    :type value: str, unicode, int, long, float, complex, tuple
-    :param value: A value of which we want to know if it is a valid metadata value.
-    :returns: bool
-    """
-    accepted = (bool, str, unicode, int, long, float, complex)
-    if isinstance(value, accepted):
-        return True
-    elif isinstance(value, tuple):
-        for element in value:
-            if not check_value_type_is_valid(element):
-                raise TypeError("Value must be one of %s or a nested tuple thereof. Not %r" % (accepted, type(value)))
-        else:
-            return True
-

File MoinMoin/storage/_tests/__init__.py

-# Copyright: 2011 The MoinMoin development team
-# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
-
-"""
-    MoinMoin - Storage test package.
-"""
-

File MoinMoin/storage/_tests/test_backends.py

-# -*- coding: utf-8 -*-
-# Copyright: 2008 MoinMoin:PawelPacana
-# Copyright: 2008 MoinMoin:ChristopherDenter
-# Copyright: 2008 MoinMoin:JohannesBerg
-# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
-
-"""
-    MoinMoin - Backend tests
-
-    This module provides class for testing backends. This class tries
-    to cover sane backend usage examples.
-
-    This class should be inherited by descendant backend test classes.
-    Add tests general for all backends here. Your backend-specific tests
-    put in class inherited from this one.
-"""
-
-
-import pytest, re, time
-
-from flask import g as flaskg
-
-from MoinMoin.storage import Item, NewRevision
-from MoinMoin.storage.backends import memory
-from MoinMoin.storage.error import NoSuchItemError, ItemAlreadyExistsError, NoSuchRevisionError, RevisionAlreadyExistsError
-from MoinMoin.config import SIZE
-
-item_names = (u"quite_normal",
-              u"äöüßłóąćółąńśćżź",
-              u"with space",
-              u"name#special(characters?.\,",
-              u"very_long_name_" * 100 + u"ending_1",
-              u"very_long_name_" * 100 + u"ending_2", )
-
-invalid_names = (42, {}, (1, ), [1], )
-
-class BackendTest(object):
-    """
-    Generic class for backend tests.
-
-    Creates a new backend for each test so they can assume to be
-    sandboxed.
-    """
-
-    valid_names = item_names
-    invalid_names = invalid_names
-
-    def setup_method(self, method):
-        self.backend = self.create_backend()
-
-    def teardown_method(self, method):
-        self.kill_backend()
-        self.backend = None
-
-    def create_rev_item_helper(self, name):
-        item = self.backend.create_item(name)
-        item.create_revision(0)
-        item.commit()
-        return item
-
-    def create_meta_item_helper(self, name):
-        item = self.backend.create_item(name)
-        item.change_metadata()
-        item.publish_metadata()
-        return item
-
-    def get_item_check(self, name):
-        item = self.backend.get_item(name)
-        assert item.name == name
-
-    def rename_item_check(self, old_name, new_name):
-        item = self.backend.get_item(old_name)
-        item.rename(new_name)
-        assert item.name == new_name
-        assert self.backend.has_item(new_name)
-        assert not self.backend.has_item(old_name)
-
-    def test_create_get_rename_get_rev_item(self):
-        def create_rev_item(name):
-            item = self.backend.create_item(name)
-            assert item.name == name
-            item.create_revision(0)
-            item.commit()
-            assert self.backend.has_item(name)
-
-        for num, item_name in enumerate(self.valid_names):
-            yield create_rev_item, item_name
-            yield self.get_item_check, item_name
-            new_name = u"renamed_revitem_%d" % num
-            yield self.rename_item_check, item_name, new_name
-            yield self.get_item_check, new_name
-
-    def test_create_get_rename_get_meta_item(self):
-        def create_meta_item(name):
-            item = self.backend.create_item(name)
-            assert item.name == name
-            item.change_metadata()
-            item.publish_metadata()
-            assert self.backend.has_item(name)
-
-        for num, item_name in enumerate(self.valid_names):
-            yield create_meta_item, item_name
-            yield self.get_item_check, item_name
-            new_name = u"renamed_revitem_%d" % num
-            yield self.rename_item_check, item_name, new_name
-            yield self.get_item_check, new_name
-
-    def test_item_rename_to_existing(self):
-        item1 = self.create_rev_item_helper(u"fresh_item")
-        item2 = self.create_rev_item_helper(u"try to rename")
-        pytest.raises(ItemAlreadyExistsError, item1.rename, item2.name)
-
-    def rename_item_invalid_name(self, name, newname):
-        item = self.backend.create_item(name)
-        pytest.raises(TypeError, item.rename, newname)
-
-    def test_item_rename_to_invalid(self):
-        for num, invalid_name in enumerate(self.invalid_names):
-            yield self.rename_item_invalid_name, u"item_%s" % num, invalid_name
-
-    def test_item_rename_threesome(self):
-        item1 = self.create_rev_item_helper(u"item1")
-        item2 = self.create_rev_item_helper(u"item2")
-        item1.create_revision(1)
-        item1.commit()
-        item2.rename(u"item3")
-        item1.rename(u"item2")
-        assert len(item1.list_revisions()) == 2
-
-    def create_item_invalid_name(self, name):
-        pytest.raises(TypeError, self.backend.create_item, name)
-
-    def test_create_item_wrong_itemname(self):
-        for item_name in self.invalid_names:
-            yield self.create_item_invalid_name, item_name
-
-    def test_create_order(self):
-        item1 = self.backend.create_item(u'1')
-        item2 = self.backend.create_item(u'2')
-        revision1 = item1.create_revision(0)
-        revision2 = item2.create_revision(0)
-        revision1.write('1')
-        revision2.write('2')
-        item2.commit()
-        item1.commit()
-        item1 = self.backend.get_item(u'1')
-        item2 = self.backend.get_item(u'2')
-        revision1 = item1.get_revision(0)
-        revision2 = item2.get_revision(0)
-        assert revision1.read() == '1'
-        assert revision2.read() == '2'
-
-    def test_create_rev_item_again(self):
-        self.create_rev_item_helper(u"item1")
-        pytest.raises(ItemAlreadyExistsError, self.backend.create_item, u"item1")
-
-    def test_create_meta_item_again(self):
-        self.create_meta_item_helper(u"item2")
-        pytest.raises(ItemAlreadyExistsError, self.backend.create_item, u"item2")
-
-    def test_get_item_that_doesnt_exist(self):
-        pytest.raises(NoSuchItemError, self.backend.get_item, u"i_do_not_exist")
-
-    def test_has_item(self):
-        self.create_rev_item_helper(u"versioned")
-        self.create_meta_item_helper(u"unversioned")
-        assert self.backend.has_item(u"versioned")
-        assert self.backend.has_item(u"unversioned")
-
-    def test_has_item_that_doesnt_exist(self):
-        assert not self.backend.has_item(u"i_do_not_exist")
-
-    def test_iteritems_1(self):
-        for num in range(10, 20):
-            self.create_rev_item_helper(u"item_" + str(num).zfill(2))
-        for num in range(10):
-            self.create_meta_item_helper(u"item_" + str(num).zfill(2))
-        itemlist = sorted([item.name for item in self.backend.iteritems()])
-        for num, itemname in enumerate(itemlist):
-            assert itemname == u"item_" + str(num).zfill(2)
-        assert len(itemlist) == 20
-
-    def test_iteritems_2(self):
-        self.create_rev_item_helper(u'abcdefghijklmn')
-        count = 0
-        for item in self.backend.iteritems():
-            count += 1
-        assert count > 0
-
-    def test_iteritems_3(self):
-        self.create_rev_item_helper(u"without_meta")
-        self.create_rev_item_helper(u"with_meta")
-        item = self.backend.get_item(u"with_meta")
-        item.change_metadata()
-        item[u"meta"] = u"data"
-        item.publish_metadata()
-        itemlist = [item for item in self.backend.iteritems()]
-        assert len(itemlist) == 2
-
-    def test_existing_item_create_revision(self):
-        self.create_rev_item_helper(u"existing")
-        item = self.backend.get_item(u"existing")
-        old_rev = item.get_revision(-1)
-        rev = item.create_revision(old_rev.revno + 1)
-        item.rollback()
-        rev = item.get_revision(-1)
-        old_keys = old_rev.keys()
-        new_keys = rev.keys()
-        old_keys.sort()
-        new_keys.sort()
-        assert old_keys == new_keys
-        for key, value in old_rev.iteritems():
-            assert rev[key] == value
-        assert old_rev.read() == rev.read()
-
-    def test_new_item_create_revision(self):
-        item = self.backend.create_item(u'internal')
-        rev = item.create_revision(0)
-        item.rollback()
-        assert not self.backend.has_item(item.name)
-
-    def test_item_commit_revision(self):
-        item = self.backend.create_item(u"item#11")
-        rev = item.create_revision(0)
-        rev.write("python rocks")
-        item.commit()
-        rev = item.get_revision(0)
-        assert rev.read() == "python rocks"
-
-    def test_item_writing_data_multiple_times(self):
-        item = self.backend.create_item(u"multiple")
-        rev = item.create_revision(0)
-        rev.write("Alle ")
-        rev.write("meine ")
-        rev.write("Entchen")
-        item.commit()
-        rev = item.get_revision(0)
-        assert rev.read() == "Alle meine Entchen"
-
-    def test_item_write_seek_read(self):
-        item = self.backend.create_item(u"write_seek_read")
-        rev = item.create_revision(0)
-        write_data = "some data"
-        rev.write(write_data)
-        rev.seek(0)
-        read_data = rev.read()
-        assert read_data == write_data
-        item.commit()
-        rev = item.get_revision(0)
-        assert rev.read() == write_data
-
-    def test_item_seek_tell_read(self):
-        item = self.backend.create_item(u"write_seek_read")
-        rev = item.create_revision(0)
-        write_data = "0123456789"
-        rev.write(write_data)
-        rev.seek(0)
-        assert rev.tell() == 0
-        read_data = rev.read()
-        assert read_data == write_data
-        rev.seek(4)
-        assert rev.tell() == 4
-        read_data = rev.read()
-        assert read_data == write_data[4:]
-        item.commit()
-        rev = item.get_revision(0)
-        rev.seek(0)
-        assert rev.tell() == 0
-        read_data = rev.read()
-        assert read_data == write_data
-        rev.seek(4)
-        assert rev.tell() == 4
-        read_data = rev.read()
-        assert read_data == write_data[4:]
-
-    def test_item_reading_chunks(self):
-        item = self.backend.create_item(u"slices")
-        rev = item.create_revision(0)
-        rev.write("Alle meine Entchen")
-        item.commit()
-        rev = item.get_revision(0)
-        chunk = rev.read(1)
-        data = ""
-        while chunk != "":
-            data += chunk
-            chunk = rev.read(1)
-        assert data == "Alle meine Entchen"
-
-    def test_item_reading_negative_chunk(self):
-        item = self.backend.create_item(u"negative_chunk")
-        rev = item.create_revision(0)
-        rev.write("Alle meine Entchen" * 10)
-        item.commit()
-        rev = item.get_revision(0)
-        assert rev.read(-1) == "Alle meine Entchen" * 10
-        rev.seek(0)
-        assert rev.read(-123) == "Alle meine Entchen" * 10
-
-    def test_seek_and_tell(self):
-        item = self.backend.create_item(u"seek&tell")
-        rev = item.create_revision(0)
-        data = "wilhelm tell seekfried what time it is"
-        rev.write(data)
-        item.commit()
-
-        rev = item.get_revision(0)
-        offset = 5
-
-        # absolute
-        rev.seek(offset)
-        assert rev.tell() == offset
-        assert rev.read() == data[offset:]
-
-        # relative
-        rev.seek(offset)
-        rev.seek(offset, 1)
-        assert rev.tell() == 2 * offset
-        assert rev.read() == data[2*offset:]
-
-        # relative to EOF
-        rev.seek(-offset, 2)
-        assert rev.tell() == len(data) - offset
-        assert rev.read() == data[-offset:]
-
-    def test_item_get_revision(self):
-        item = self.backend.create_item(u"item#12")
-        rev = item.create_revision(0)
-        rev.write("jefferson airplane rocks")
-        item.commit()
-        another_rev = item.get_revision(0)
-        assert another_rev.read() == "jefferson airplane rocks"
-
-    def test_item_next_revno(self):
-        item = self.backend.create_item(u"next_revno")
-        assert item.next_revno == 0
-        item.create_revision(item.next_revno).write("foo")
-        item.commit()
-        assert item.next_revno == 1
-
-    def test_item_list_revisions_with_revmeta_changes(self):
-        item = self.backend.create_item(u"item_13")
-        for revno in range(0, 10):
-            rev = item.create_revision(revno)
-            rev[u"revno"] = u"%s" % revno
-            item.commit()
-        assert item.list_revisions() == range(0, 10)
-
-    def test_item_list_revisions_with_revdata_changes(self):
-        item = self.backend.create_item(u"item_13")
-        for revno in range(0, 10):
-            rev = item.create_revision(revno)
-            rev.write("%s" % revno)
-            item.commit()
-        assert item.list_revisions() == range(0, 10)
-
-    def test_item_list_revisions_without_changes(self):
-        item = self.backend.create_item(u"item_13")
-        for revno in range(0, 10):
-            item.create_revision(revno)
-            item.commit()
-        assert item.list_revisions() == range(0, 10)
-
-    def test_item_list_revisions_equality(self):
-        item = self.backend.create_item(u"new_item_15")
-        revs_before = item.list_revisions()
-        rev = item.create_revision(0)
-        assert item.list_revisions() == revs_before
-        item.rollback()
-
-    def test_item_list_revisions_equality_nonempty_revlist(self):
-        item = self.backend.create_item(u"new_item_16")
-        rev = item.create_revision(0)
-        rev.write("something interesting")
-        item.commit()
-        revs_before = item.list_revisions()
-        rev2 = item.create_revision(1)
-        assert item.list_revisions() == revs_before
-        item.rollback()
-
-    def test_item_list_revisions_without_committing(self):
-        item = self.backend.create_item(u"new_item_14")
-        assert item.list_revisions() == []
-
-    def test_mixed_commit_metadata1(self):
-        item = self.backend.create_item(u'mixed1')
-        item.create_revision(0)
-        pytest.raises(RuntimeError, item.change_metadata)
-        item.rollback()
-
-    def test_mixed_commit_metadata2(self):
-        item = self.backend.create_item(u'mixed2')
-        item.change_metadata()
-        pytest.raises(RuntimeError, item.create_revision, 0)
-
-    def test_item_metadata_change_and_publish(self):
-        item = self.backend.create_item(u"test item metadata change")
-        item.change_metadata()
-        item[u"creator"] = u"Vincent van Gogh"
-        item.publish_metadata()
-        item2 = self.backend.get_item(u"test item metadata change")
-        assert item2[u"creator"] == u"Vincent van Gogh"
-
-    def test_item_metadata_invalid_change(self):
-        item = self.backend.create_item(u"test item metadata invalid change")
-        try:
-            item[u"this should"] = "FAIL!"
-            assert False
-        except AttributeError:
-            pass
-
-    def test_item_metadata_without_publish(self):
-        item = self.backend.create_item(u"test item metadata invalid change")
-        item.change_metadata()
-        item[u"change but"] = u"don't publish"
-        pytest.raises(NoSuchItemError, self.backend.get_item, "test item metadata invalid change")
-
-    def test_item_create_existing_mixed_1(self):
-        item1 = self.backend.create_item(u'existing now 0')
-        item1.change_metadata()
-        item2 = self.backend.create_item(u'existing now 0')
-        item1.publish_metadata()
-        item2.create_revision(0)
-        pytest.raises(ItemAlreadyExistsError, item2.commit)
-
-    def test_item_create_existing_mixed_2(self):
-        item1 = self.backend.create_item(u'existing now 0')
-        item1.change_metadata()
-        item2 = self.backend.create_item(u'existing now 0')
-        item2.create_revision(0)
-        item2.commit()
-        pytest.raises(ItemAlreadyExistsError, item1.publish_metadata)
-
-    def test_item_multiple_change_metadata_after_create(self):
-        name = u"foo"
-        item1 = self.backend.create_item(name)
-        item2 = self.backend.create_item(name)
-        item1.change_metadata()
-        item2.change_metadata()
-        item1[u"a"] = u"a"
-        item2[u"a"] = u"b"
-        item1.publish_metadata()
-        pytest.raises(ItemAlreadyExistsError, item2.publish_metadata)
-        item = self.backend.get_item(name)
-        assert item[u"a"] == u"a"
-
-    def test_existing_item_change_metadata(self):
-        self.create_meta_item_helper(u"existing now 2")
-        item = self.backend.get_item(u'existing now 2')
-        item.change_metadata()
-        item[u'asdf'] = u'b'
-        item.publish_metadata()
-        item = self.backend.get_item(u'existing now 2')
-        assert item[u'asdf'] == u'b'
-
-    def test_metadata(self):
-        self.create_rev_item_helper(u'no metadata')
-        item = self.backend.get_item(u'no metadata')
-        pytest.raises(KeyError, item.__getitem__, u'asdf')
-
-    def test_revision(self):
-        self.create_meta_item_helper(u'no revision')
-        item = self.backend.get_item(u'no revision')
-        pytest.raises(NoSuchRevisionError, item.get_revision, -1)
-
-    def test_create_revision_change_meta(self):
-        item = self.backend.create_item(u"double")
-        rev = item.create_revision(0)
-        rev[u"revno"] = u"0"
-        item.commit()
-        item.change_metadata()
-        item[u"meta"] = u"data"
-        item.publish_metadata()
-        item = self.backend.get_item(u"double")
-        assert item[u"meta"] == u"data"
-        rev = item.get_revision(0)
-        assert rev[u"revno"] == u"0"
-
-    def test_create_revision_change_empty_meta(self):
-        item = self.backend.create_item(u"double")
-        rev = item.create_revision(0)
-        rev[u"revno"] = u"0"
-        item.commit()
-        item.change_metadata()
-        item.publish_metadata()
-        item = self.backend.get_item(u"double")
-        rev = item.get_revision(0)
-        assert rev[u"revno"] == u"0"
-
-    def test_change_meta_create_revision(self):
-        item = self.backend.create_item(u"double")
-        item.change_metadata()
-        item[u"meta"] = u"data"
-        item.publish_metadata()
-        rev = item.create_revision(0)
-        rev[u"revno"] = u"0"
-        item.commit()
-        item = self.backend.get_item(u"double")
-        assert item[u"meta"] == u"data"
-        rev = item.get_revision(0)
-        assert rev[u"revno"] == u"0"
-
-    def test_meta_after_rename(self):
-        item = self.backend.create_item(u"re")
-        item.change_metadata()
-        item[u"previous_name"] = u"re"
-        item.publish_metadata()
-        item.rename(u"er")
-        assert item[u"previous_name"] == u"re"
-
-    def test_long_names_back_and_forth(self):
-        item = self.backend.create_item(u"long_name_" * 100 + u"with_happy_end")
-        item.create_revision(0)
-        item.commit()
-        assert self.backend.has_item(u"long_name_" * 100 + u"with_happy_end")
-        item = self.backend.iteritems().next()
-        assert item.name == u"long_name_" * 100 + u"with_happy_end"
-
-    def test_revisions_after_rename(self):
-        item = self.backend.create_item(u"first one")
-        for revno in xrange(10):
-            rev = item.create_revision(revno)
-            rev[u"revno"] = unicode(revno)
-            item.commit()
-        assert item.list_revisions() == range(10)
-        item.rename(u"second one")
-        assert not self.backend.has_item(u"first one")
-        assert self.backend.has_item(u"second one")
-        item1 = self.backend.create_item(u"first_one")
-        item1.create_revision(0)
-        item1.commit()
-        assert len(item1.list_revisions()) == 1
-        item2 = self.backend.get_item(u"second one")
-        assert item2.list_revisions() == range(10)
-        for revno in xrange(10):
-            rev = item2.get_revision(revno)
-            assert rev[u"revno"] == unicode(revno)
-
-    def test_concurrent_create_revision(self):
-        self.create_rev_item_helper(u"concurrent")
-        item1 = self.backend.get_item(u"concurrent")
-        item2 = self.backend.get_item(u"concurrent")
-        item1.create_revision(1)
-        item2.create_revision(1)
-        item1.commit()
-        pytest.raises(RevisionAlreadyExistsError, item2.commit)
-
-    def test_timestamp(self):
-        tnow = int(time.time())
-        item = self.backend.create_item(u'ts1')
-        rev = item.create_revision(0)
-        item.commit()
-        item = self.backend.get_item(u'ts1')
-        ts = item.get_revision(0).timestamp
-        assert tnow <= ts <= ts + 60
-
-    def test_size(self):
-        item = self.backend.create_item(u'size1')
-        rev = item.create_revision(0)
-        rev.write('asdf')
-        rev.write('asdf')
-        item.commit()
-        rev = item.get_revision(0)
-        assert rev[SIZE] == 8
-
-    def test_size_2(self):
-        item = self.backend.create_item(u'size2')
-        rev0 = item.create_revision(0)
-        data0 = 'asdf'
-        rev0.write(data0)
-        item.commit()
-        rev1 = item.create_revision(1)
-        item.commit()
-        rev1 = item.get_revision(1)
-        assert rev1[SIZE] == 0
-        rev0 = item.get_revision(0)
-        assert rev0[SIZE] == len(data0)
-
-    def test_various_revision_metadata_values(self):
-        def test_value(value, no):
-            item = self.backend.create_item(u'valid_values_%d' % no)
-            rev = item.create_revision(0)
-            key = u"key%d" % no
-            rev[key] = value
-            item.commit()
-            rev = item.get_revision(0)
-            assert rev[key] == value
-
-        for no, value in enumerate(('string', 13, 42L, 3.14, 23+0j,
-                                       ('1', 1, 1L, 1+0j, (1, ), ), u'ąłć', (u'ó', u'żźć'), )):
-            yield test_value, value, no
-
-    def test_destroy_item(self):
-        itemname = u"I will be completely destroyed"
-        rev_data = "I will be completely destroyed, too, hopefully"
-        item = self.backend.create_item(itemname)
-        rev = item.create_revision(0)
-        rev.write(rev_data)
-        item.commit()
-
-        item.destroy()
-        assert not self.backend.has_item(itemname)
-        item_names = [item.name for item in self.backend.iteritems()]
-        assert not itemname in item_names
-
-    def test_destroy_revision(self):
-        itemname = u"I will see my children die"        # removed the smiley ':-(' temporarily as it slows the test in addition with a failure
-        rev_data = "I will die!"
-        persistent_rev = "I will see my sibling die :-("
-        item = self.backend.create_item(itemname)
-        rev = item.create_revision(0)
-        rev.write(rev_data)
-        item.commit()
-        rev = item.create_revision(1)
-        rev.write(persistent_rev)
-        item.commit()
-        assert item.list_revisions() == range(2)
-
-        rev = item.get_revision(0)
-        rev.destroy()
-        assert item.list_revisions() == [1]
-        assert self.backend.has_item(itemname)
-        assert item.get_revision(-1).read() == persistent_rev
-
-        third = "3rd revision"
-        rev = item.create_revision(2)
-        rev.write(third)
-        item.commit()
-        rev = item.get_revision(2)
-        assert item.get_revision(-1).read() == third
-        assert len(item.list_revisions()) == 2
-        rev.destroy()
-        assert len(item.list_revisions()) == 1
-        last = item.get_revision(-1)
-        assert last.revno == 1
-        last_data = last.read()
-        assert last_data != third
-        assert last_data == persistent_rev
-
-    def test_clone_backend(self):
-        src = flaskg.storage
-        dst = memory.MemoryBackend()
-
-        dollys_name = u"Dolly The Sheep"
-        item = src.create_item(dollys_name)
-        rev = item.create_revision(0)
-        rev.write("maeh")
-        rev[u'origin'] = u'reagenzglas'
-        item.commit()
-
-        brothers_name = u"Dolly's brother"
-        item = src.create_item(brothers_name)
-        item.change_metadata()
-        item[u'no revisions'] = True
-        item.publish_metadata()
-
-        dst.clone(src, verbose=False)
-
-        assert len(list(dst.iteritems())) == 2
-        assert dst.has_item(dollys_name)
-        rev = dst.get_item(dollys_name).get_revision(0)
-        data = rev.read()
-        assert data == "maeh"
-        meta = dict(rev.iteritems())
-        assert u'origin' in meta
-        assert meta[u'origin'] == u'reagenzglas'
-
-        assert dst.has_item(brothers_name)
-        item = dst.get_item(brothers_name)
-        meta = dict(item.iteritems())
-        assert u'no revisions' in meta
-        assert meta[u'no revisions'] is True
-
-    def test_iteritems_item_names_after_rename(self):
-        item = self.backend.create_item(u'first')
-        item.create_revision(0)
-        item.commit()
-        item.rename(u'second')
-        item.create_revision(1)
-        item.commit()
-        # iteritems provides actual name
-        items = [item for item in self.backend.iteritems()]
-        assert len(items) == 1
-        assert items[0].name == u'second'
-        rev0 = items[0].get_revision(0)
-        assert rev0.item.name == u'second'
-        rev1 = items[0].get_revision(1)
-        assert rev1.item.name == u'second'
-
-    def test_iteritems_after_destroy(self):
-        item = self.backend.create_item(u'first')
-        item.create_revision(0)
-        item.commit()
-        item.create_revision(1)
-        item.commit()
-        assert len([item for item in self.backend.iteritems()]) == 1
-        rev = item.get_revision(-1)
-        rev.destroy()
-        assert len([item for item in self.backend.iteritems()]) == 1
-        item.destroy()
-        assert len([item for item in self.backend.iteritems()]) == 0
-

File MoinMoin/storage/_tests/test_backends_flatfile.py

-# Copyright: 2009 MoinMoin:ThomasWaldmann
-# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
-
-"""
-    MoinMoin - Test - FlatFileBackend
-"""
-
-
-import pytest
-
-pytest.skip("BackendTest base class tests quite some stuff that this very simple backend does not provide")
-# e.g.: revisioning, extremely long item names, metadata support
-# TODO: either fix base class so that it is more useful even to test simple backends,
-#       or implement some specific, more simple tests here.
-
-import tempfile, shutil
-
-from MoinMoin.storage._tests.test_backends import BackendTest
-from MoinMoin.storage.backends.flatfile import FlatFileBackend
-
-class TestFlatFileBackend(BackendTest):
-
-    def create_backend(self):
-        self.tempdir = tempfile.mkdtemp('', 'moin-')
-        return FlatFileBackend(self.tempdir)
-
-    def kill_backend(self):
-        shutil.rmtree(self.tempdir)
-

File MoinMoin/storage/_tests/test_backends_fs.py

-# Copyright: 2008 MoinMoin:JohannesBerg
-# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
-
-"""
-    MoinMoin - Test - FSBackend
-"""
-
-
-import py, os, tempfile, shutil
-
-from MoinMoin.storage._tests.test_backends import BackendTest
-from MoinMoin.storage.backends.fs import FSBackend
-
-class TestFSBackend(BackendTest):
-
-    def create_backend(self):
-        self.tempdir = tempfile.mkdtemp('', 'moin-')
-        return FSBackend(self.tempdir)
-
-    def kill_backend(self):
-        try:
-            for root, dirs, files in os.walk(self.tempdir):
-                for d in dirs:
-                    assert not d.endswith('.lock')
-                for f in files:
-                    assert not f.endswith('.lock')
-                    assert not f.startswith('tmp-')
-        finally:
-            shutil.rmtree(self.tempdir)
-
-    def test_large(self):
-        i = self.backend.create_item(u'large')
-        r = i.create_revision(0)
-        r[u'0'] = u'x' * 100
-        r[u'1'] = u'y' * 200
-        r[u'2'] = u'z' * 300
-        for x in xrange(1000):
-            r.write('lalala! ' * 10)
-        i.commit()
-
-        i = self.backend.get_item(u'large')
-        r = i.get_revision(0)
-        assert r[u'0'] == u'x' * 100
-        assert r[u'1'] == u'y' * 200
-        assert r[u'2'] == u'z' * 300
-        for x in xrange(1000):
-            assert r.read(8 * 10) == 'lalala! ' * 10
-        assert r.read() == ''
-
-    def test_all_unlocked(self):
-        i1 = self.backend.create_item(u'existing now 1')
-        i1.create_revision(0)
-        i1.commit()
-        i2 = self.backend.get_item(u'existing now 1')
-        i2.change_metadata()
-        # if we leave out the latter line, it fails
-        i2.publish_metadata()
-

File MoinMoin/storage/_tests/test_backends_fs19.py

-# -*- coding: utf-8 -*-
-# Copyright: 2008-2010 MoinMoin:ThomasWaldmann
-# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
-
-"""
-    MoinMoin - fs19 read-only backend tests
-"""
-
-
-import os, re, tempfile, shutil
-
-import pytest
-
-from flask import current_app as app
-
-from MoinMoin.config import CONTENTTYPE, TAGS
-from MoinMoin.storage import Item
-from MoinMoin.storage.backends._fsutils import quoteWikinameFS, unquoteWikiname
-from MoinMoin.storage.backends.fs19 import FSPageBackend, regenerate_acl, process_categories
-from MoinMoin.storage.error import NoSuchItemError, NoSuchRevisionError
-
-item_data = "Foo Bar"
-item_name = "test_page"
-item_mtime = 12345678
-item_comment = "saved test item"
-item_revisions = 2
-
-deleted_item_acl = "All:"
-deleted_item_data = "#acl %s\r\nFoo bar" % deleted_item_acl
-deleted_item_name = "deleted_page"
-
-attachment_name = u"test.txt"
-attachment_data = "attachment"
-attachment_mtime1 = 12340000
-attachment_mtime2 = 12345000
-attachment_comment = "saved test attachment"
-
-logentry = lambda *items: "\t".join(items)
-item_editlog = "\r\n".join([
-    logentry(str(item_mtime * 1000000), '00000001', 'SAVE', item_name, '', '', '', '', item_comment),
-    logentry(str(attachment_mtime1 * 1000000), '99999999', 'ATTNEW', item_name, '', '', '', attachment_name, attachment_comment),
-    logentry(str(item_mtime * 1000000 + 1), '00000002', 'SAVE', item_name, '', '', '', '', item_comment),
-    logentry(str(attachment_mtime2 * 1000000), '99999999', 'ATTNEW', item_name, '', '', '', attachment_name, attachment_comment),
-])
-
-deleted_item_editlog = "\r\n".join([
-    logentry(str(item_mtime * 1000000), '00000001', 'SAVE', item_name, '', '', '', '', item_comment),
-    logentry(str(item_mtime * 1000000 + 1), '00000002', 'SAVE/DELETE', item_name, '', '', '', '', item_comment),
-])
-
-items = [# name, rev, data, logline, attachments
-         (item_name, 1, item_data, item_editlog, [attachment_name]),
-         (item_name, 2, item_data, item_editlog, []),
-         (u"äöüßłó ąćółąńśćżź", 1, item_data, '', []),
-         (ur"name#special(characters?.\,", 1, item_data, '', []),
-         (deleted_item_name, 1, deleted_item_data, '', [attachment_name]),
-         (deleted_item_name, 2, '', '', []), # no rev 2 data, no edit-log
-        ]
-
-class TestFS19Backend(object):
-    """
-    MoinMoin - fs19 read-only backend tests
-    """
-
-    def setup_method(self, method):
-        # create backend
-        self.tempdir = d = tempfile.mkdtemp('', 'moin-')
-        self.backend = FSPageBackend(self.tempdir, self.tempdir)
-        # populate it manually because the backend is just read-only
-        join = os.path.join
-        for name, revno, revdata, logdata, attachments in items:
-            pagedir = join(d, 'pages', quoteWikinameFS(name))
-            try:
-                os.makedirs(join(pagedir, 'revisions'))
-                os.makedirs(join(pagedir, 'attachments'))
-            except:
-                pass
-            f = file(join(pagedir, 'current'), 'w')
-            f.write('%08d' % revno)
-            f.close()
-            if revdata:
-                f = file(join(pagedir, 'revisions', '%08d' % revno), 'w')
-                f.write(revdata)
-                f.close()
-            if logdata:
-                f = file(join(pagedir, 'edit-log'), 'a')
-                f.write(logdata)
-                f.close()
-            for attachment in attachments:
-                f = file(join(pagedir, 'attachments', attachment.encode('utf-8')), 'w')
-                f.write(attachment_data)
-                f.close()
-
-    def teardown_method(self, method):
-        # remove backend data
-        shutil.rmtree(self.tempdir)
-        self.backend = None
-
-    def test_get_item_that_doesnt_exist(self):
-        pytest.raises(NoSuchItemError, self.backend.get_item, "i_do_not_exist")
-        pytest.raises(NoSuchItemError, self.backend.get_item, item_name + "/not_exist.txt")
-
-    def test_has_item_that_doesnt_exist(self):
-        assert not self.backend.has_item("i_do_not_exist")
-        assert not self.backend.has_item(item_name + "/not_exist.txt")
-
-    def test_get_item_that_exists(self):
-        for itemdata in items:
-            name = itemdata[0]
-            item = self.backend.get_item(name)
-            assert isinstance(item, Item)
-            assert item.name == name
-
-    def test_get_item_attachment(self):
-        name = item_name + '/' + attachment_name
-        item = self.backend.get_item(name)
-        assert isinstance(item, Item)
-        assert item.name == name
-
-    def test_has_item(self):
-        for itemdata in items:
-            name = itemdata[0]
-            exists = self.backend.has_item(name)
-            assert exists
-
-    def test_iteritems(self):
-        have_items = set([item.name for item in self.backend.iteritems()])
-        expected_items = set()
-        for itemdata in items:
-            itemname = itemdata[0]
-            attachments = itemdata[4]
-            expected_items |= set([itemname] + ['%s/%s' % (itemname, att) for att in attachments])
-        assert have_items == expected_items
-
-    def test_rev_reading_chunks(self):
-        item = self.backend.get_item(item_name)
-        rev = item.get_revision(0)
-        chunk = rev.read(1)
-        data = ""
-        while chunk != "":
-            data += chunk
-            chunk = rev.read(1)
-        assert data == item_data
-
-    def test_rev_reading_attachment(self):
-        name = item_name + '/' + attachment_name
-        item = self.backend.get_item(name)
-        rev = item.get_revision(0)
-        data = rev.read()
-        assert data == attachment_data
-
-    def test_deleted_rev_reading(self):
-        item = self.backend.get_item(deleted_item_name)
-        rev = item.get_revision(0)
-        data = rev.read()
-        assert data != ""
-        rev = item.get_revision(1)
-        data = rev.read()
-        assert data == ""
-
-    def test_metadata_that_doesnt_exist(self):
-        item = self.backend.get_item(item_name)
-        pytest.raises(KeyError, item.__getitem__, 'asdf')
-
-    def test_metadata_mtime(self):
-        item = self.backend.get_item(item_name)
-        rev = item.get_revision(0)
-        assert rev.timestamp == item_mtime
-
-    def test_metadata_mtime_attachment(self):
-        name = item_name + '/' + attachment_name
-        item = self.backend.get_item(name)
-        rev = item.get_revision(0)
-        rev_timestamp = rev.timestamp
-        assert rev_timestamp == attachment_mtime2
-
-    def test_item_revision_count(self):
-        item = self.backend.get_item(item_name)
-        revs = item.list_revisions()
-        assert revs == range(item_revisions)
-
-    def test_revision_that_doesnt_exist(self):
-        item = self.backend.get_item(item_name)
-        pytest.raises(NoSuchRevisionError, item.get_revision, 42)
-
-    def test_revision_attachment_that_doesnt_exist(self):
-        name = item_name + '/' + attachment_name
-        item = self.backend.get_item(name)
-        pytest.raises(NoSuchRevisionError, item.get_revision, 1) # attachment only has rev 0
-
-    def test_revision_attachment_acl(self):
-        name = deleted_item_name + '/' + attachment_name
-        item = self.backend.get_item(name)
-        rev = item.get_revision(0)
-        assert rev['acl'] == deleted_item_acl
-
-
-class TestAclRegeneration(object):
-    """
-    test ACL regeneration
-
-    We need to regenerate ACLs for moin 1.9 (fs19) -> 2.0 migration, because we need to cleanly
-    remove revert and delete rights.
-    """
-    def testAclRegeneration(self):
-        tests = [
-            (u'', u''),
-            (u'All:', u'All:'), # no rights, no change
-            (u'All:read', u'All:read'), # single entry, no change
-            (u'All:read,write,revert', u'All:read,write'), # single entry, remove 'revert'
-            (u'All:read,write,delete', u'All:read,write'), # single entry, remove 'delete'
-            (u'BadGuy: Default', u'BadGuy: Default'), # multiple entries, do not expand Default
-            (u'Known:read,delete,write,revert All:read',
-             u'Known:read,write All:read'), # multiple entries, remove 'delete'/'revert'
-            (u'Joe Doe,Jane Doe:delete,read,write All:',
-             u'Joe Doe,Jane Doe:read,write All:'), # multiple entries, blanks in names, remove 'delete'
-        ]
-        acl_rights_valid = app.cfg.acl_rights_contents
-        for acl, expected in tests:
-            result = regenerate_acl(acl, acl_rights_valid)
-            assert result == expected
-
-
-class TestTagsGeneration(object):
-    """
-    test tags generation from categories
-    """
-    def testTagsGeneration(self):
-        tests = [
-            (u'', u'', []),
-            (u"""1\r
-----\r
-""",
-             u"""1\r
-""",
-             []),
-            (u"""2\r
-----\r