Page MenuHomec4science
Files F61848155

phabricator.py
No OneTemporary
Actions

File Metadata

Created
Thu, May 9, 08:36

phabricator.py
View Options

# -*- coding: utf-8 -*-
import logging
import copy
from ... import export
from ... import colored
from ... import dry_do
from ...repo import Repo
from ...utils import get_phabricator_instance
from .. import color_phid
__author__ = "Nicolas Richart"
__copyright__ = "Copyright (C) 2016, EPFL (Ecole Polytechnique Fédérale " \
"de Lausanne) - SCITAS (Scientific IT and Application " \
"Support)"
__credits__ = ["Nicolas Richart"]
__license__ = "BSD"
__version__ = "0.1"
__maintainer__ = "Nicolas Richart"
__email__ = "nicolas.richart@epfl.ch"
_logger = logging.getLogger(__name__)
@export
class PhabRepo(Repo):
def __init__(self, *args, host=None, username=None, token=None, **kwargs):
super().__init__(**kwargs, username=username)
options = copy.copy(kwargs)
self._repo_type = options.pop('repo_type', 'git')
# _create = options.pop('create', False)
self._phab = get_phabricator_instance(host=host,
username=username,
token=token)
_data = self._phab.diffusion.repository.search(
queryKey="all", constraints={"name": self._name})['data']
self._phab_id = None
if len(_data) > 0:
self._phab_id = _data[0]['phid']
_logger.debug('Repositories {0} has id {1}'.format(
self._colored_name,
self._phab_id))
else:
_logger.debug('Repositories {0} not in phabricator'.format(
self._colored_name))
def create(self):
if self._phab_id is not None:
_msg = 'The repository {0}:{1} already exists'.format(
self._colored_name, self._phab_id)
_logger.error(_msg)
raise RuntimeError(_msg)
if self._dry_run:
self._phab_id = "PHID-REPO-notarealrepo"
else:
_data = self._phab.diffusion.repository.edit(
transactions=[{'type': 'name',
'value': self._name},
{'type': 'vcs',
'value': self._repo_type}])
self._phab_id = _data['object']['phid']
_msg = 'Created repository {0} id {1}'.format(self._colored_name,
color_phid(self._phab_id))
_logger.info(_msg)
if self._dry_run:
dry_do(_msg)
def set_permissions(self, permissions):
_perms = {'edit': [],
'push': [],
'view': []}
_equivalent = {'edit': Repo.EDIT,
'view': Repo.VIEW,
'push': Repo.PUSH}
_phab_perms = {'edit': 'edit',
'view': 'view',
'push': 'policy.push'}
_special_perms = {'_author_': 'obj.repository.author',
'_users_': 'users',
'_public_': 'public'}
for _type in {'group', 'user'}:
_perms_ug = getattr(permissions, '{0}s'.format(_type))
print(_perms_ug)
for _entity in _perms_ug:
_id = _entity['id']
if _id in _special_perms:
_id = _special_perms[_id]
for _phab, _gen in _equivalent.items():
if _entity['perm'] & _gen:
_perms[_phab].append(_id)
if permissions.anonymous:
_perms['view'] = ['public']
for _type in ['push', 'view', 'edit']:
if _type not in _perms:
continue
_perms[_type] = set(_perms[_type])
if len(_perms[_type]) > 1:
# create custom policy to replace the list
_logger.warning(
'Cannot create complicated policies, {0}'.format(
_perms[_type]) +
' changing permissions to user instead')
_perms[_type] = ['obj.repository.author']
_msg = 'Setting \'{0}\' permissions for {1} to {2}:'.format(
_type, self._colored_name,
', '.join([color_phid(_id) for _id in _perms[_type]]))
_logger.info(_msg)
if not self._dry_run:
self._in_directory
self._phab.diffusion.repository.edit(
transactions=[{'type': _phab_perms[_type],
'value': self._perms[_type][0]}],
objectIdentifier=self._phab_id)
else:
dry_do(_msg)

Event Timeline

c4science · Help