Commit 693d7ab4 authored by Pascal's avatar Pascal
Browse files

Merge branch 'master' of gitlab.math.univ-paris-diderot.fr:molin/lektor-commit

parents 5bf9484b 559b5ec4
# -*- coding: utf-8 -*-
import shlex
import subprocess
def git_decode(letter):
code = {
' ': 'unmodified',
'M': 'modified',
'A': 'added',
'D': 'deleted',
'U': 'unmerged',
'?': 'untracked'
}
return code.get(letter, None)
class PathStatus:
"""
File status as output by git status --porcelain
TODO: consider using porcelain=v2
"""
def __init__(self, line, number):
self.number = number
self.line = line
self.index = git_decode(line[0])
self.tree = git_decode(line[1])
self.path = line[3:]
def __repr__(self):
return self.line
class Status(list):
states = [ ('changed', 'modifiés non publiés'),
('saved', 'validés pour publication') ]
actions = [ ('save', 'valider'),
('skip', 'ignorer'),
('discard', 'abandonner') ]
def __init__(self, gitstatus):
files = [ PathStatus(l,i) for i,l in enumerate(gitstatus) ]
#super().__init__()
super().__init__(files)
class GitCommand(object):
"""
TODO: consider turning this class
into a function,
using __new__(*args)
instead of __call__(self, *args)
"""
git_cmd = []
def git_args(self, *args):
return []
def result(self, rv):
return rv.returncode == 0
def before(self):
pass
def check_before(self):
return True
def __call__(self, *args, **kwargs):
self.before()
assert self.check_before()
args = self.git_cmd + self.git_args(*args)
rv = subprocess.check_output(args)
return self.result(rv)
def instantiate(cls):
return cls()
@instantiate
class git_skip(GitCommand):
pass
@instantiate
class git_add(GitCommand):
git_cmd = ['git', 'add']
def git_args(self, path):
return [path]
@instantiate
class git_add_tree(GitCommand):
git_cmd = ['git', 'add', '-u', ':/']
@instantiate
class git_status(GitCommand):
git_cmd = ['git', 'status', '--porcelain']
def result(self, rv):
return Status(rv.decode('utf-8').splitlines())
class GitCommit(GitCommand):
git_cmd = ['git', 'commit']
def git_args(self, message, author):
author = shlex.quote(author)
return ['-m', message, '--author="%s"'%(author)]
@instantiate
class git_commit_index(GitCommit):
pass
@instantiate
class git_commit_tree(GitCommit):
git_cmd = ['git', 'commit']
def check_before(self):
return GitAddTree()
#@instantiate
#class GitCheckUpstream(GitCommand):
# """
# git fetch origin master
# git merge-tree `git merge-base FETCH_HEAD master` FETCH_HEAD master
# """
# branch = 'master'
# origin = 'origin'
# class GitMergeBase(GitCommand):
# git_cmd = ['git', 'merge-base', 'FETCH_HEAD', branch]
# class GitMergeTree(GitCommand):
# git_cmd = ['git', 'merge-tree']
# def git_args(self, sha):
# sha = merge_base()
# return []
# fetch = GitFetch()
# merge_base = GitMergeBase()
# merge_tree = GitMergeTree()
# def __call__(self, *args):
# self.fetch()
# sha = self.merge_base()
# return self.merge_tree(sha)
# -*- coding: utf-8 -*-
import os
import shutil
import subprocess
from flask import Blueprint, \
current_app, \
render_template, \
url_for, \
Markup, \
has_app_context, \
has_request_context
from lektor.pluginsystem import Plugin
from lektor.publisher import Publisher
from lektor.pluginsystem import get_plugin
from lektor.admin.modules import serve, api
from git import git_add, git_status, git_commit_tree
gitbp = Blueprint('git', __name__,
url_prefix='/git',
static_folder='static',
template_folder='templates'
)
@gitbp.route('/status',methods=['GET'])
def status():
values = git_status()
return render_template("status.html", status = values)
#@gitbp.route('/add',methods=['POST'])
#def add():
# path = request.values.get('path', None)
# if path:
# rv = git_add(path)
# return "done"
# return "failed"
#
#@gitbp.route('/commit',methods=['POST'])
#def commit():
# #author = request.values.get(
# rv = git_commit()
# #subprocess.call(['git', 'commit', '--author="John Doe <john@doe.org>"'])
# return "failed"
class CommitPublisher(Publisher):
"""
commit all changes from current tree
get username from session cookie
"""
def publish(self, target_url, credentials=None, **extra):
rv = git_commit_tree()
yield rv
class AskReviewPublisher(CommitPublisher):
"""
commit changes and ask for review before push
"""
def send_mail(self, status):
#mail -s "Test" pascal.molin@math.univ-paris-diderot.fr < /dev/null
pass
def publish(self, target_url, credentials=None, **extra):
status = git_status()
yield '%d fichiers modifiés'
for s in status:
yield str(s)
self.send_mail(status)
yield 'Message envoyé'
yield 'Les changements seront publiés après validation.'
class GitPushPublisher(CommitPublisher):
"""
commit modified files and push
caveat: the publish method does not have access to the request
which triggered it, hence the publisher name.
As a tentative workaround, we let flask store the user associated
to the last valid /publish request
"""
def publish(self, target_url, credentials=None, **extra):
#src_path = self.output_path
#dst_path = target_url.path
#self.fail("Vous n'avez pas le droit de publier directement, \
# sélectionnez plutôt «signaler les changements pour publication»")
app = current_app
login = get_plugin('login', self.env)
with app.app_context():
user = login.get_auth_user()
if user is not None and user.get('publish',False):
msg = 'user %s'%user['username']
yield msg
yield 'Done'
self.fail('Please commit all changes before publishing')
class CommitPlugin(Plugin):
name = 'lektor-commit'
description = u'Publish by git commit+push, or ask review.'
def on_setup_env(self, *args, **extra):
self.env.add_publisher('askreview', AskReviewPublisher)
self.env.add_publisher('gitpush', GitPushPublisher)
@serve.bp.before_app_first_request
def setup_git_bp():
app = current_app
app.register_blueprint(gitbp)
login = get_plugin('login', self.env)
login.add_button( url_for('git.status'),
'see all changes',
Markup('<i class="fa fa-save"></i>'))
## use to add/commit after each use
#@api.bp.after_request
##pylint: disable=unused-variable
#def after_request_api(response):
# # add modification to user cookie
# # can use request
# return response
......@@ -114,79 +114,3 @@ def server(lektorproject, port):
yield server
print("[HALT LEKTOR SERVER]")
server.kill()
def login(client, name, **kwargs):
password = kwargs.pop('password',name)
url = kwargs.pop('url', '/admin/root/edit')
return client.post('/auth/login',
data = dict( username=name, password=password, url=url),
**kwargs)
def logout(client, **kwargs):
return client.get('/auth/logout', **kwargs)
@pytest.fixture(scope='function')
def anonymous(server):
session = BaseUrlSession(base_url=server.base_url)
yield session
session.close()
@pytest.fixture(scope='function')
def admin(server):
session = BaseUrlSession(base_url=server.base_url)
login(session, 'admin')
yield session
session.close()
@pytest.fixture(scope='function')
def blog(server):
session = BaseUrlSession(base_url=server.base_url)
login(session, 'blog')
yield session
session.close()
@pytest.fixture(scope='function')
def test(server):
session = BaseUrlSession(base_url=server.base_url)
login(session, 'test')
yield session
session.close()
@pytest.fixture(scope='module')
def project(request, lektorproject):
from lektor.project import Project
return Project.from_path(lektorproject)
@pytest.fixture(scope='module')
def env(request, project):
env = project.make_env()
return env
@pytest.fixture(scope='function')
def pad(request, env):
from lektor.db import Database
return Database(env).new_pad()
@pytest.fixture(scope='module')
def webui(request, env):
from lektor.admin.webui import WebUI
output_path = tempfile.mkdtemp()
def cleanup():
try:
shutil.rmtree(output_path)
except (OSError, IOError):
pass
request.addfinalizer(cleanup)
return WebUI(env,
debug=True,
output_path=output_path)
@pytest.fixture(scope='module')
def webclient(webui):
webui.config['TESTING'] = True
with webui.test_client() as client:
yield client
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment