import mercurial
from mercurial import localrepo as hglrepo
from mercurial import ui as hgui
-from mercurial.node import hex as to_hex
+from mercurial.node import nullid
import re
import wlrepo
lock = self._lock(True)
try:
if self._has_branch(bname):
- return MercurialCabinet(self, bname, doc=docid, user=user)
+ return MercurialCabinet(self, doc=docid, user=user)
if not create:
raise wlrepo.CabinetNotFound(bname)
# check if the docid exists in the main cabinet
needs_touch = not self._maincab.exists(docid)
- cab = MercurialCabinet(self, bname, doc=docid, user=user)
+ cab = MercurialCabinet(self, doc=docid, user=user)
name, fileid = cab._filename(None)
# create the branch
self._create_branch(bname, before_commit=cleanup_action)
- return MercurialCabinet(self, bname, doc=docid, user=user)
+ return MercurialCabinet(self, doc=docid, user=user)
finally:
lock.release()
self._checkout(self._branch_tip(branchname))
return branchname
- #
- # Merges
- #
-
+ def shelf(self, nodeid):
+ return MercurialShelf(self, self._changectx(nodeid))
#
class MercurialCabinet(wlrepo.Cabinet):
- def __init__(self, library, branchname, doc=None, user=None):
+ def __init__(self, library, branchname=None, doc=None, user=None):
if doc and user:
super(MercurialCabinet, self).__init__(library, doc=doc, user=user)
- else:
+ self._branchname = library._bname(user=user, docid=doc)
+ elif branchname:
super(MercurialCabinet, self).__init__(library, name=branchname)
-
- self._branchname = branchname
+ self._branchname = branchname
+ else:
+ raise ValueError("Provide either doc/user or branchname")
def shelf(self, selector=None):
- if selector is not None:
- raise NotImplementedException()
-
- return MercurialShelf(self, self._library._changectx(self._branchname))
+ return self._library.shelf(self._branchname)
def documents(self):
return self._execute_in_branch(action=lambda l, c: (e[1] for e in l._filelist()))
def retrieve(self, part=None, shelf=None):
name, fileid = self._filename(part)
+ print "Retrieving document %s from cab %s" % (name, self._name)
+
if fileid is None:
raise wlrepo.LibraryException("Can't retrieve main document from main cabinet.")
+
+ def retrieve_action(l,c):
+ if l._fileexists(fileid):
+ return MercurialDocument(c, name=name, fileid=fileid)
+ return None
- return self._execute_in_branch(lambda l, c: MercurialDocument(c, name, fileid))
+ return self._execute_in_branch(retrieve_action)
def create(self, name, initial_data):
name, fileid = self._filename(name)
def _filename(self, part):
part = self._library._sanitize_string(part)
- fileid = None
+ docid = None
if self._maindoc == '':
- if part is None: return [None, None]
- fileid = part
+ if part is None: rreeturn [None, None]
+ docid = part
else:
- fileid = self._maindoc + (('$' + part) if part else '')
+ docid = self._maindoc + (('$' + part) if part else '')
- return fileid, 'pub_' + fileid + '.xml'
+ return docid, 'pub_' + docid + '.xml'
def _fileopener(self):
return self._library._fileopener()
def _filectx(self, fileid):
return self._library._filectx(fileid, self._branchname)
+ def ismain(self):
+ return (self._library.main_cabinet == self)
+
class MercurialDocument(wlrepo.Document):
def __init__(self, cabinet, name, fileid):
super(MercurialDocument, self).__init__(cabinet, name=name)
self._opener = self._cabinet._fileopener()
self._fileid = fileid
- self._refresh()
+ self.refresh()
- def _refresh(self):
- self._filectx = self._cabinet._filectx(self._fileid)
- self._shelf = MercurialShelf(self, self._filectx.node())
+ def refresh(self):
+ self._filectx = self._cabinet._filectx(self._fileid)
def read(self):
return self._opener(self._filectx.path(), "r").read()
self.library._commit(self._fileid, message, user)
def update(self):
- if self._cabinet.is_main():
- return True # always up-to-date
-
- mdoc = self.library.document(self._fileid)
+ lock = self.library._lock()
+ try:
+ if self._cabinet.ismain():
+ return True # always up-to-date
- mshelf = mdoc.shelf()
- shelf = self.shelf()
+ user = self._cabinet.username or 'library'
+ mdoc = self.library.document(self._fileid)
- if not mshelf.ancestorof(shelf) and not shelf.parentof(mshelf):
- shelf.merge_with(mshelf)
+ mshelf = mdoc.shelf()
+ shelf = self.shelf()
- return rc.ALL_OK
+ if not mshelf.ancestorof(shelf) and not shelf.parentof(mshelf):
+ shelf.merge_with(mshelf, user=user)
- def share(self, message, user):
- if self._cabinet.is_main():
- return True # always shared
+ return True
+ finally:
+ lock.release()
- main = self.shared_version()
- local = self.shelf()
+ def share(self, message):
+ lock = self.library._lock()
+ try:
+ print "sharing from", self._cabinet, self._cabinet.username
+
+ if self._cabinet.ismain():
+ return True # always shared
- no_changes = True
+ if self._cabinet.username is None:
+ raise ValueError("Can only share documents from personal cabinets.")
+
+ user = self._cabinet.username
- # Case 1:
- # * local
- # |
- # * <- can also be here!
- # /|
- # / |
- # main * *
- # | |
- # The local branch has been recently updated,
- # so we don't need to update yet again, but we need to
- # merge down to default branch, even if there was
- # no commit's since last update
+ main = self.shared().shelf()
+ local = self.shelf()
- if main.ancestorof(local):
- main.merge_with(local, user=user, message=message)
- no_changes = False
-
- # Case 2:
- #
- # main * * local
- # |\ |
- # | \|
- # | *
- # | |
- #
- # Default has no changes, to update from this branch
- # since the last merge of local to default.
- elif main.has_common_ancestor(local):
- if not local.parentof(main):
+ no_changes = True
+
+ # Case 1:
+ # * local
+ # |
+ # * <- can also be here!
+ # /|
+ # / |
+ # main * *
+ # | |
+ # The local branch has been recently updated,
+ # so we don't need to update yet again, but we need to
+ # merge down to default branch, even if there was
+ # no commit's since last update
+
+ if main.ancestorof(local):
main.merge_with(local, user=user, message=message)
no_changes = False
- # Case 3:
- # main *
- # |
- # * <- this case overlaps with previos one
- # |\
- # | \
- # | * local
- # | |
- #
- # There was a recent merge to the defaul branch and
- # no changes to local branch recently.
- #
- # Use the fact, that user is prepared to see changes, to
- # update his branch if there are any
- elif local.ancestorof(main):
- if not local.parentof(main):
+ # Case 2:
+ #
+ # main * * local
+ # |\ |
+ # | \|
+ # | *
+ # | |
+ #
+ # Default has no changes, to update from this branch
+ # since the last merge of local to default.
+ elif local.has_common_ancestor(main):
+ if not local.parentof(main):
+ main.merge_with(local, user=user, message=message)
+ no_changes = False
+
+ # Case 3:
+ # main *
+ # |
+ # * <- this case overlaps with previos one
+ # |\
+ # | \
+ # | * local
+ # | |
+ #
+ # There was a recent merge to the defaul branch and
+ # no changes to local branch recently.
+ #
+ # Use the fact, that user is prepared to see changes, to
+ # update his branch if there are any
+ elif local.ancestorof(main):
+ if not local.parentof(main):
+ local.merge_with(main, user=user, message='Local branch update.')
+ no_changes = False
+ else:
local.merge_with(main, user=user, message='Local branch update.')
- no_changes = False
- else:
- local.merge_with(main, user=user, message='Local branch update.')
- self._refresh()
- local = self.shelf()
+ self._refresh()
+ local = self.shelf()
- main.merge_with(local, user=user, message=message)
+ main.merge_with(local, user=user, message=message)
+ finally:
+ lock.release()
def shared(self):
- return self.library.document(self._fileid)
+ return self.library.main_cabinet.retrieve(self._name)
+
+ def exists(self):
+ return self._cabinet.exists(self._fileid)
@property
def size(self):
return self._filectx.size()
-
def shelf(self):
- return self._shelf
+ return MercurialShelf(self.library, self._filectx.node())
@property
def last_modified(self):
def __str__(self):
return u"Document(%s->%s)" % (self._cabinet.name, self._name)
+ def __eq__(self, other):
+ return self._filectx == other._filectx
+
+
class MercurialShelf(wlrepo.Shelf):
- def __init__(self, cabinet, changectx):
- super(MercurialShelf, self).__init__(cabinet)
- self._changectx = changectx
+ def __init__(self, lib, changectx):
+ super(MercurialShelf, self).__init__(lib)
+
+ if isinstance(changectx, str):
+ self._changectx = lib._changectx(changectx)
+ else:
+ self._changectx = changectx
@property
def _rev(self):
def __str__(self):
return self._changectx.hex()
+ def __repr__(self):
+ return "MercurialShelf(%s)" % self._changectx.hex()
def ancestorof(self, other):
- pass
+ 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)
+ print a, self._changectx.branch(), a.branch()
+
+ return (a.branch() == self._changectx.branch())
+
+ def merge_with(self, other, user, message):
+ lock = self._library._lock(True)
+ try:
+ self._library._checkout(self._changectx.node())
+ self._library._merge(other._changectx.node())
+ finally:
+ lock.release()
+
+ def __eq__(self, other):
+ return self._changectx.node() == other._changectx.node()
+
class MergeStatus(object):
def __init__(self, mstatus):
shutil.rmtree(self.path, True)
pass
- def testOpening(self):
+ def test_opening(self):
library = MercurialLibrary(self.path + '/cleanrepo')
- def testMainCabinet(self):
+ def test_main_cabinett(self):
library = MercurialLibrary(self.path + '/cleanrepo')
mcab = library.main_cabinet
- assert_equal(mcab.maindoc_name(), '')
+ assert_equal(mcab.maindoc_name, '')
# @type mcab MercurialCabinet
doclist = mcab.documents()
assert_equal( list(doclist), ['valid_file'])
- def testReadDocument(self):
+ def test_read_document(self):
library = MercurialLibrary(self.path + '/testrepoI')
doc = library.main_cabinet.retrieve('valid_file')
assert_equal(doc.read().strip(), 'Ala ma kota')
- def testReadUTF8Document(self):
+ def test_read_UTF8_document(self):
library = MercurialLibrary(self.path + '/testrepoI')
doc = library.main_cabinet.retrieve('polish_file')
assert_equal(doc.read().strip(), u'Gąska!'.encode('utf-8'))
- def testWriteDocument(self):
+ def test_write_document(self):
library = MercurialLibrary(self.path + '/testrepoI')
doc = library.main_cabinet.retrieve('valid_file')
assert_equal(doc.read(), STRING)
- def testCreateDocument(self):
+ def test_create_document(self):
repopath = os.path.join(self.path, 'testrepoI')
library = MercurialLibrary(repopath)
- doc = library.main_cabinet.create("another_file")
- doc.write("Some text")
+ doc = library.main_cabinet.create("another_file", "Some text")
assert_equal( doc.read(), "Some text")
assert_true( os.path.isfile( os.path.join(repopath, "pub_another_file.xml")) )
- def testSwitchBranch(self):
+ def test_switch_branch(self):
library = MercurialLibrary(self.path + '/testrepoII')
tester_cab = library.cabinet("valid_file", "tester", create=False)
assert_equal( list(tester_cab.documents()), ['valid_file'])
@raises(wlrepo.CabinetNotFound)
- def testNoBranch(self):
+ def test_branch_not_found(self):
library = MercurialLibrary(self.path + '/testrepoII')
tester_cab = library.cabinet("ugh", "tester", create=False)
+ def test_no_branches(self):
+ library = MercurialLibrary(self.path + '/testrepoII')
+ 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) )
+
+
+ def test_once_branched(self):
+ library = MercurialLibrary(self.path + '/test3')
+
+ 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) )
+
+ def test_after_merge(self):
+ library = MercurialLibrary(self.path + '/test4')
+ 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) )
+
+
+ def test_after_merge_and_local_commit(self):
+ library = MercurialLibrary(self.path + '/test5b')
+ 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) )
+
+
+ def test_merge_personal_to_default(self):
+ library = MercurialLibrary(self.path + '/test3')
+
+ main = library.shelf(2)
+ local = library.shelf(7)
+
+ document = library.document("ala", "admin")
+ shared = document.shared()
+ print document, shared
+
+ document.share("Here is my copy!")
+
+ assert_equal( document.shelf(), local) # local didn't change
+
+
+ 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) )
+
+
def testCreateBranch(self):
repopath = os.path.join(self.path, 'testrepoII')
-
\ No newline at end of file
+
fnp / redakcja.git / commitdiff