Homec4science
Diffusion Phabricator API scripts (feature/getmystuph) Browse

phab-api/getmystuph/backends/epfl/repo.py
868520afb475feature/getmystuph

Last Change
Raw File

repo.py
View Options

# -*- coding: utf-8 -*-
from bs4 import BeautifulSoup
import re
import copy
import logging
from ... import export
from ... import colored
from ...repo import Repo
from .tequila import TequilaGet
__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__)
class RepoEPFL(Repo):
'''
Description of a repostitory on {svn git}.epfl.ch
'''
_LIST_REPOS = '{root}/repository/my.go'
_MANAGE_REPO = '{root}/repository/manage.go?id={id}'
_PERMISSION_URL = '{root}/objectRole/list.go?objectId={id}'
_REPO_REGEX = '/polyrepo/private/repository/manage\.go\?id=([0-9]+)'
_PERMISSIONS = {'Reader': Repo.READ,
'Contributor': Repo.WRITE,
'Administrator': Repo.ADMIN}
_repo_list_cache = {}
def __init__(self, name, *args, **kwargs):
super().__init__(name, *args, **kwargs)
option = copy.copy(kwargs)
self._id = option.pop('id', None)
tequila_ctx = option.pop('tequila_ctx', None)
_logger.info(
"Getting a Tequilla context for user \'{0}\'".format(
colored(self._username, 'blue')))
self.__tequila_ctx = self._tequila_ctx(tequila_ctx=tequila_ctx,
**kwargs)
if self._id is None:
if self.repo_type in self._repo_list_cache:
cache = self._repo_list_cache[self.repo_type]
if name in cache:
self._id = cache[name]
_logger.debug('repo_id {0} for {1} was found in'
' repositories cache'.format(
self._id,
self._colored_name))
else:
_logger.debug('No repo_id provided for {0}'.format(
self._colored_name))
self.list_repositories(
tequila_ctx=self.__tequila_ctx)
cache = self._repo_list_cache[self.repo_type]
if name in cache:
self._id = cache[name]
_logger.debug('repo_id {0} for {1} was found in'
' repositories list'.format(
self._id,
self._colored_name))
if self._id is None:
_msg = 'The repo {0} was not found in' \
' your list of repositories'.format(
self._colored_name)
_logger.error(_msg)
raise RuntimeError(_msg)
@property
def permissions(self):
'''Get the group and user permissions on the repository'''
_logger.info('Retrieving list of permissions' +
' for repositories {0}'.format(self._colored_name))
_html_resp = self.__tequila_ctx.get(
self._PERMISSION_URL.format(root=self._ROOT_URL,
id=self._id))
_html_soup = BeautifulSoup(_html_resp.text, 'html.parser')
_anonymous_perm = _html_soup.find(
'input', {'id': 'anonymousAccess'}).has_attr('checked')
_logger.debug(' anonymous access: {0}'.format(_anonymous_perm))
_permissions = Repo.Permissions(self)
_permissions._anonymous = _anonymous_perm
_permissions_tmp = {
'groups': [],
'users': [],
}
_group_regex = re.compile('([US][0-9]+)')
_list_soup = _html_soup.find(
'form', {'name': 'lister'}).find_all('tr')
for _tr in _list_soup:
_tds = _tr.find_all('td')
if not _tds:
continue
_perm_txt = _tds[-2].text.strip()
_perm = self._PERMISSIONS[_perm_txt]
_id_td = _tds[-1]
_ug_id = _id_td.text
_is_group = _group_regex.match(_ug_id)
_name = ''
_color = 'blue'
if _is_group:
_perm_type = 'groups'
if _logger.getEffectiveLevel() == logging.DEBUG:
_name = self.directory.get_group_name(_ug_id)
_color = 'green'
else:
_perm_type = 'users'
if _logger.getEffectiveLevel() == logging.DEBUG:
_name = self.directory.get_user_name(_ug_id)
_permissions_tmp[_perm_type].append({'id': _ug_id, 'perm': _perm})
_logger.debug(' {0}: {1} [{2}] -> {3} [{4}]'.format(
_perm_type,
colored(_name, _color),
colored(_ug_id, attrs=['bold']),
_perm_txt.lower(),
_perm
))
_permissions._groups = _permissions_tmp['groups']
_permissions._users = _permissions_tmp['users']
return _permissions
@classmethod
def _tequila_ctx(cls, tequila_ctx=None, **kwargs):
if tequila_ctx is None:
return TequilaGet(
cls._LIST_REPOS.format(root=cls._ROOT_URL), **kwargs)
else:
return tequila_ctx
@classmethod
def list_repositories(cls, tequila_ctx=None, **kwargs):
_logger.info("Retrieving the list of repositories")
_repos = []
_extra_info = {}
_tequila_ctx = cls._tequila_ctx(tequila_ctx=tequila_ctx, **kwargs)
_html_resp = _tequila_ctx.get(
cls._LIST_REPOS.format(root=cls._ROOT_URL)
)
_html_soup = BeautifulSoup(_html_resp.text, 'html.parser')
_list_soup = _html_soup.find('tbody')
_id_regex = re.compile(cls._REPO_REGEX)
for _link in _list_soup.find_all('a'):
_repo = _link.get_text()
_repos.append(_repo)
_repo_link = _link.get('href')
_match = _id_regex.match(_repo_link)
if _match:
_id = _match.group(1)
_extra_info[_repo] = {'id': _id}
_logger.debug(" List of repositories:")
for repo in _repos:
_logger.debug(" {0} -- {1}".format(
Repo.color_name(repo),
colored(_extra_info[repo], attrs=['bold'])))
if cls.repo_type not in cls._repo_list_cache:
cls._repo_list_cache[cls.repo_type] = dict()
cache = cls._repo_list_cache[cls.repo_type]
for name, info in _extra_info.items():
if 'id' in info:
cache[name] = info['id']
return (_repos, _extra_info)
@export
class RepoGitEPFL(RepoEPFL):
_ROOT_URL = 'https://git.epfl.ch/polyrepo/private'
repo_type = 'git'
def __init__(self, name, *args, **kwargs):
super(RepoGitEPFL, self).__init__(name, *args, **kwargs)
self._url = 'https://{0}@git.epfl.ch/repo/{1}.git'.format(
self._username, name)
@export
class RepoSvnEPFL(RepoEPFL):
_ROOT_URL = 'https://svn.epfl.ch/polyrepo/private'
repo_type = 'svn'
def __init__(self, name, *args, **kwargs):
super(RepoSvnEPFL, self).__init__(name, *args, **kwargs)
self._url = 'https://{0}@svn.epfl.ch/svn/{1}'.format(
self._username, name)
c4science · Help