Merge pull request #147 from gazpachoking/page_class

Create WikiPage class to encapsulate wiki page functions
This commit is contained in:
Matthew Scragg 2016-07-07 08:29:43 -05:00 committed by GitHub
commit 93ba987a3a
6 changed files with 266 additions and 248 deletions

View file

@ -1,17 +1,20 @@
import os
import shutil
import tempfile
from flask.ext.testing import TestCase
from realms.lib.util import random_string
from realms import create_app
from subprocess import call
class BaseTest(TestCase):
def create_app(self):
self.tempdir = tempfile.mkdtemp()
app = create_app()
app.config['TESTING'] = True
app.config['PRESERVE_CONTEXT_ON_EXCEPTION'] = False
app.config['WIKI_PATH'] = '/tmp/%s' % random_string(12)
app.config['DB_URI'] = 'sqlite:////tmp/%s.db' % random_string(12)
app.config['WIKI_PATH'] = os.path.join(self.tempdir, random_string(12))
app.config['DB_URI'] = 'sqlite:///%s/%s.db' % (self.tempdir, random_string(12))
app.config.update(self.configure())
return app
@ -19,5 +22,4 @@ class BaseTest(TestCase):
return {}
def tearDown(self):
call(['rm', '-rf', self.app.config['WIKI_PATH']])
call(['rm', '-f', self.app.config['DB_URI'][10:]])
shutil.rmtree(self.tempdir)

View file

@ -31,9 +31,9 @@ def rebuild_index():
name = filename_to_cname(page['path'])
# TODO add email?
body = dict(name=name,
content=page['data'],
message=page['info']['message'],
username=page['info']['author'],
content=page.data,
message=page.info['message'],
username=page.info['author'],
updated_on=entry['mtime'],
created_on=entry['ctime'])
search.index_wiki(name, body)

View file

@ -47,7 +47,7 @@ class SimpleSearch(BaseSearch):
# this can be None, not sure how
if page:
res.append(dict(name=name, content=page['data']))
res.append(dict(name=name, content=page.data))
return res
def users(self, query):
@ -129,7 +129,7 @@ class WhooshSearch(BaseSearch):
res = []
for hit in results:
name = hit["path"]
page_data = g.current_wiki.get_page(name)["data"].decode("utf-8")
page_data = g.current_wiki.get_page(name).data.decode("utf-8")
content = hit.highlights('body', text=page_data)
res.append(dict(name=name, content=content))

View file

@ -5,8 +5,9 @@ import ghdiff
import gittle.utils
import yaml
from gittle import Gittle
from dulwich.object_store import tree_lookup_path
from dulwich.repo import NotGitRepository
from realms.lib.util import to_canonical, cname_to_filename, filename_to_cname
from realms.lib.util import cname_to_filename, filename_to_cname
from realms import cache
from realms.lib.hook import HookMixin
@ -39,145 +40,6 @@ class Wiki(HookMixin):
def __repr__(self):
return "Wiki: %s" % self.path
def _get_user(self, username, email):
if not username:
username = self.default_committer_name
if not email:
email = self.default_committer_email
return username, email
def _cache_key(self, name, sha='HEAD'):
return 'page/%s[%s]' % (name, sha)
def revert_page(self, name, commit_sha, message, username, email):
"""Revert page to passed commit sha1
:param name: Name of page to revert.
:param commit_sha: Commit Sha1 to revert to.
:param message: Commit message.
:param username: Committer name.
:param email: Committer email.
:return: Git commit sha1
"""
page = self.get_page(name, commit_sha)
if not page:
raise PageNotFound('Commit not found')
if not message:
commit_info = gittle.utils.git.commit_info(self.gittle[commit_sha.encode('latin-1')])
message = commit_info['message']
return self.write_page(name, page['data'], message=message, username=username, email=email)
def write_page(self, name, content, message=None, create=False, username=None, email=None):
"""Write page to git repo
:param name: Name of page.
:param content: Content of page.
:param message: Commit message.
:param create: Perform git add operation?
:param username: Commit Name.
:param email: Commit Email.
:return: Git commit sha1.
"""
cname = to_canonical(name)
filename = cname_to_filename(cname)
dirname = posixpath.join(self.path, posixpath.dirname(filename))
if not os.path.exists(dirname):
os.makedirs(dirname)
with open(self.path + "/" + filename, 'w') as f:
f.write(content)
if create:
self.gittle.add(filename)
if not message:
message = "Updated %s" % name
username, email = self._get_user(username, email)
ret = self.gittle.commit(name=username,
email=email,
message=message,
files=[filename])
cache.delete(self._cache_key(cname))
return ret
def rename_page(self, old_name, new_name, username=None, email=None, message=None):
"""Rename page.
:param old_name: Page that will be renamed.
:param new_name: New name of page.
:param username: Committer name
:param email: Committer email
:return: str -- Commit sha1
"""
old_filename, new_filename = map(cname_to_filename, [old_name, new_name])
if old_filename not in self.gittle.index:
# old doesn't exist
return None
if old_filename == new_filename:
return
if new_filename in self.gittle.index:
# file is being overwritten, but that is ok, it's git!
pass
username, email = self._get_user(username, email)
if not message:
message = "Moved %s to %s" % (old_name, new_name)
os.rename(os.path.join(self.path, old_filename), os.path.join(self.path, new_filename))
self.gittle.add(new_filename)
self.gittle.rm(old_filename)
commit = self.gittle.commit(name=username,
email=email,
message=message,
files=[old_filename, new_filename])
cache.delete_many(self._cache_key(old_name), self._cache_key(new_name))
return commit
def delete_page(self, name, username=None, email=None, message=None):
"""Delete page.
:param name: Page that will be deleted
:param username: Committer name
:param email: Committer email
:return: str -- Commit sha1
"""
username, email = self._get_user(username, email)
if not message:
message = "Deleted %s" % name
filename = cname_to_filename(name)
# gittle.rm won't actually remove the file, have to do it ourselves
os.remove(os.path.join(self.path, filename))
self.gittle.rm(filename)
commit = self.gittle.commit(name=username,
email=email,
message=message,
files=[filename])
cache.delete_many(self._cache_key(name))
return commit
def get_page(self, name, sha='HEAD'):
"""Get page data, partials, commit info.
@ -186,67 +48,7 @@ class Wiki(HookMixin):
:return: dict
"""
cached = cache.get(self._cache_key(name, sha))
if cached:
return cached
# commit = gittle.utils.git.commit_info(self.repo[sha])
filename = cname_to_filename(name).encode('utf8')
sha = sha.encode('latin-1')
try:
data = self.gittle.get_commit_files(sha, paths=[filename]).get(filename)
if not data:
return None
partials = {}
if data.get('data'):
meta = self.get_meta(data['data'])
if meta and 'import' in meta:
for partial_name in meta['import']:
partials[partial_name] = self.get_page(partial_name)
data['partials'] = partials
data['info'] = self.get_history(name, limit=1)[0]
cache.set(self._cache_key(name, sha), data)
return data
except KeyError:
# HEAD doesn't exist yet
return None
def get_meta(self, content):
"""Get metadata from page if any.
:param content: Page content
:return: dict
"""
if not content.startswith("---"):
return None
meta_end = re.search("\n(\.{3}|\-{3})", content)
if not meta_end:
return None
try:
return yaml.safe_load(content[0:meta_end.start()])
except Exception as e:
return {'error': e.message}
def compare(self, name, old_sha, new_sha):
"""Compare two revisions of the same page.
:param name: Name of page.
:param old_sha: Older sha.
:param new_sha: Newer sha.
:return: str - Raw markup with styles
"""
# TODO: This could be effectively done in the browser
old = self.get_page(name, sha=old_sha)
new = self.get_page(name, sha=new_sha)
return ghdiff.diff(old['data'], new['data'])
return WikiPage(name, self, sha=sha)
def get_index(self):
"""Get repo index of head.
@ -266,28 +68,56 @@ class Wiki(HookMixin):
return rv
def get_history(self, name, limit=100):
class WikiPage(object):
def __init__(self, name, wiki, sha='HEAD'):
self.name = name
self.filename = cname_to_filename(name)
self.sha = sha.encode('latin-1')
self.wiki = wiki
@property
def data(self):
cache_key = self._cache_key('data')
cached = cache.get(cache_key)
if cached:
return cached
data = self.wiki.gittle.get_commit_files(self.sha, paths=[self.filename]).get(self.filename).get('data')
cache.set(cache_key, data)
return data
@property
def info(self):
cache_key = self._cache_key('info')
cached = cache.get(cache_key)
if cached:
return cached
info = self.get_history(limit=1)[0]
cache.set(cache_key, info)
return info
def get_history(self, limit=100):
"""Get page history.
:param name: Name of page.
:param limit: Limit history size.
:return: list -- List of dicts
"""
if not len(self.repo.open_index()):
if not len(self.wiki.repo.open_index()):
# Index is empty, no commits
return []
file_path = cname_to_filename(name)
versions = []
walker = self.repo.get_walker(paths=[file_path], max_entries=limit)
walker = self.wiki.repo.get_walker(paths=[self.filename], max_entries=limit)
for entry in walker:
change_type = None
for change in entry.changes():
if change.old.path == file_path:
if change.old.path == self.filename:
change_type = change.type
elif change.new.path == file_path:
elif change.new.path == self.filename:
change_type = change.type
author_name, author_email = entry.commit.author.rstrip('>').split('<')
versions.append(dict(
@ -299,3 +129,193 @@ class Wiki(HookMixin):
type=change_type))
return versions
@property
def partials(self):
data = self.data
if not data:
return {}
partials = {}
meta = self._get_meta(data)
if meta and 'import' in meta:
for partial_name in meta['import']:
partials[partial_name] = self.wiki.get_page(partial_name, sha=self.sha)
return partials
@staticmethod
def _get_meta(content):
"""Get metadata from page if any.
:param content: Page content
:return: dict
"""
if not content.startswith("---"):
return None
meta_end = re.search("\n(\.{3}|\-{3})", content)
if not meta_end:
return None
try:
return yaml.safe_load(content[0:meta_end.start()])
except Exception as e:
return {'error': e.message}
def _cache_key(self, property):
return 'page/{0}[{1}].{2}'.format(self.name, self.sha, property)
def _get_user(self, username, email):
if not username:
username = self.wiki.default_committer_name
if not email:
email = self.wiki.default_committer_email
return username, email
def _clear_cache(self):
cache.delete_many(self._cache_key(p) for p in ['data', 'info'])
def delete(self, username=None, email=None, message=None):
"""Delete page.
:param username: Committer name
:param email: Committer email
:return: str -- Commit sha1
"""
username, email = self._get_user(username, email)
if not message:
message = "Deleted %s" % self.name
# gittle.rm won't actually remove the file, have to do it ourselves
os.remove(os.path.join(self.wiki.path, self.filename))
self.wiki.gittle.rm(self.filename)
commit = self.wiki.gittle.commit(name=username,
email=email,
message=message,
files=[self.filename])
self._clear_cache()
return commit
def rename(self, new_name, username=None, email=None, message=None):
"""Rename page.
:param new_name: New name of page.
:param username: Committer name
:param email: Committer email
:return: str -- Commit sha1
"""
assert self.sha == 'HEAD'
old_filename, new_filename = self.filename, cname_to_filename(new_name)
if old_filename not in self.wiki.gittle.index:
# old doesn't exist
return None
elif old_filename == new_filename:
return None
else:
# file is being overwritten, but that is ok, it's git!
pass
username, email = self._get_user(username, email)
if not message:
message = "Moved %s to %s" % (self.name, new_name)
os.rename(os.path.join(self.wiki.path, old_filename), os.path.join(self.wiki.path, new_filename))
self.wiki.gittle.add(new_filename)
self.wiki.gittle.rm(old_filename)
commit = self.wiki.gittle.commit(name=username,
email=email,
message=message,
files=[old_filename, new_filename])
self._clear_cache()
self.name = new_name
self.filename = new_filename
# We need to clear the cache for the new name as well as the old
self._clear_cache()
return commit
def write(self, content, message=None, create=False, username=None, email=None):
"""Write page to git repo
:param content: Content of page.
:param message: Commit message.
:param create: Perform git add operation?
:param username: Commit Name.
:param email: Commit Email.
:return: Git commit sha1.
"""
assert self.sha == 'HEAD'
dirname = posixpath.join(self.wiki.path, posixpath.dirname(self.filename))
if not os.path.exists(dirname):
os.makedirs(dirname)
with open(self.wiki.path + "/" + self.filename, 'w') as f:
f.write(content)
if create:
self.wiki.gittle.add(self.filename)
if not message:
message = "Updated %s" % self.name
username, email = self._get_user(username, email)
ret = self.wiki.gittle.commit(name=username,
email=email,
message=message,
files=[self.filename])
self._clear_cache()
return ret
def revert(self, commit_sha, message, username, email):
"""Revert page to passed commit sha1
:param commit_sha: Commit Sha1 to revert to.
:param message: Commit message.
:param username: Committer name.
:param email: Committer email.
:return: Git commit sha1
"""
assert self.sha == 'HEAD'
new_page = self.wiki.get_page(self.name, commit_sha)
if not new_page:
raise PageNotFound('Commit not found')
if not message:
commit_info = gittle.utils.git.commit_info(self.wiki.gittle[commit_sha.encode('latin-1')])
message = commit_info['message']
return self.write(new_page.data, message=message, username=username, email=email)
def compare(self, old_sha):
"""Compare two revisions of the same page.
:param old_sha: Older sha.
:return: str - Raw markup with styles
"""
# TODO: This could be effectively done in the browser
old = self.wiki.get_page(self.name, sha=old_sha)
return ghdiff.diff(old.data, self.data)
def __nonzero__(self):
# Verify this file is in the tree for the given commit sha
try:
tree_lookup_path(self.wiki.repo.get_object, self.wiki.repo[self.sha].tree, self.filename)
except KeyError:
# We'll get a KeyError if self.sha isn't in the repo, or if self.filename isn't in the tree of our commit
return False
return True

View file

@ -41,8 +41,8 @@ class WikiTest(WikiBaseTest):
self.assert_200(rv)
self.assert_context('name', 'test')
eq_(self.get_context_variable('page')['info']['message'], 'test message')
eq_(self.get_context_variable('page')['data'], 'testing')
eq_(self.get_context_variable('page').info['message'], 'test message')
eq_(self.get_context_variable('page').data, 'testing')
def test_history(self):
self.assert_200(self.client.get(url_for('wiki.history', name='test')))
@ -68,7 +68,7 @@ class WikiTest(WikiBaseTest):
data = json.loads(rv1.data)
self.client.post(url_for('wiki.revert'), data=dict(name='test', commit=data['sha']))
self.client.get(url_for('wiki.page', name='test'))
eq_(self.get_context_variable('page')['data'], 'testing_old')
eq_(self.get_context_variable('page').data, 'testing_old')
self.assert_404(self.client.post(url_for('wiki.revert'), data=dict(name='test', commit='does not exist')))
self.app.config['WIKI_LOCKED_PAGES'] = ['test']

View file

@ -28,7 +28,7 @@ def compare(name, fsha, dots, lsha):
if current_app.config.get('PRIVATE_WIKI') and current_user.is_anonymous():
return current_app.login_manager.unauthorized()
diff = g.current_wiki.compare(name, fsha, lsha)
diff = g.current_wiki.get_page(name, sha=lsha).compare(fsha)
return render_template('wiki/compare.html',
name=name, diff=diff, old=fsha, new=lsha)
@ -47,11 +47,10 @@ def revert():
return dict(error=True, message="Page is locked"), 403
try:
sha = g.current_wiki.revert_page(cname,
commit,
message=message,
username=current_user.username,
email=current_user.email)
sha = g.current_wiki.get_page(cname).revert(commit,
message=message,
username=current_user.username,
email=current_user.email)
except PageNotFound as e:
return dict(error=True, message=e.message), 404
@ -66,7 +65,7 @@ def history(name):
if current_app.config.get('PRIVATE_WIKI') and current_user.is_anonymous():
return current_app.login_manager.unauthorized()
hist = g.current_wiki.get_history(name)
hist = g.current_wiki.get_page(name).get_history()
for item in hist:
item['gravatar'] = gravatar_url(item['author_email'])
return render_template('wiki/history.html', name=name, history=hist)
@ -85,10 +84,10 @@ def edit(name):
g.assets['js'].append('editor.js')
return render_template('wiki/edit.html',
name=cname,
content=page.get('data'),
info=page.get('info'),
sha=page.get('sha'),
partials=page.get('partials'))
content=page.data,
info=page.info,
sha=page.sha,
partials=page.partials)
@blueprint.route("/_create/", defaults={'name': None})
@ -167,12 +166,11 @@ def page_write(name):
if cname in current_app.config.get('WIKI_LOCKED_PAGES'):
return dict(error=True, message="Page is locked"), 403
sha = g.current_wiki.write_page(cname,
request.form['content'],
message=request.form['message'],
create=True,
username=current_user.username,
email=current_user.email)
sha = g.current_wiki.get_page(cname).write(request.form['content'],
message=request.form['message'],
create=True,
username=current_user.username,
email=current_user.email)
elif request.method == 'PUT':
edit_cname = to_canonical(request.form['name'])
@ -181,13 +179,12 @@ def page_write(name):
return dict(error=True, message="Page is locked"), 403
if edit_cname != cname:
g.current_wiki.rename_page(cname, edit_cname)
g.current_wiki.get_page(cname).rename(edit_cname)
sha = g.current_wiki.write_page(edit_cname,
request.form['content'],
message=request.form['message'],
username=current_user.username,
email=current_user.email)
sha = g.current_wiki.get_page(edit_cname).write(request.form['content'],
message=request.form['message'],
username=current_user.username,
email=current_user.email)
return dict(sha=sha)
@ -196,9 +193,8 @@ def page_write(name):
if cname in current_app.config.get('WIKI_LOCKED_PAGES'):
return dict(error=True, message="Page is locked"), 403
sha = g.current_wiki.delete_page(cname,
username=current_user.username,
email=current_user.email)
sha = g.current_wiki.get_page(cname).delete(username=current_user.username,
email=current_user.email)
return dict(sha=sha)
@ -216,6 +212,6 @@ def page(name):
data = g.current_wiki.get_page(cname)
if data:
return render_template('wiki/page.html', name=cname, page=data, partials=data.get('partials'))
return render_template('wiki/page.html', name=cname, page=data, partials=data.partials)
else:
return redirect(url_for('wiki.create', name=cname))