--- /dev/null
+# -*- encoding: utf-8 -*-
+
+__author__= "Łukasz Rekucki"
+__date__ = "$2009-09-20 21:34:52$"
+__doc__ = "Micro-forms for the API."
+
+
+from django import forms
+
+class DocumentGetForm(forms.Form):
+ autocabinet = forms.BooleanField(required=False)
+
+
+class DocumentUploadForm(forms.Form):
+ ocr = forms.FileField(label='Source OCR file')
+ bookname = forms.RegexField(regex=r'[0-9\.\w_-]+', \
+ label='Publication name', help_text='Example: slowacki-beniowski')
+
+ generate_dc = forms.BooleanField(required=False, initial=True, label=u"Generate DublinCore template")
+
--- /dev/null
+from piston.handler import BaseHandler, AnonymousBaseHandler
+from piston.utils import rc, validate
+
+import settings
+import librarian
+import api.forms as forms
+from datetime import date
+
+from django.core.urlresolvers import reverse
+from wlrepo import MercurialLibrary, CabinetNotFound
+
+#
+# Document List Handlers
+#
+class BasicLibraryHandler(AnonymousBaseHandler):
+ allowed_methods = ('GET',)
+
+ def read(self, request):
+ """Return the list of documents."""
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+ cab = lib.main_cabinet
+
+ document_list = [{
+ 'url': reverse('document_view', args=[docid]),
+ 'name': docid } for docid in cab.documents() ]
+
+ return {
+ 'cabinet': cab.name,
+ 'latest_rev': cab.shelf(),
+ 'documents' : document_list }
+
+class LibraryHandler(BaseHandler):
+ allowed_methods = ('GET', 'POST')
+ anonymous = BasicLibraryHandler
+
+ def read(self, request):
+ """Return the list of documents."""
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+ cab = lib.main_cabinet
+
+ document_list = [{
+ 'url': reverse('document_view', args=[docid]),
+ 'name': docid } for docid in cab.documents() ]
+
+ return {
+ 'cabinet': cab.name,
+ 'latest_rev': cab.shelf(),
+ 'documents' : document_list }
+
+ def create(self, request):
+ """Create a new document."""
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+ cab = lib.main_cabinet
+
+ form = forms.DocumentUploadForm(request.POST, request.FILES)
+ if not form.is_valid():
+ return rc.BAD_REQUEST
+
+ f = request.FILES['ocr']
+ data = f.read().decode('utf-8')
+
+ if form.cleaned_data['generate_dc']:
+ data = librarian.wrap_text(data, unicode(date.today()))
+
+ doc = cab.create(form.cleaned_data['bookname'], initial_data=data)
+
+ return {
+ 'url': reverse('document_view', args=[doc.name]),
+ 'name': doc.name,
+ 'size': doc.size,
+ 'revision': doc.shelf() }
+
+#
+# Document Handlers
+#
+class BasicDocumentHandler(AnonymousBaseHandler):
+ allowed_methods = ('GET',)
+
+ def read(self, request, docid):
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+
+ opts = forms.DocumentGetForm(request.GET)
+ if not opts.is_valid():
+ return rc.BAD_REQUEST
+
+ document = lib.main_cabinet.retrieve(docid)
+
+ result = {
+ 'name': document.name,
+ 'size': document.size,
+ 'text_url': reverse('doctext_view', args=[docid]),
+ #'dc_url': reverse('docdc_view', docid=document.name),
+ #'parts_url': reverse('docparts_view', docid=document.name),
+ 'latest_rev': document.shelf(),
+ }
+
+ if request.GET.get('with_part', 'no') == 'yes':
+ result['parts'] = document.parts()
+
+ return result
+
+class DocumentHandler(BaseHandler):
+ allowed_methods = ('GET', 'PUT')
+ anonymous = BasicDocumentHandler
+
+ def read(self, request, docid):
+ """Read document's meta data"""
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+
+ opts = forms.DocumentGetForm(request.GET)
+ if not opts.is_valid():
+ return rc.BAD_REQUEST
+
+ document = lib.cabinet(docid, request.user.username, \
+ create=opts.cleaned_data['autocabinet'] ).retrieve()
+
+ shared = lib.main_cabinet.retrieve(docid)
+
+ result = {
+ 'name': document.name,
+ 'size': document.size,
+ 'text_url': reverse('doctext_view', args=[docid]),
+ 'dc_url': reverse('docdc_view', args=[docid]),
+ 'parts_url': reverse('docparts_view', args=[docid]),
+ 'latest_rev': document.shelf(),
+ 'latest_shared_rev': shared.shelf(),
+ #'shared': lib.isparentof(document, shared),
+ #'up_to_date': lib.isparentof(shared, document),
+ }
+
+ if request.GET.get('with_part', 'no') == 'yes':
+ result['parts'] = document.parts()
+
+ return result
+
+#
+# Document Text View
+#
+class DocumentTextHandler(BaseHandler):
+ allowed_methods = ('GET', 'PUT')
+
+ def read(self, request, docid):
+ """Read document as raw text"""
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+ try:
+ return lib.document(docid, request.user.username).read()
+ except CabinetNotFound:
+ return rc.NOT_HERE
+
+ def update(self, request, docid):
+ lib = MercurialLibrary(path=settings.REPOSITORY_PATH)
+ try:
+ data = request.PUT['contents']
+ lib.document(docid, request.user.username).write(data)
+ return rc.ALL_OK
+ except (CabinetNotFound, KeyError):
+ return rc.NOT_HERE
\ No newline at end of file
--- /dev/null
+from django.db import models
+
+# Create your models here.
--- /dev/null
+"""
+This file demonstrates two different styles of tests (one doctest and one
+unittest). These will both pass when you run "manage.py test".
+
+Replace these with more appropriate tests for your application.
+"""
+
+from django.test import TestCase
+
+class SimpleTest(TestCase):
+ def test_basic_addition(self):
+ """
+ Tests that 1 + 1 always equals 2.
+ """
+ self.failUnlessEqual(1 + 1, 2)
+
+__test__ = {"doctest": """
+Another way to test that 1 + 1 is equal to 2.
+
+>>> 1 + 1 == 2
+True
+"""}
+
--- /dev/null
+"""
+This file demonstrates two different styles of tests (one doctest and one
+unittest). These will both pass when you run "manage.py test".
+
+Replace these with more appropriate tests for your application.
+"""
+
+from django.test import TestCase
+from django.test.client import Client
+from django.core.urlresolvers import reverse
+
+from django.utils import simplejson as json
+
+from django.contrib.auth.models import User
+
+import settings
+from os.path import join, dirname
+from StringIO import StringIO
+import shutil
+import tempfile
+
+REPO_TEMPLATES = join(dirname(__file__), 'data')
+
+def temprepo(name):
+ def decorator(func):
+ def decorated(self, *args, **kwargs):
+ clean = False
+ try:
+ temp = tempfile.mkdtemp("", "testdir_" )
+ shutil.copytree(join(REPO_TEMPLATES, name), join(temp, 'repo'), False)
+ settings.REPOSITORY_PATH = join(temp, 'repo')
+ func(self, *args, **kwargs)
+ clean = True
+ finally:
+ if not clean and self.response:
+ print "RESULT", func.__name__, ">>>"
+ print self.response
+ print "<<<"
+
+ shutil.rmtree(temp, True)
+ settings.REPOSITORY_PATH = ''
+
+ return decorated
+ return decorator
+
+
+class SimpleTest(TestCase):
+
+ def setUp(self):
+ self.response = None
+ u = User.objects.create_user('admin', 'test@localhost', 'admin')
+ u.save()
+
+ @temprepo('empty')
+ def test_documents_get_anonymous(self):
+ self.response = self.client.get( reverse("document_list_view") )
+ self.assert_json_response({
+ u'latest_rev': u'e56b2a7e06a97d7c3697fc4295974e0f20a66190',
+ u'documents': [],
+ u'cabinet': u'default',
+ }, exclude=['latest_shared_rev'])
+
+ @temprepo('empty')
+ def test_documents_get_with_login(self):
+ self.assertTrue(self.client.login(username='admin', password='admin'))
+
+ self.response = self.client.get( reverse("document_list_view") )
+ self.assert_json_response({
+ u'latest_rev': u'e56b2a7e06a97d7c3697fc4295974e0f20a66190',
+ u'documents': [],
+ u'cabinet': u'default',
+ })
+
+ @temprepo('empty')
+ def test_documents_post(self):
+ self.assertTrue(self.client.login(username='admin', password='admin'))
+
+ infile = tempfile.NamedTemporaryFile("w+")
+ infile.write('0123456789')
+ infile.flush()
+ infile.seek(0)
+
+ self.response = self.client.post( reverse("document_list_view"),
+ data = {
+ 'bookname': 'testbook',
+ 'ocr': infile,
+ 'generate_dc': False,
+ })
+
+ infile.close()
+
+ result = self.assert_json_response({
+ 'url': reverse('document_view', args=['testbook']),
+ 'name': 'testbook',
+ 'size': 10,
+ # can't test revision number, 'cause it's random
+ },)
+
+ @temprepo('empty')
+ def test_document_creation(self):
+ self.assertTrue(self.client.login(username='admin', password='admin'))
+
+ infile = tempfile.NamedTemporaryFile("w+")
+ infile.write('012340123456789')
+ infile.flush()
+ infile.seek(0)
+
+ self.response = self.client.post( reverse("document_list_view"),
+ data = {
+ 'bookname': 'testbook',
+ 'ocr': infile,
+ 'generate_dc': False,
+ })
+
+ r = self.assert_json_response({
+ 'url': reverse('document_view', args=['testbook']),
+ 'name': 'testbook',
+ 'size': 15,
+ # can't test revision number, 'cause it's random
+ })
+
+ created_rev = r['revision']
+
+ self.response = self.client.get( \
+ reverse("document_view", args=["testbook"])+'?autocabinet=true' )
+
+ result = self.assert_json_response({
+ u'latest_shared_rev': created_rev,
+ u'size': 15,
+ })
+
+
+ @temprepo('testone')
+ def test_document_meta_get_with_login(self):
+ self.assertTrue(self.client.login(username='admin', password='admin'))
+
+ self.response = self.client.get( reverse("document_list_view") )
+ self.assert_json_response({
+ u'latest_rev': u'f94a263812dbe46a3a13d5209bb119988d0078d5',
+ u'documents': [{u'url': u'/api/documents/testfile', u'name': u'testfile'}],
+ u'cabinet': u'default',
+ })
+
+ self.response = self.client.get( \
+ reverse("document_view", args=['testfile'])+'?autocabinet=true' )
+
+ self.assert_json_response({
+ u'latest_shared_rev': u'f94a263812dbe46a3a13d5209bb119988d0078d5',
+ u'text_url': reverse("doctext_view", args=[u'testfile']),
+ u'dc_url': reverse("docdc_view", args=[u'testfile']),
+ u'parts_url': reverse("docparts_view", args=[u'testfile']),
+ u'name': u'testfile',
+ u'size': 20,
+ })
+
+
+ @temprepo('test2')
+ def test_document_text_with_login(self):
+ self.assertTrue(self.client.login(username='admin', password='admin'))
+
+ self.response = self.client.get( \
+ reverse("doctext_view", args=['testfile']) )
+
+ self.assertEqual(self.response.status_code, 200)
+ self.assertEqual(self.response.content, "Test file contents.\n")
+
+
+ @temprepo('test2')
+ def test_document_text_update(self):
+ self.assertTrue(self.client.login(username='admin', password='admin'))
+ TEXT = u"Ala ma kota i psa"
+
+ self.response = self.client.put( \
+ reverse("doctext_view", args=['testfile']), {'contents': TEXT })
+ self.assertEqual(self.response.status_code, 200)
+
+ self.response = self.client.get( \
+ reverse("doctext_view", args=['testfile']) )
+ self.assertEqual(self.response.status_code, 200)
+ self.assertEqual(self.response.content, TEXT)
+
+ def assert_json_response(self, must_have={}, exclude=[]):
+ self.assertEqual(self.response.status_code, 200)
+ result = json.loads(self.response.content)
+
+ for (k,v) in must_have.items():
+ self.assertTrue(result.has_key(k), "Required field '%s' missing in response." % k)
+ self.assertEqual(result[k], v)
+
+ if exclude is True:
+ for (k,v) in result.items():
+ self.assertTrue(must_have.has_key(k))
+ self.assertEqual(must_have[k], v)
+
+ for key in exclude:
+ self.assertFalse(result.has_key(key))
+
+ return result
+
\ No newline at end of file
--- /dev/null
+e56b2a7e06a97d7c3697fc4295974e0f20a66190 0
+e56b2a7e06a97d7c3697fc4295974e0f20a66190 default
--- /dev/null
+revlogv1
+store
+fncache
--- /dev/null
+data/.library.i
--- /dev/null
+default
\ No newline at end of file
--- /dev/null
+93618f13c33aab5a3d049e0860cd3d21a457cd7d 2
+f94a263812dbe46a3a13d5209bb119988d0078d5 default
+93618f13c33aab5a3d049e0860cd3d21a457cd7d personal_admin_file_testfile
--- /dev/null
+revlogv1
+store
+fncache
--- /dev/null
+data/.library.i
+data/pub_testfile.xml.i
--- /dev/null
+personal_admin_file_testfile
\ No newline at end of file
--- /dev/null
+Test file contents.
--- /dev/null
+f94a263812dbe46a3a13d5209bb119988d0078d5 1
+f94a263812dbe46a3a13d5209bb119988d0078d5 default
--- /dev/null
+revlogv1
+store
+fncache
--- /dev/null
+data/.library.i
+data/pub_testfile.xml.i
--- /dev/null
+default
\ No newline at end of file
--- /dev/null
+Test file contents.
--- /dev/null
+__author__="lreqc"
+__date__ ="$2009-09-17 16:16:54$"
+
+from django.conf.urls.defaults import *
+from piston.resource import Resource
+
+from api.handlers import *
+from api.utils import TextEmitter, DjangoAuth
+
+authdata = {'authentication': DjangoAuth()}
+
+FORMAT_EXT = r"\.(?P<emitter_format>xml|json|yaml|django)$"
+
+library_resource = Resource(LibraryHandler, **authdata)
+document_resource = Resource(DocumentHandler, **authdata)
+document_text_resource = Resource(DocumentTextHandler, **authdata)
+
+urlpatterns = patterns('',
+# url(r'^hello$', hello_resource, {'emitter_format': 'json'}),
+# url(r'^hello\.(?P<emitter_format>.+)$', hello_resource),
+
+ # Documents
+ url(r'^documents$', library_resource, {'emitter_format': 'json'},
+ name="document_list_view"),
+
+ url(r'^documents/(?P<docid>[^/]+)'+FORMAT_EXT,
+ document_resource, name="document_view_withformat"),
+
+ url(r'^documents/(?P<docid>[^/]+)$',
+ document_resource, {'emitter_format': 'json'},
+ name="document_view"),
+
+ url(r'^documents/(?P<docid>[^/]+)/text$',
+ document_text_resource, {'emitter_format': 'rawxml'},
+ name="doctext_view"),
+
+ url(r'^documents/(?P<docid>[^/]+)/dc$',
+ document_resource, {'emitter_format': 'json'},
+ name="docdc_view"),
+
+ url(r'^documents/(?P<docid>[^/]+)/parts$',
+ document_resource, {'emitter_format': 'json'},
+ name="docparts_view"),
+
+ # url(r'^posts/(?P<post_slug>[^/]+)/$', blogpost_resource),
+ # url(r'^other/(?P<username>[^/]+)/(?P<data>.+)/$', arbitrary_resource),
+)
+
--- /dev/null
+# -*- encoding: utf-8 -*-
+
+__author__= "Łukasz Rekucki"
+__date__ = "$2009-09-20 21:48:03$"
+__doc__ = "Module documentation."
+
+
+from piston.emitters import Emitter
+from piston.utils import rc
+
+class TextEmitter(Emitter):
+ def render(self, request):
+ return unicode(self.construct())
+
+Emitter.register('text', TextEmitter, 'text/plain; charset=utf-8')
+Emitter.register('rawxml', TextEmitter, 'application/xml; charset=utf-8')
+
+
+class DjangoAuth(object):
+
+ def is_authenticated(self, request):
+ return request.user.is_authenticated()
+
+ def challenge(self):
+ return rc.FORBIDDEN
\ No newline at end of file
--- /dev/null
+# Create your views here.
--- /dev/null
+PYTYONPATH=./lib:$PYTHONPATH nosetests --detailed-errors --with-doctest --with-coverage --cover-package=wlrepo
--- /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 main_cabinet(self):
+ """Return the "main" cabinet of the library."""
+ pass
+
+ def cabinets(self):
+ """List all cabinets in the library."""
+ pass
+
+ def cabinet(self, document, user, create=False):
+ """Open a cabinet belonging to the _user_ for a given _document_.
+ If the _document_ is actually a sub-document, it's parent's cabinet is
+ opened istead.
+
+ If the cabinet doesn't exists and create is False (the default), a
+ CabinetNotFound exception is raised.
+
+ If create is True, a new cabinet is created if it doesn't exist yet."""
+ pass
+
+
+class Cabinet(object):
+
+ def __init__(self, library, name=None, doc=None, user=None):
+ self._library = library
+ if name:
+ self._name = name
+ self._maindoc = ''
+ elif doc and user:
+ self._name = user + ':' + doc
+ self._maindoc = doc
+ else:
+ raise ValueError("You must provide either name or doc and user.")
+
+ def documents(self):
+ """Lists all documents and sub-documents in this cabinet."""
+ pass
+
+ def retrieve(self, parts=None, shelve=None):
+ """Retrieve a document from a given shelve in the cabinet. If no
+ part is given, the main document is retrieved. If no shelve is given,
+ the top-most shelve is used.
+
+ If parts is a list, all the given parts are retrieved atomicly. Use None
+ as the name for the main document"""
+ pass
+
+ def create(self, name):
+ """Create a new sub-document in the cabinet with the given name."""
+ pass
+
+ def maindoc_name(self):
+ return self._maindoc
+
+
+ @property
+ def library(self):
+ return self._library
+
+ @property
+ def name(self):
+ return self._name
+
+class Document(object):
+ def __init__(self, cabinet, name):
+ self._cabinet = cabinet
+ self._name = name
+
+ def read(self):
+ pass
+
+ def write(self, data):
+ pass
+
+ @property
+ def cabinet(self):
+ return self._cabinet
+
+ @property
+ def library(self):
+ return self._cabinet.library
+
+
+#
+# Exception classes
+#
+
+class LibraryException(Exception):
+
+ def __init__(self, msg, cause=None):
+ Exception.__init__(self, msg)
+ self.cause = cause
+
+class CabinetNotFound(LibraryException):
+ pass
+
+
+# import backends to local namespace
+from backend_mercurial import MercurialLibrary
\ No newline at end of file
--- /dev/null
+# -*- encoding: utf-8 -*-
+__author__ = "Łukasz Rekucki"
+__date__ = "$2009-09-18 10:49:24$"
+
+__doc__ = """RAL implementation over Mercurial"""
+
+import mercurial
+from mercurial import localrepo as hglrepo
+from mercurial import ui as hgui
+import re
+import wlrepo
+
+FILTER = re.compile(r"^pub_(.+)\.xml$", re.UNICODE)
+
+def default_filter(name):
+ m = FILTER.match(name)
+ if m is not None:
+ return name, m.group(1)
+ return None
+
+class MercurialLibrary(wlrepo.Library):
+
+ def __init__(self, path, maincabinet="default", ** kwargs):
+ super(wlrepo.Library, self).__init__( ** kwargs)
+
+ 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))
+
+ maincabinet = self._sanitize_string(maincabinet)
+
+ 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['create']:
+ 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)
+
+ # fetch the main cabinet
+ lock = self._hgrepo.lock()
+ try:
+ btags = self._hgrepo.branchtags()
+
+ if not self._has_branch(maincabinet):
+ raise wlrepo.LibraryException("[HGLibrary] No branch named '%s' to init main cabinet" % maincabinet)
+
+ self._maincab = MercurialCabinet(self, maincabinet)
+ finally:
+ lock.release()
+
+ @property
+ def main_cabinet(self):
+ return self._maincab
+
+ def cabinet(self, docid, user, create=False):
+ bname = self._bname(user, docid)
+
+ lock = self._lock(True)
+ try:
+ if self._has_branch(bname):
+ return MercurialCabinet(self, bname, doc=docid, user=user)
+
+ if not create:
+ raise wlrepo.CabinetNotFound(docid, user)
+
+ # check if the docid exists in the main cabinet
+ needs_touch = not self._maincab.exists(docid)
+ print "Creating branch: ", needs_touch
+ cab = MercurialCabinet(self, bname, doc=docid, user=user)
+
+ fileid = cab._fileid(None)
+
+ def cleanup_action(l):
+ if needs_touch:
+ print "Touch for file", docid
+ l._fileopener()(fileid, "w").write('')
+ l._fileadd(fileid)
+
+ garbage = [fid for (fid, did) in l._filelist() if not did.startswith(docid)]
+ print "Garbage: ", garbage
+ l._filesrm(garbage)
+
+ # create the branch
+ self._create_branch(bname, before_commit=cleanup_action)
+ return MercurialCabinet(self, bname, doc=docid, user=user)
+ finally:
+ lock.release()
+
+ #
+ # 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="library"):
+ return self._hgrepo.commit(\
+ text=self._sanitize_string(message), \
+ user=self._sanitize_string(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 _filelist(self, filter=default_filter):
+ for name in self._hgrepo[None]:
+ result = filter(name)
+ if result is None: continue
+
+ yield result
+
+ def _fileopener(self):
+ return self._hgrepo.wopener
+
+ #
+ # 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):
+ name = self._sanitize_string(name)
+
+ if self._has_branch(name): return # just exit
+
+ if parent is None:
+ parent = self._maincab
+
+ parentrev = parent._hgtip()
+
+ self._checkout(parentrev)
+ self._hgrepo.dirstate.setbranch(name)
+
+ if before_commit: before_commit(self)
+
+ print "commiting"
+ self._commit("[AUTO] Initial commit for branch '%s'." % name, user='library')
+
+ # revert back to main
+ self._checkout(self._maincab._hgtip())
+ return self._branch_tip(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
+
+ #
+ # Merges
+ #
+
+
+
+ #
+ # Utils
+ #
+
+ @staticmethod
+ def _sanitize_string(s):
+ if isinstance(s, unicode): #
+ return s.encode('utf-8')
+ else: # it's a string, so we have no idea what encoding it is
+ return s
+
+class MercurialCabinet(wlrepo.Cabinet):
+
+ def __init__(self, library, branchname, doc=None, user=None):
+ if doc and user:
+ super(MercurialCabinet, self).__init__(library, doc=doc, user=user)
+ else:
+ super(MercurialCabinet, self).__init__(library, name=branchname)
+
+ self._branchname = 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, shelve=None):
+ fileid = self._fileid(part)
+
+ if fileid is None:
+ raise wlrepo.LibraryException("Can't retrieve main document from main cabinet.")
+
+ return self._execute_in_branch(lambda l, c: MercurialDocument(c, fileid))
+
+ def create(self, name, initial_data=''):
+ fileid = self._fileid(name)
+
+ if name is None:
+ raise ValueError("Can't create main doc for maincabinet.")
+
+ def create_action(l, c):
+ if l._fileexists(fileid):
+ raise wlrepo.LibraryException("Can't create document '%s' in cabinet '%s' - it already exists" % (fileid, c.name))
+
+ fd = l._fileopener()(fileid, "w")
+ fd.write(initial_data)
+ l._fileadd(fileid)
+ l._commit("File '%d' created.")
+
+ return MercurialDocument(c, fileid)
+
+ return self._execute_in_branch(create_action)
+
+ def exists(self, part=None, shelve=None):
+ fileid = self._fileid(part)
+
+ if fileid is None: return false
+ return self._execute_in_branch(lambda l, c: l._fileexists(fileid))
+
+ def _execute_in_branch(self, action, write=False):
+ def switch_action(library):
+ old = library._switch_to_branch(self._branchname)
+ try:
+ return action(library, self)
+ finally:
+ library._switch_to_branch(old)
+
+ return self._library._transaction(write_mode=write, action=switch_action)
+
+ def _fileid(self, part):
+ fileid = None
+
+ if self._maindoc == '':
+ if part is None: return None
+ fileid = part
+ else:
+ fileid = self._maindoc + (('$' + part) if part else '')
+
+ return 'pub_' + fileid + '.xml'
+
+ def _fileopener(self):
+ return self._library._fileopener()
+
+ def _hgtip(self):
+ return self._library._branch_tip(self._branchname)
+
+class MercurialDocument(wlrepo.Document):
+
+ def __init__(self, cabinet, fileid):
+ super(MercurialDocument, self).__init__(cabinet, fileid)
+ self._opener = self._cabinet._fileopener()
+
+ def read(self):
+ return self._opener(self._name, "r").read()
+
+ def write(self, data):
+ return self._opener(self._name, "w").write(data)
+
+
+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))
+
+
+__all__ = ["MercurialLibrary", "MercurialCabinet", "MercurialDocument"]
\ No newline at end of file
--- /dev/null
+# -*- encoding: utf-8 -*-
+
+__author__= "Łukasz Rekucki"
+__date__ = "$2009-09-18 14:42:53$"
+__doc__ = "Test package for WL-RAL"
+
+
--- /dev/null
+revlogv1
+store
+fncache
--- /dev/null
+data/.hgignore.i
+data/ignored_file.i
+data/pub_valid_file.xml.i
--- /dev/null
+default
\ No newline at end of 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
+default
\ No newline at end of file
--- /dev/null
+Ala ma kota
--- /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
+# -*- encoding: utf-8 -*-
+
+__author__= "Łukasz Rekucki"
+__date__ = "$2009-09-18 14:43:27$"
+__doc__ = "Tests for RAL mercurial backend."
+
+from nose.tools import *
+
+import wlrepo
+from wlrepo import MercurialLibrary
+from wlrepo.backend_mercurial import *
+
+import os, os.path, tempfile
+import shutil
+
+REPO_TEMPLATES = os.path.join( os.path.dirname(__file__), 'data/repos')
+ROOT_PATH = None
+
+class testBasicLibrary(object):
+
+ def setUp(self):
+ self.path = tempfile.mkdtemp("", "testdir_" )
+ print self.path
+ for subdir in os.listdir(REPO_TEMPLATES):
+ shutil.copytree(REPO_TEMPLATES + '/' + subdir, self.path + '/' + subdir, False)
+
+ def tearDown(self):
+ if self.path is not None:
+ shutil.rmtree(self.path, True)
+ pass
+
+ def testOpening(self):
+ library = MercurialLibrary(self.path + '/cleanrepo')
+
+ def testMainCabinet(self):
+ library = MercurialLibrary(self.path + '/cleanrepo')
+
+ mcab = library.main_cabinet
+ assert_equal(mcab.maindoc_name(), '')
+
+ # @type mcab MercurialCabinet
+ doclist = mcab.documents()
+ assert_equal( list(doclist), ['valid_file'])
+
+
+ def testReadDocument(self):
+ library = MercurialLibrary(self.path + '/testrepoI')
+ doc = library.main_cabinet.retrieve('valid_file')
+
+ assert_equal(doc.read().strip(), 'Ala ma kota')
+
+ def testReadUTF8Document(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):
+ library = MercurialLibrary(self.path + '/testrepoI')
+ doc = library.main_cabinet.retrieve('valid_file')
+
+ assert_equal(doc.read().strip(), 'Ala ma kota')
+
+ STRING = u'Gąski lubią pływać!\n'.encode('utf-8')
+ doc.write(STRING)
+
+ assert_equal(doc.read(), STRING)
+
+ def testCreateDocument(self):
+ repopath = os.path.join(self.path, 'testrepoI')
+
+ library = MercurialLibrary(repopath)
+ doc = library.main_cabinet.create("another_file")
+ doc.write("Some text")
+ assert_equal( doc.read(), "Some text")
+ assert_true( os.path.isfile( os.path.join(repopath, "pub_another_file.xml")) )
+
+ def testSwitchBranch(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):
+ library = MercurialLibrary(self.path + '/testrepoII')
+ tester_cab = library.cabinet("ugh", "tester", create=False)
+
+
+ def testCreateBranch(self):
+ repopath = os.path.join(self.path, 'testrepoII')
+ library = MercurialLibrary(repopath)
+
+ tester_cab = library.cabinet("anotherone", "tester", create=True)
+ assert_equal( list(tester_cab.documents()), ['anotherone'])
+
+
+
+
\ No newline at end of file
fnp / redakcja.git / commitdiff