--- /dev/null
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+import os
+import tempfile
+from nose.tools import *
+
+import vstorage
+
+
+def clear_directory(top):
+ for root, dirs, files in os.walk(top, topdown=False):
+ for name in files:
+ os.remove(os.path.join(root, name))
+ for name in dirs:
+ os.rmdir(os.path.join(root, name))
+ try:
+ os.removedirs(top)
+ except OSError:
+ pass
+
+
+class TestMercurialStorage(object):
+ def setUp(self):
+ self.repo_path = tempfile.mkdtemp()
+ self.repo = vstorage.VersionedStorage(self.repo_path)
+
+ def tearDown(self):
+ clear_directory(self.repo_path)
+
+ def test_save_text(self):
+ text = u"test text"
+ title = u"test title"
+ author = u"test author"
+ comment = u"test comment"
+ self.repo.save_text(title, text, author, comment, parent=-1)
+ saved = self.repo.open_page(title).read()
+ assert saved == text
+
+ def test_save_text_noparent(self):
+ text = u"test text"
+ title = u"test title"
+ author = u"test author"
+ comment = u"test comment"
+ self.repo.save_text(title, text, author, comment, parent=None)
+ saved = self.repo.open_page(title).read()
+ assert saved == text
+
+ def test_save_merge_no_conflict(self):
+ text = u"test\ntext"
+ title = u"test title"
+ author = u"test author"
+ comment = u"test comment"
+ self.repo.save_text(title, text, author, comment, parent=-1)
+ self.repo.save_text(title, text, author, comment, parent=-1)
+ saved = self.repo.open_page(title).read()
+ assert saved == text
+
+ def test_save_merge_line_conflict(self):
+ text = u"test\ntest\n"
+ text1 = u"test\ntext\n"
+ text2 = u"text\ntest\n"
+ title = u"test title"
+ author = u"test author"
+ comment = u"test comment"
+ self.repo.save_text(title, text, author, comment, parent=-1)
+ self.repo.save_text(title, text1, author, comment, parent=0)
+ self.repo.save_text(title, text2, author, comment, parent=0)
+ saved = self.repo.open_page(title).read()
+ # Other conflict markers placement can also be correct
+ assert_equal(saved, u'''\
+text
+test
+<<<<<<< local
+=======
+text
+>>>>>>> other
+''')
+
+ def test_delete(self):
+ text = u"text test"
+ title = u"test title"
+ author = u"test author"
+ comment = u"test comment"
+ self.repo.save_text(title, text, author, comment, parent=-1)
+ assert title in self.repo
+ self.repo.delete_page(title, author, comment)
+ assert title not in self.repo
+
+ @raises(vstorage.DocumentNotFound)
+ def test_document_not_found(self):
+ self.repo.open_page(u'unknown entity')
+
+
--- /dev/null
+# -*- coding: utf-8 -*-
+import os
+import tempfile
+import datetime
+import mimetypes
+import urllib
+
+import sys
+
+# Note: we have to set these before importing Mercurial
+os.environ['HGENCODING'] = 'utf-8'
+os.environ['HGMERGE'] = "internal:merge"
+
+import mercurial.hg
+import mercurial.ui
+import mercurial.revlog
+import mercurial.util
+
+
+def urlquote(url, safe='/'):
+ """Quotes URL
+
+ >>> urlquote(u'Za\u017c\xf3\u0142\u0107 g\u0119\u015bl\u0105 ja\u017a\u0144')
+ 'Za%C5%BC%C3%B3%C5%82%C4%87_g%C4%99%C5%9Bl%C4%85_ja%C5%BA%C5%84'
+ """
+ return urllib.quote(url.replace(' ', '_').encode('utf-8', 'ignore'), safe)
+
+def urlunquote(url):
+ """Unqotes URL
+
+ # >>> urlunquote('Za%C5%BC%C3%B3%C5%82%C4%87_g%C4%99%C5%9Bl%C4%85_ja%C5%BA%C5%84')
+ # u'Za\u017c\xf3\u0142\u0107 g\u0119\u015bl\u0105 ja\u017a\u0144'
+ """
+ return unicode(urllib.unquote(url), 'utf-8', 'ignore').replace('_', ' ')
+
+def find_repo_path(path):
+ """Go up the directory tree looking for a Mercurial repository (a directory containing a .hg subdirectory)."""
+ while not os.path.isdir(os.path.join(path, ".hg")):
+ old_path, path = path, os.path.dirname(path)
+ if path == old_path:
+ return None
+ return path
+
+def locked_repo(func):
+ """A decorator for locking the repository when calling a method."""
+
+ def new_func(self, *args, **kwargs):
+ """Wrap the original function in locks."""
+
+ wlock = self.repo.wlock()
+ lock = self.repo.lock()
+ try:
+ func(self, *args, **kwargs)
+ finally:
+ lock.release()
+ wlock.release()
+
+ return new_func
+
+
+class DocumentNotFound(Exception):
+ pass
+
+
+class VersionedStorage(object):
+ """
+ Provides means of storing text pages and keeping track of their
+ change history, using Mercurial repository as the storage method.
+ """
+
+ def __init__(self, path, charset=None):
+ """
+ Takes the path to the directory where the pages are to be kept.
+ If the directory doen't exist, it will be created. If it's inside
+ a Mercurial repository, that repository will be used, otherwise
+ a new repository will be created in it.
+ """
+
+ self.charset = charset or 'utf-8'
+ self.path = path
+ if not os.path.exists(self.path):
+ os.makedirs(self.path)
+ self.repo_path = find_repo_path(self.path)
+ try:
+ self.ui = mercurial.ui.ui(report_untrusted=False,
+ interactive=False, quiet=True)
+ except TypeError:
+ # Mercurial 1.3 changed the way we setup the ui object.
+ self.ui = mercurial.ui.ui()
+ self.ui.quiet = True
+ self.ui._report_untrusted = False
+ self.ui.setconfig('ui', 'interactive', False)
+ if self.repo_path is None:
+ self.repo_path = self.path
+ create = True
+ else:
+ create = False
+ self.repo_prefix = self.path[len(self.repo_path):].strip('/')
+ self.repo = mercurial.hg.repository(self.ui, self.repo_path,
+ create=create)
+
+ def reopen(self):
+ """Close and reopen the repo, to make sure we are up to date."""
+
+ self.repo = mercurial.hg.repository(self.ui, self.repo_path)
+
+ def _file_path(self, title):
+ return os.path.join(self.path, urlquote(title, safe=''))
+
+ def _title_to_file(self, title):
+ return os.path.join(self.repo_prefix, urlquote(title, safe=''))
+
+ def _file_to_title(self, filename):
+ assert filename.startswith(self.repo_prefix)
+ name = filename[len(self.repo_prefix):].strip('/')
+ return urlunquote(name)
+
+ def __contains__(self, title):
+ return os.path.exists(self._file_path(title))
+
+ def __iter__(self):
+ return self.all_pages()
+
+ def merge_changes(self, changectx, repo_file, text, user, parent):
+ """Commits and merges conflicting changes in the repository."""
+ tip_node = changectx.node()
+ filectx = changectx[repo_file].filectx(parent)
+ parent_node = filectx.changectx().node()
+
+ self.repo.dirstate.setparents(parent_node)
+ node = self._commit([repo_file], text, user)
+
+ partial = lambda filename: repo_file == filename
+
+ # If p1 is equal to p2, there is no work to do. Even the dirstate is correct.
+ p1, p2 = self.repo[None].parents()[0], self.repo[tip_node]
+ if p1 == p2:
+ return text
+
+ try:
+ unresolved = mercurial.merge.update(self.repo, tip_node, True, False, partial)
+ except mercurial.util.Abort:
+ raise
+ unresolved = 1, 1, 1, 1
+
+ self.repo.dirstate.setparents(tip_node, node)
+ # Mercurial 1.1 and later need updating the merge state
+ try:
+ mercurial.merge.mergestate(self.repo).mark(repo_file, "r")
+ except (AttributeError, KeyError):
+ pass
+ return u'merge of edit conflict'
+
+ @locked_repo
+ def save_file(self, title, file_name, author=u'', comment=u'', parent=None):
+ """Save an existing file as specified page."""
+
+ user = author.encode('utf-8') or u'anon'.encode('utf-8')
+ text = comment.encode('utf-8') or u'comment'.encode('utf-8')
+ repo_file = self._title_to_file(title)
+ file_path = self._file_path(title)
+ mercurial.util.rename(file_name, file_path)
+ changectx = self._changectx()
+ try:
+ filectx_tip = changectx[repo_file]
+ current_page_rev = filectx_tip.filerev()
+ except mercurial.revlog.LookupError:
+ self.repo.add([repo_file])
+ current_page_rev = -1
+ if parent is not None and current_page_rev != parent:
+ msg = self.merge_changes(changectx, repo_file, text, user, parent)
+ user = '<wiki>'
+ text = msg.encode('utf-8')
+ self._commit([repo_file], text, user)
+
+
+ def _commit(self, files, text, user):
+ try:
+ return self.repo.commit(files=files, text=text, user=user,
+ force=True, empty_ok=True)
+ except TypeError:
+ # Mercurial 1.3 doesn't accept empty_ok or files parameter
+ match = mercurial.match.exact(self.repo_path, '', list(files))
+ return self.repo.commit(match=match, text=text, user=user,
+ force=True)
+
+
+ def save_data(self, title, data, author=u'', comment=u'', parent=None):
+ """Save data as specified page."""
+
+ try:
+ temp_path = tempfile.mkdtemp(dir=self.path)
+ file_path = os.path.join(temp_path, 'saved')
+ f = open(file_path, "wb")
+ f.write(data)
+ f.close()
+ self.save_file(title, file_path, author, comment, parent)
+ finally:
+ try:
+ os.unlink(file_path)
+ except OSError:
+ pass
+ try:
+ os.rmdir(temp_path)
+ except OSError:
+ pass
+
+ def save_text(self, title, text, author=u'', comment=u'', parent=None):
+ """Save text as specified page, encoded to charset."""
+
+ data = text.encode(self.charset)
+ self.save_data(title, data, author, comment, parent)
+
+ def page_text(self, title):
+ """Read unicode text of a page."""
+
+ data = self.open_page(title).read()
+ text = unicode(data, self.charset, 'replace')
+ return text
+
+ def page_lines(self, page):
+ for data in page:
+ yield unicode(data, self.charset, 'replace')
+
+ @locked_repo
+ def delete_page(self, title, author=u'', comment=u''):
+ user = author.encode('utf-8') or 'anon'
+ text = comment.encode('utf-8') or 'deleted'
+ repo_file = self._title_to_file(title)
+ file_path = self._file_path(title)
+ try:
+ os.unlink(file_path)
+ except OSError:
+ pass
+ self.repo.remove([repo_file])
+ self._commit([repo_file], text, user)
+
+ def open_page(self, title):
+ try:
+ return open(self._file_path(title), "rb")
+ except IOError:
+ raise DocumentNotFound()
+
+ def page_file_meta(self, title):
+ """Get page's inode number, size and last modification time."""
+
+ try:
+ (st_mode, st_ino, st_dev, st_nlink, st_uid, st_gid, st_size,
+ st_atime, st_mtime, st_ctime) = os.stat(self._file_path(title))
+ except OSError:
+ return 0, 0, 0
+ return st_ino, st_size, st_mtime
+
+ def page_meta(self, title):
+ """Get page's revision, date, last editor and his edit comment."""
+
+ filectx_tip = self._find_filectx(title)
+ if filectx_tip is None:
+ raise DocumentNotFound()
+ #return -1, None, u'', u''
+ rev = filectx_tip.filerev()
+ filectx = filectx_tip.filectx(rev)
+ date = datetime.datetime.fromtimestamp(filectx.date()[0])
+ author = unicode(filectx.user(), "utf-8",
+ 'replace').split('<')[0].strip()
+ comment = unicode(filectx.description(), "utf-8", 'replace')
+ return rev, date, author, comment
+
+ def repo_revision(self):
+ return self._changectx().rev()
+
+ def page_mime(self, title):
+ """
+ Guess page's mime type ased on corresponding file name.
+ Default ot text/x-wiki for files without an extension.
+
+ # >>> page_mime('something.txt')
+ # 'text/plain'
+ # >>> page_mime('SomePage')
+ # 'text/x-wiki'
+ # >>> page_mime(u'ąęśUnicodePage')
+ # 'text/x-wiki'
+ # >>> page_mime('image.png')
+ # 'image/png'
+ # >>> page_mime('style.css')
+ # 'text/css'
+ # >>> page_mime('archive.tar.gz')
+ # 'archive/gzip'
+ """
+
+ addr = self._file_path(title)
+ mime, encoding = mimetypes.guess_type(addr, strict=False)
+ if encoding:
+ mime = 'archive/%s' % encoding
+ if mime is None:
+ mime = 'text/x-wiki'
+ return mime
+
+ def _changectx(self):
+ """Get the changectx of the tip."""
+ try:
+ # This is for Mercurial 1.0
+ return self.repo.changectx()
+ except TypeError:
+ # Mercurial 1.3 (and possibly earlier) needs an argument
+ return self.repo.changectx('tip')
+
+ def _find_filectx(self, title):
+ """Find the last revision in which the file existed."""
+
+ repo_file = self._title_to_file(title)
+ changectx = self._changectx()
+ stack = [changectx]
+ while repo_file not in changectx:
+ if not stack:
+ return None
+ changectx = stack.pop()
+ for parent in changectx.parents():
+ if parent != changectx:
+ stack.append(parent)
+ return changectx[repo_file]
+
+ def page_history(self, title):
+ """Iterate over the page's history."""
+
+ filectx_tip = self._find_filectx(title)
+ if filectx_tip is None:
+ return
+ maxrev = filectx_tip.filerev()
+ minrev = 0
+ for rev in range(maxrev, minrev-1, -1):
+ filectx = filectx_tip.filectx(rev)
+ date = datetime.datetime.fromtimestamp(filectx.date()[0])
+ author = unicode(filectx.user(), "utf-8",
+ 'replace').split('<')[0].strip()
+ comment = unicode(filectx.description(), "utf-8", 'replace')
+ yield rev, date, author, comment
+
+ def page_revision(self, title, rev):
+ """Get unicode contents of specified revision of the page."""
+
+ filectx_tip = self._find_filectx(title)
+ if filectx_tip is None:
+ raise DocumentNotFound()
+ try:
+ data = filectx_tip.filectx(rev).data()
+ except IndexError:
+ raise DocumentNotFound()
+ return data
+
+ def revision_text(self, title, rev):
+ data = self.page_revision(title, rev)
+ text = unicode(data, self.charset, 'replace')
+ return text
+
+ def history(self):
+ """Iterate over the history of entire wiki."""
+
+ changectx = self._changectx()
+ maxrev = changectx.rev()
+ minrev = 0
+ for wiki_rev in range(maxrev, minrev-1, -1):
+ change = self.repo.changectx(wiki_rev)
+ date = datetime.datetime.fromtimestamp(change.date()[0])
+ author = unicode(change.user(), "utf-8",
+ 'replace').split('<')[0].strip()
+ comment = unicode(change.description(), "utf-8", 'replace')
+ for repo_file in change.files():
+ if repo_file.startswith(self.repo_prefix):
+ title = self._file_to_title(repo_file)
+ try:
+ rev = change[repo_file].filerev()
+ except mercurial.revlog.LookupError:
+ rev = -1
+ yield title, rev, date, author, comment
+
+ def all_pages(self):
+ """Iterate over the titles of all pages in the wiki."""
+
+ for filename in os.listdir(self.path):
+ if (os.path.isfile(os.path.join(self.path, filename))
+ and not filename.startswith('.')):
+ yield urlunquote(filename)
+
+ def changed_since(self, rev):
+ """Return all pages that changed since specified repository revision."""
+
+ try:
+ last = self.repo.lookup(int(rev))
+ except IndexError:
+ for page in self.all_pages():
+ yield page
+ return
+ current = self.repo.lookup('tip')
+ status = self.repo.status(current, last)
+ modified, added, removed, deleted, unknown, ignored, clean = status
+ for filename in modified+added+removed+deleted:
+ if filename.startswith(self.repo_prefix):
+ yield self._file_to_title(filename)
+++ /dev/null
-# -*- encoding: utf-8 -*-
-__author__="Łukasz Rekucki"
-__date__ ="$2009-09-18 10:49:24$"
-__doc__ = """Main module for the Repository Abstraction Layer"""
-
-class Library(object):
-
- def __init__(self, create=False):
- """Open an existing library, or create a new one. By default, fails if
- the library doesn't exist."""
- self.create = create
-
- def documents(self):
- """List all documents in the library."""
- pass
-
- def document_for_revision(self, rev):
- """Retrieve a document in the specified revision."""
- pass
-
- def document(self, docid, user=None, rev='latest'):
- """Retrieve a document from a library."""
- pass
-
- def get_revision(self, revid):
- """Retrieve a handle to a specified revision."""
- return None
-
- def document_create(self, docid):
- """Create a new document. The document will have it's own branch."""
-
-
-class Document(object):
- """A class representing a document package boundled with a revision."""
-
- def __init__(self, library, revision):
- """_library_ should be an instance of a Library."""
- self._library = library
- if isinstance(revision, Revision):
- self._revision = revision
- else:
- self._revision = library.get_revision(revision)
-
-
- def take(self, user):
- """Make a user copy of the document. This is persistant."""
- pass
-
- def giveback(self):
- """Informs the library, that the user no longer needs this document.
- Should be called on the user version of document. If not, it doesn nothing."""
-
- def data(self, entry):
- """Returns the specified entry as a unicode data."""
- pass
-
- @property
- def library(self):
- return self._library
-
- @property
- def revision(self):
- return self._revision
-
- @property
- def id(self):
- return self._revision.document_name
-
- @property
- def owner(self):
- return self._revision.user_name
-
- def parentof(self, other):
- return self._revision.parentof(other._revision)
-
- def parent(self):
- return self._library.document_for_revision(self._revision.parent())
-
- def has_parent_from(self, other):
- return self._revision.has_parent_from(other._revision)
-
- def ancestorof(self, other):
- return self._revision.ancestorof(other._revision)
-
-
-class Revision(object):
-
- def __init__(self, lib):
- self._library = lib
-
- def parentof(self, other):
- return False
-
- def ancestorof(self, other):
- return False
-
- @property
- def document_name(self):
- raise ValueError()
-
- @property
- def user_name(self):
- raise ValueError()
-
-#
-# Exception classes
-#
-
-class LibraryException(Exception):
- def __init__(self, msg, cause=None):
- Exception.__init__(self, msg)
- self.cause = cause
-
-class UpdateException(LibraryException):
- pass
-
-class OutdatedException(LibraryException):
- pass
-
-class RevisionNotFound(LibraryException):
- def __init__(self, rev):
- LibraryException.__init__(self, "Revision %r not found." % rev)
-
-class RevisionMismatch(LibraryException):
- def __init__(self, fdi, rev):
- LibraryException.__init__(self, "No revision %r for document %r." % (rev, fdi))
-
-class EntryNotFound(LibraryException):
- def __init__(self, rev, entry, guesses=[]):
- LibraryException.__init__(self, \
- u"Entry '%s' at revision %r not found. %s" % (entry, rev, \
- (u"Posible values:\n" + u',\n'.join(guesses)) if len(guesses) else u'') )
-
-class DocumentAlreadyExists(LibraryException):
- pass
-
-# import backends to local namespace
-
-def open_library(path, proto, *args, **kwargs):
- if proto == 'hg':
- import wlrepo.mercurial_backend.library
- return wlrepo.mercurial_backend.library.MercurialLibrary(path, *args, **kwargs)
-
- raise NotImplemented()
\ No newline at end of file
+++ /dev/null
-# -*- encoding: utf-8 -*-
-
-import logging
-log = logging.getLogger('ral.mercurial')
-
-__author__= "Łukasz Rekucki"
-__date__ = "$2009-09-25 09:20:22$"
-__doc__ = "Module documentation."
-
-import wlrepo
-
-
-from mercurial import encoding
-encoding.encoding = 'utf-8'
\ No newline at end of file
+++ /dev/null
-# -*- encoding: utf-8 -*-
-
-import logging
-log = logging.getLogger('ral.mercurial')
-
-__author__ = "Łukasz Rekucki"
-__date__ = "$2009-09-25 09:35:06$"
-__doc__ = "Module documentation."
-
-import wlrepo
-import mercurial.error
-import re
-
-import logging
-log = logging.getLogger('wlrepo.document')
-
-class MercurialDocument(wlrepo.Document):
-
- def data(self, entry):
- path = self._library._sanitize_string(self.id + u'.' + entry)
- try:
- return self._library._filectx(path, \
- self._revision.hgrev()).data().decode('utf-8')
- except mercurial.error.LookupError, e:
- fl = [x.decode('utf-8') for x in self._revision._changectx]
- raise wlrepo.EntryNotFound(self._revision, path.decode('utf-8'), fl)
-
- def quickwrite(self, entry, data, msg, user=None):
- user = user or self.owner
-
- if isinstance(data, unicode):
- data = data.encode('utf-8')
-
- user = self._library._sanitize_string(user)
- msg = self._library._sanitize_string(msg)
- entry = self._library._sanitize_string(entry)
-
- if user is None:
- raise ValueError("Can't determine user.")
-
- def write(l, r):
- f = l._fileopen(r(entry), "w+")
- f.write(data)
- f.close()
- l._fileadd(r(entry))
-
- return self.invoke_and_commit(write, lambda d: (msg, \
- self._library._sanitize_string(self.owner)) )
-
- def invoke_and_commit(self, ops, commit_info):
- lock = self._library.lock()
- try:
- self._library._checkout(self._revision.hgrev())
-
- def entry_path(entry):
- return self._library._sanitize_string(self.id + u'.' + entry)
-
- ops(self._library, entry_path)
- message, user = commit_info(self)
-
- message = self._library._sanitize_string(message)
- user = self._library._sanitize_string(user)
-
- self._library._commit(message, user)
- try:
- return self._library.document(docid=self.id, user=user)
- except Exception, e:
- # rollback the last commit
- self._library._rollback()
- raise e
- finally:
- lock.release()
-
- # def commit(self, message, user):
- # """Make a new commit."""
- # self.invoke_and_commit(message, user, lambda *a: True)
-
- def ismain(self):
- return self._revision.user_name is None
-
- def islatest(self):
- return (self == self.latest())
-
- def shared(self):
- if self.ismain():
- return self
-
- return self._library.document(docid=self.id)
-
- def latest(self):
- return self._library.document(docid=self.id, user=self.owner)
-
- def take(self, user):
- fullid = self._library.fulldocid(self.id, user)
-
- def take_action(library, resolve):
- # branch from latest
- library._set_branchname(fullid)
-
- if not self._library.has_revision(fullid):
- log.info("Checking out document %s" % fullid)
-
- self.invoke_and_commit(take_action, \
- lambda d: ("$AUTO$ File checkout.", user) )
-
- return self._library.document_for_revision(fullid)
-
- def up_to_date(self):
- if self.ismain():
- return True
-
- shared = self.shared()
-
- if shared.ancestorof(self):
- return True
-
- if shared.has_parent_from(self):
- return True
-
- return False
-
- def update(self, user):
- """Update parts of the document."""
- lock = self.library.lock()
- try:
- if self.ismain():
- # main revision of the document
- return self
-
- # check for children in this branch
- if self._revision.has_children(limit_branch=True):
- raise wlrepo.UpdateException("Revision %s has children." % self.revision)
-
- shared = self.shared()
- # *
- # /|
- # * |
- # | |
- #
- # we carry the latest version
- if shared.ancestorof(self):
- return self
-
- # *
- # |\
- # | *
- # | |
- #
- # We just shared
- if self.parentof(shared):
- return self
-
- # s s S
- # | | |
- # *<-S *<-*
- # | | | .
- #
- # This is ok (s - shared, S - self)
-
- if self._revision.merge_with(shared._revision, user=user,\
- message="$AUTO$ Personal branch update."):
- return self.latest()
- else:
- raise wlrepo.UpdateException("Merge failed.")
- finally:
- lock.release()
-
-
- def would_share(self):
- if self.ismain():
- return False, "Main version is always shared"
-
- shared = self.shared()
-
- # we just did this - move on
- if self.parentof(shared):
- return False, "Document has been recetly shared - no changes"
-
- # *
- # /|
- # * *
- # |\|
- # | *
- # | |
- # Situation above is ok - what we don't want, is:
- # *
- # /|
- # * |
- # |\|
- # | *
- # | |
- # We want to prevent stuff like this.
- if self.parent().parentof(shared) and shared.parentof(self):
- return False, "Preventing zig-zag"
-
- return True, "All ok"
-
- def share(self, message):
- lock = self.library.lock()
- try:
- # check if the document is in "updated" state
- if not self.up_to_date():
- raise wlrepo.OutdatedException("You must update your document before share.")
-
- # now check if there is anything to do
- need_work, info = self.would_share()
-
- if not need_work:
- return self.shared()
-
- # The good situation
- #
- # * local
- # |
- # >*
- # ||
- # / |
- # main * *
- # | |
- shared = self.shared()
-
- try:
- success = shared._revision.merge_with(self._revision, user=self.owner, message=message)
- if not success:
- raise wlrepo.LibraryException("Merge failed.")
-
- return shared.latest()
- except Abort, e:
- raise wlrepo.LibraryException( repr(e) )
- finally:
- lock.release()
-
-
- def has_conflict_marks(self):
- return re.search("^(?:<<<<<<< .*|=======|>>>>>>> .*)$", self.data('xml'), re.MULTILINE)
-
- def __unicode__(self):
- return u"Document(%s:%s)" % (self.id, self.owner)
-
- def __str__(self):
- return self.__unicode__().encode('utf-8')
-
- def __eq__(self, other):
- return (self._revision == other._revision)
-
+++ /dev/null
-# -*- encoding: utf-8 -*-
-
-import logging
-log = logging.getLogger('ral.mercurial')
-
-__author__= "Łukasz Rekucki"
-__date__ = "$2009-09-25 09:33:02$"
-__doc__ = "Module documentation."
-
-import mercurial
-from mercurial import localrepo as hglrepo
-from mercurial import ui as hgui
-from mercurial import error
-import wlrepo
-
-from wlrepo.mercurial_backend.revision import MercurialRevision
-from wlrepo.mercurial_backend.document import MercurialDocument
-
-class MergeStatus(object):
- def __init__(self, mstatus):
- self.updated = mstatus[0]
- self.merged = mstatus[1]
- self.removed = mstatus[2]
- self.unresolved = mstatus[3]
-
- def isclean(self):
- return self.unresolved == 0
-
-class UpdateStatus(object):
-
- def __init__(self, mstatus):
- self.modified = mstatus[0]
- self.added = mstatus[1]
- self.removed = mstatus[2]
- self.deleted = mstatus[3]
- self.untracked = mstatus[4]
- self.ignored = mstatus[5]
- self.clean = mstatus[6]
-
- def has_changes(self):
- return bool(len(self.modified) + len(self.added) + \
- len(self.removed) + len(self.deleted))
-
-class MercurialLibrary(wlrepo.Library):
- """Mercurial implementation of the Library API"""
-
- def __init__(self, path, **kwargs):
- super(wlrepo.Library, self).__init__( ** kwargs)
-
- self._revcache = {}
- self._doccache = {}
-
- self._hgui = hgui.ui()
- self._hgui.config('ui', 'quiet', 'true')
- self._hgui.config('ui', 'interactive', 'false')
-
- import os.path
- self._ospath = self._sanitize_string(os.path.realpath(path))
-
- if os.path.isdir(path):
- try:
- self._hgrepo = hglrepo.localrepository(self._hgui, path)
- except mercurial.error.RepoError:
- raise wlrepo.LibraryException("[HGLibrary] Not a valid repository at path '%s'." % path)
- elif kwargs.get('create', False):
- os.makedirs(path)
- try:
- self._hgrepo = hglrepo.localrepository(self._hgui, path, create=1)
- except mercurial.error.RepoError:
- raise wlrepo.LibraryException("[HGLibrary] Can't create a repository on path '%s'." % path)
- else:
- raise wlrepo.LibraryException("[HGLibrary] Can't open a library on path '%s'." % path)
-
-
- def documents(self):
- return [ key[5:].decode('utf-8') for key in \
- self._hgrepo.branchmap() if key.startswith("$doc:") ]
-
- @property
- def ospath(self):
- return self._ospath.decode('utf-8')
-
- def document_for_revision(self, revision):
- if revision is None:
- raise ValueError("Revision can't be None.")
-
- if not isinstance(revision, MercurialRevision):
- rev = self.get_revision(revision)
- else:
- rev = revision
-
- if not self._doccache.has_key(str(rev)):
- self._doccache[str(rev)] = MercurialDocument(self, rev)
-
- # every revision is a document
- return self._doccache[str(rev)]
-
- def document(self, docid, user=None, rev=u'latest'):
- rev = self._sanitize_string(rev)
-
- if rev != u'latest':
- doc = self.document_for_revision(rev)
-
- if doc.id != docid or (doc.owner != user):
- raise wlrepo.RevisionMismatch(self.fulldocid(docid, user)+u'@'+unicode(rev))
-
- return doc
- else:
- return self.document_for_revision(self.fulldocid(docid, user))
-
- def get_revision(self, revid):
- revid = self._sanitize_string(revid)
-
- log.info("Looking up rev %r (%s)" %(revid, type(revid)) )
-
- try:
- ctx = self._changectx( revid )
- except mercurial.error.RepoError, e:
- raise wlrepo.RevisionNotFound(revid)
-
- if ctx is None:
- raise wlrepo.RevisionNotFound(revid)
-
- return self._revision(ctx)
-
- def _revision(self, ctx):
- if self._revcache.has_key(ctx):
- return self._revcache[ctx]
-
- return MercurialRevision(self, ctx)
-
- def fulldocid(self, docid, user=None):
- fulldocid = u''
- if user is not None:
- fulldocid += u'$user:' + user
- fulldocid += u'$doc:' + docid
- return fulldocid
-
- def has_revision(self, revid):
- revid = self._sanitize_string(revid)
-
- try:
- self._hgrepo[revid]
- return True
- except mercurial.error.RepoError:
- return False
-
- def document_create(self, docid):
-
- # check if it already exists
- fullid = self.fulldocid(docid)
-
- if self.has_revision(fullid):
- raise wlrepo.DocumentAlreadyExists(u"Document %s already exists!" % docid);
-
- # doesn't exist
- self._create_branch(self._sanitize_string(fullid))
- return self.document_for_revision(fullid)
-
- #
- # Private methods
- #
-
- #
- # Locking
- #
-
- def lock(self, write_mode=False):
- return self._hgrepo.wlock() # no support for read/write mode yet
-
- def _transaction(self, write_mode, action):
- lock = self.lock(write_mode)
- try:
- return action(self)
- finally:
- lock.release()
-
- #
- # Basic repo manipulation
- #
-
- def _checkout(self, rev, force=True):
- return MergeStatus(mercurial.merge.update(self._hgrepo, rev, False, force, None))
-
- def _merge(self, rev):
- """ Merge the revision into current working directory """
- return MergeStatus(mercurial.merge.update(self._hgrepo, rev, True, False, None))
-
- def _common_ancestor(self, revA, revB):
- return self._hgrepo[revA].ancestor(self.repo[revB])
-
- def _commit(self, message, user=u"library"):
- return self._hgrepo.commit(text=message, user=user)
-
-
- def _fileexists(self, fileid):
- return (fileid in self._hgrepo[None])
-
- def _fileadd(self, fileid):
- return self._hgrepo.add([fileid])
-
- def _filesadd(self, fileid_list):
- return self._hgrepo.add(fileid_list)
-
- def _filerm(self, fileid):
- return self._hgrepo.remove([fileid])
-
- def _filesrm(self, fileid_list):
- return self._hgrepo.remove(fileid_list)
-
- def _fileopen(self, path, mode):
- return self._hgrepo.wopener(path, mode)
-
- def _filectx(self, fileid, revid):
- return self._hgrepo.filectx(fileid, changeid=revid)
-
- def _changectx(self, nodeid):
- return self._hgrepo.changectx(nodeid)
-
- def _rollback(self):
- return self._hgrepo.rollback()
-
- #
- # BASIC BRANCH routines
- #
-
- def _bname(self, user, docid):
- """Returns a branch name for a given document and user."""
- docid = self._sanitize_string(docid)
- uname = self._sanitize_string(user)
- return "personal_" + uname + "_file_" + docid;
-
- def _has_branch(self, name):
- return self._hgrepo.branchmap().has_key(self._sanitize_string(name))
-
- def _branch_tip(self, name):
- name = self._sanitize_string(name)
- return self._hgrepo.branchtags()[name]
-
- def _create_branch(self, name, parent=None, before_commit=None, message=None):
- name = self._sanitize_string(name)
-
- if self._has_branch(name): return # just exit
-
- if parent is None:
- parentrev = self._hgrepo['$branchbase'].node()
- else:
- parentrev = parent.hgrev()
-
- self._checkout(parentrev)
- self._set_branchname(name)
-
- if before_commit: before_commit(self)
-
- message = message or "$ADMN$ Initial commit for branch '%s'." % name
- self._commit(message, user='$library')
-
- def _set_branchname(self, name):
- name = self._sanitize_string(name)
- self._hgrepo.dirstate.setbranch(name)
-
- def _switch_to_branch(self, branchname):
- current = self._hgrepo[None].branch()
-
- if current == branchname:
- return current # quick exit
-
- self._checkout(self._branch_tip(branchname))
- return branchname
-
- #
- # Utils
- #
-
- @staticmethod
- def _sanitize_string(s):
- if s is None:
- return None
-
- if isinstance(s, unicode):
- s = s.encode('utf-8')
-
- return s
\ No newline at end of file
+++ /dev/null
-# -*- encoding: utf-8 -*-
-
-__author__= "Łukasz Rekucki"
-__date__ = "$2009-10-20 12:31:48$"
-__doc__ = "Module documentation."
-
-import wlrepo
-from mercurial.node import nullid
-
-class MercurialRevision(wlrepo.Revision):
- def __init__(self, lib, changectx):
- super(MercurialRevision, self).__init__(lib)
- self._changectx = changectx
-
- branchname = self._changectx.branch()
- if branchname.startswith("$doc:"):
- self._docname = branchname[5:]
- self._username = None
- elif branchname.startswith("$user:"):
- idx = branchname.find("$doc:")
- if(idx < 0):
- raise ValueError("Revision %s is not a valid document revision." % changectx.hex());
- self._username = branchname[6:idx]
- self._docname = branchname[idx+5:]
- else:
- raise ValueError("Revision %s is not a valid document revision." % changectx.hex());
-
- @property
- def document_name(self):
- return self._docname and self._docname.decode('utf-8')
-
- @property
- def user_name(self):
- return self._username and self._username.decode('utf-8')
-
- def hgrev(self):
- return self._changectx.node()
-
- def hgcontext(self):
- return self._changectx
-
- def hgbranch(self):
- return self._changectx.branch()
-
- @property
- def timestamp(self):
- return self._changectx.date()[0]
-
- def __unicode__(self):
- return u"%s" % self._changectx.hex()
-
- def __str__(self):
- return self.__unicode__().encode('utf-8')
-
- def __repr__(self):
- return "%s" % self._changectx.hex()
-
- def ancestorof(self, other):
- nodes = list(other._changectx._parents)
- while nodes[0].node() != nullid:
- v = nodes.pop(0)
- if v == self._changectx:
- return True
- nodes.extend( v._parents )
- return False
-
- def parentof(self, other):
- return self._changectx in other._changectx._parents
-
- def has_common_ancestor(self, other):
- a = self._changectx.ancestor(other._changectx)
- return (a.branch() == self._changectx.branch())
-
- def has_children(self, limit_branch=False):
- for child in self._changectx.children():
- cbranch = child.branch()
- if (not limit_branch) or (cbranch == self.hgbranch()):
- return True
- return False
-
- def has_parent_from(self, rev):
- branch = rev.hgbranch()
- for parent in self._changectx.parents():
- if parent.branch() == branch:
- return True
- return False
-
- def merge_with(self, other, user, message):
- message = self._library._sanitize_string(message)
- lock = self._library.lock(True)
- try:
- self._library._checkout(self._changectx.node())
- status = self._library._merge(other._changectx.node())
- if status.isclean():
- self._library._commit(user=user, message=message)
- return True
- else:
- return False
- finally:
- lock.release()
-
- def parent(self):
- parents = self._changectx.parents()
-
- if len(parents) == 1:
- return self._library._revision(parents[0])
-
- if parents[0].branch() == self.hgbranch():
- return self._library._revision(parents[0])
- else:
- return self._library._revision(parents[1])
-
- def __eq__(self, other):
- return self._changectx.node() == other._changectx.node()
+++ /dev/null
-# -*- encoding: utf-8 -*-
-
-__author__= "Łukasz Rekucki"
-__date__ = "$2009-09-18 14:42:53$"
-__doc__ = "Test package for WL-RAL"
-
-
+++ /dev/null
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 4
-26ecd403388cce326f6ca1e32625db0651900eb7 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
+++ /dev/null
-personal_tester_file_valid_file
\ No newline at end of file
+++ /dev/null
-Ala ma kota
+++ /dev/null
-9208a537d2f003ccc46a90595e7f0801c6b64d45 7
-26ecd403388cce326f6ca1e32625db0651900eb7 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-9208a537d2f003ccc46a90595e7f0801c6b64d45 personal_admin_file_ala
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
-data/pub_ala.xml.i
+++ /dev/null
-personal_admin_file_ala
\ No newline at end of file
+++ /dev/null
-Ala ma kota
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/$meta.i
-data/.hgignore.i
+++ /dev/null
-default
\ No newline at end of file
+++ /dev/null
-c0c0576584c8b0c74c7efd4aac52da3af0ee16a9 8
-c0c0576584c8b0c74c7efd4aac52da3af0ee16a9 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-9208a537d2f003ccc46a90595e7f0801c6b64d45 personal_admin_file_ala
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
-data/pub_ala.xml.i
+++ /dev/null
-default
\ No newline at end of file
+++ /dev/null
-Super tekst.
+++ /dev/null
-Ala ma kota
+++ /dev/null
-0985918ffd936a45d40005b7e0fa6feb504c94c5 9
-c0c0576584c8b0c74c7efd4aac52da3af0ee16a9 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-0985918ffd936a45d40005b7e0fa6feb504c94c5 personal_admin_file_ala
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
-data/pub_ala.xml.i
+++ /dev/null
-personal_admin_file_ala
\ No newline at end of file
+++ /dev/null
-Super tekst.
+++ /dev/null
-Ala ma kota
+++ /dev/null
-65046aefea667e38c7a01d7ca62412d7e81f68bc 2
-d5e516ebd357bca24c6afb737e97db3b3d4f111a default
-00e4aa6f4f53f31020e4d3e0579cb3c0fa0bd056 $doc:sample
-65046aefea667e38c7a01d7ca62412d7e81f68bc $doc:sample_pl
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/$meta.i
-data/.hgignore.i
-data/sample.parts.i
-data/sample.xml.i
-data/sample_pl.xml.i
-data/.hgtags.i
+++ /dev/null
-default
\ No newline at end of file
+++ /dev/null
-d5e516ebd357bca24c6afb737e97db3b3d4f111a $branchbase
+++ /dev/null
-2c0751c1c4cf5665d0a0af9c80b3050a45110436 9
-2c0751c1c4cf5665d0a0af9c80b3050a45110436 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-9208a537d2f003ccc46a90595e7f0801c6b64d45 personal_admin_file_ala
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
-data/pub_ala.xml.i
+++ /dev/null
-default
\ No newline at end of file
+++ /dev/null
-Super tekst. - w default.
+++ /dev/null
-Ala ma kota
+++ /dev/null
-046a51b7783408c0011f01ae6943791e62652c7e 10
-c0c0576584c8b0c74c7efd4aac52da3af0ee16a9 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-046a51b7783408c0011f01ae6943791e62652c7e personal_admin_file_ala
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
-data/pub_ala.xml.i
+++ /dev/null
-personal_admin_file_ala
\ No newline at end of file
+++ /dev/null
-Super tekst.
+++ /dev/null
-Ala ma kota
+++ /dev/null
-046a51b7783408c0011f01ae6943791e62652c7e 10
-c0c0576584c8b0c74c7efd4aac52da3af0ee16a9 default
-34d4aa807563b28774bcbe07f6acbb984aa4dc8d personal_tester_valid_file
-046a51b7783408c0011f01ae6943791e62652c7e personal_admin_file_ala
-e0fbb8dc3f54c34581848d59a215d1d222a9e891 personal_tester_file_valid_file
+++ /dev/null
-revlogv1
-store
-fncache
+++ /dev/null
-data/.hgignore.i
-data/ignored_file.i
-data/pub_valid_file.xml.i
-data/pub_polish_file.xml.i
-data/pub_ala.xml.i
+++ /dev/null
-default
\ No newline at end of file
+++ /dev/null
-Super tekst.
+++ /dev/null
-Ala ma kota
+++ /dev/null
-# -*- encoding: utf-8 -*-
-
-__author__= "Łukasz Rekucki"
-__date__ = "$2009-09-18 14:43:27$"
-__doc__ = "Tests for RAL mercurial backend."
-
-from nose.tools import *
-
-from wlrepo import MercurialLibrary
-
-import os, os.path, tempfile
-import shutil
-
-REPO_TEMPLATES = os.path.join( os.path.dirname(__file__), 'data')
-
-def temprepo(name):
-
- from functools import wraps
-
- def decorator(func):
- def decorated(*args, **kwargs):
- clean = False
- try:
- temp = tempfile.mkdtemp("", "testdir_" )
- path = os.path.join(temp, 'repo')
- shutil.copytree(os.path.join(REPO_TEMPLATES, name), path, False)
- kwargs['library'] = MercurialLibrary(path)
- func(*args, **kwargs)
- clean = True
- finally:
- #if not clean and self.response:
- # print "RESULT", func.__name__, ">>>"
- # print self.response
- # print "<<<"
- shutil.rmtree(temp, True)
-
- decorated = make_decorator(func)(decorated)
- return decorated
-
- return decorator
-
-@temprepo('clean')
-def test_opening(library):
- pass
-
-@temprepo('simple')
-def test_read_document(library):
- doc = library.document('sample')
- assert_equal(doc.data('xml'), 'Ala ma kota\n')
-
-@temprepo('simple')
-def test_read_UTF8_document(library):
- doc = library.document('sample_pl')
- assert_equal(doc.data('xml'), u'Gżegżółka\n'.encode('utf-8'))
-
-@temprepo('simple')
-def test_change_document(library):
- doc = library.document('sample')
- STRING = u'Gąski lubią pływać!\n'.encode('utf-8')
-
- def write_action(library, resolve):
- f = library._fileopen(resolve('xml'), 'r+')
- assert_equal(f.read(), 'Ala ma kota\n')
- f.seek(0)
- f.write(STRING)
- f.flush()
- f.seek(0)
- assert_equal(f.read(), STRING)
-
- def commit_info(document):
- return ("Document rewrite", "library")
-
- ndoc = doc.invoke_and_commit(write_action, commit_info)
- assert_equal(ndoc.data('xml'), STRING)
-
-
-@temprepo('simple')
-def test_create_document(library):
- assert_equal(sorted(library.documents()), sorted(['sample', 'sample_pl']))
-
- doc = library.document_create("creation")
- doc.quickwrite("xml", "<ala />", "Quick write", user="library")
-
- assert_equal(sorted(library.documents()), sorted(['sample', 'sample_pl', 'creation']))
-
-#
-#@temprepo('branched')
-#def test_switch_branch(library):
-# tester_cab = library.cabinet("valid_file", "tester", create=False)
-# assert_equal( list(tester_cab.parts()), ['valid_file'])
-#
-#@raises(wlrepo.CabinetNotFound)
-#@temprepo('branched')
-#def test_branch_not_found(library):
-# tester_cab = library.cabinet("ugh", "tester", create=False)
-#
-#@temprepo('branched')
-#def test_no_branches(library):
-# n4 = library.shelf(4)
-# n3 = library.shelf(3)
-# n2 = library.shelf(2)
-# n1 = library.shelf(1)
-# n0 = library.shelf(0)
-#
-# assert_true( n3.parentof(n4) )
-# assert_false( n4.parentof(n3) )
-# assert_true( n0.parentof(n1) )
-# assert_false( n1.parentof(n0) )
-# assert_false( n0.parentof(n4) )
-#
-## def test_ancestor_of_simple(self):
-# assert_true( n3.ancestorof(n4) )
-# assert_true( n2.ancestorof(n4) )
-# assert_true( n1.ancestorof(n4) )
-# assert_true( n0.ancestorof(n4) )
-#
-# assert_true( n2.ancestorof(n3) )
-# assert_true( n1.ancestorof(n3) )
-# assert_true( n0.ancestorof(n3) )
-#
-# assert_false( n4.ancestorof(n4) )
-# assert_false( n4.ancestorof(n3) )
-# assert_false( n3.ancestorof(n2) )
-# assert_false( n3.ancestorof(n1) )
-# assert_false( n3.ancestorof(n0) )
-#
-## def test_common_ancestor_simple(self):
-# assert_true( n3.has_common_ancestor(n4) )
-# assert_true( n3.has_common_ancestor(n3) )
-# assert_true( n3.has_common_ancestor(n3) )
-#
-#
-#@temprepo('branched2')
-#def test_once_branched(library):
-# n7 = library.shelf(7)
-# n6 = library.shelf(6)
-# n5 = library.shelf(5)
-# n4 = library.shelf(4)
-# n3 = library.shelf(3)
-# n2 = library.shelf(2)
-#
-# assert_true( n2.parentof(n3) )
-# assert_false( n3.parentof(n2) )
-#
-# assert_true( n2.parentof(n5) )
-# assert_false( n5.parentof(n2) )
-#
-# assert_false( n2.parentof(n4) )
-# assert_false( n2.parentof(n6) )
-# assert_false( n3.parentof(n5) )
-# assert_false( n5.parentof(n3) )
-#
-## def test_ancestorof_branched(self):
-# assert_true( n2.ancestorof(n7) )
-# assert_false( n7.ancestorof(n2) )
-# assert_true( n2.ancestorof(n6) )
-# assert_false( n6.ancestorof(n2) )
-# assert_true( n2.ancestorof(n5) )
-# assert_false( n5.ancestorof(n2) )
-#
-# assert_false( n3.ancestorof(n5) )
-# assert_false( n5.ancestorof(n3) )
-# assert_false( n4.ancestorof(n5) )
-# assert_false( n5.ancestorof(n4) )
-# assert_false( n3.ancestorof(n7) )
-# assert_false( n7.ancestorof(n3) )
-# assert_false( n4.ancestorof(n6) )
-# assert_false( n6.ancestorof(n4) )
-#
-## def test_common_ancestor_branched(self):
-# assert_true( n2.has_common_ancestor(n4) )
-# assert_true( n2.has_common_ancestor(n7) )
-# assert_true( n2.has_common_ancestor(n6) )
-#
-# # cause it's not in the right branch
-# assert_false( n5.has_common_ancestor(n3) )
-# assert_false( n7.has_common_ancestor(n4) )
-#
-#@temprepo('merged')
-#def test_after_merge(library):
-# n8 = library.shelf(8)
-# n7 = library.shelf(7)
-# n6 = library.shelf(6)
-#
-# assert_true( n7.parentof(n8) )
-# assert_false( n8.parentof(n7) )
-#
-# assert_true( n7.ancestorof(n8) )
-# assert_true( n6.ancestorof(n8) )
-#
-#
-# assert_true( n7.has_common_ancestor(n8) )
-# # cause it's not in the right branch
-# assert_false( n8.has_common_ancestor(n7) )
-#
-#@temprepo('merged_with_local_commit')
-#def test_after_merge_and_local_commit(library):
-# n9 = library.shelf(9)
-# n8 = library.shelf(8)
-# n7 = library.shelf(7)
-# n6 = library.shelf(6)
-#
-# assert_true( n7.parentof(n8) )
-# assert_false( n8.parentof(n7) )
-#
-# assert_true( n9.has_common_ancestor(n8) )
-# # cause it's not in the right branch
-# assert_false( n8.has_common_ancestor(n9) )
-#
-#
-#@temprepo('branched2')
-#def test_merge_personal_to_default(library):
-# main = library.shelf(2)
-# print main
-#
-# local = library.shelf(7)
-# print local
-#
-# document = library.document("ala", "admin")
-# shared = document.shared()
-# assert_true( shared is None )
-# document.share("Here is my copy!")
-#
-# assert_equal( document.shelf(), local) # local didn't change
-#
-# shared = document.shared()
-# assert_true( shared is not None )
-#
-# print library.shelf()
-#
-# new_main = shared.shelf()
-# assert_not_equal( new_main, main) # main has new revision
-#
-# # check for parents
-# assert_true( main.parentof(new_main) )
-# assert_true( local.parentof(new_main) )
-#
-#@temprepo('clean')
-#def test_create_branch(library):
-# tester_cab = library.cabinet("anotherone", "tester", create=True)
-# assert_equal( list(tester_cab.parts()), ['anotherone'])
-
fnp / redakcja.git / commitdiff