Page MenuHomec4science
Files F63723176

repo.py
No OneTemporary
Actions

File Metadata

Created
Wed, May 22, 02:21

repo.py
View Options

# -*- coding: utf-8 -*-
from bs4 import BeautifulSoup
import re
import copy
import logging
from ... import export
from ... import colored
from ... import color_code
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.VIEW,
'Contributor': Repo.PUSH + Repo.VIEW,
'Administrator': Repo.EDIT + Repo.PUSH + Repo.VIEW}
_repo_list_cache = {}
def __init__(self, name, *args, **kwargs):
super().__init__(name, *args, **kwargs)
option = copy.copy(kwargs)
self._id = option.pop('id', None)
self.__tequila_ctx = self._tequila_ctx(**kwargs)
self._permissions = None
if self._id is None:
if self._type in self._repo_list_cache:
cache = self._repo_list_cache[self._type]
if name in cache:
self._id = cache[name]
_logger.debug('repo_id {0} for {1} was found in'
' repositories cache'.format(
colored(self._id, color_code['repo']),
self._colored_name))
else:
_logger.debug('No repo_id provided for {0}'.format(
self._colored_name))
self.list_repositories(
tequila_ctx=self.__tequila_ctx, list_perm=False)
cache = self._repo_list_cache[self._type]
if name in cache:
self._id = cache[name]
_logger.debug('repo_id {0} for {1} was found in'
' repositories list'.format(
colored(self._id, color_code['repo']),
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'''
if self._permissions is not None:
pass
self._permissions = Repo.Permissions(self)
_logger.info('Retrieving list of permissions' +
' for repositories {0}'.format(self._colored_name))
_html_resp = None
try:
_html_resp = self.__tequila_ctx.get(
self._PERMISSION_URL.format(root=self._ROOT_URL,
id=self._id))
except:
return self._permissions
_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))
self._permissions.anonymous = _anonymous_perm
_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 = ''
if _is_group:
_perm_type = 'group'
if _logger.getEffectiveLevel() == logging.DEBUG:
_name = self.directory.get_group_name(_ug_id)
else:
_perm_type = 'user'
if _logger.getEffectiveLevel() == logging.DEBUG:
_name = self._user_db.get_user_name(_ug_id, self.directory)
if _name is None and (_perm_type == 'user' and
self._user_db is not None and
self._user_db.is_id_in_cache(_ug_id)):
_name = _ug_id
_logger.warning('{0} {1} unknown'.format(
_perm_type,
self.directory.color_name(_ug_id)))
else:
getattr(
self._permissions,
'add_{0}'.format(_perm_type))(_ug_id, _perm)
_logger.debug(' {0}: {1} [{2}] -> {3} [{4}]'.format(
_perm_type,
self.directory.color_name(_name, type=_perm_type),
self.directory.color_name(_ug_id),
_perm_txt.lower(),
_perm))
return self._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, list_perm=False, **kwargs):
_type = cls._repo_type
_logger.info("Retrieving the {0} list of repositories".format(
_type))
_repos = []
_extra_info = {}
tequila_ctx = kwargs.pop('tequila_ctx', None)
_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}
if list_perm and ('directory' in kwargs):
_repo_epfl = Repo(name=_repo,
type=_type,
id=_id,
tequila_ctx=_tequila_ctx,
**kwargs)
_perms = _repo_epfl.permissions
_perm = _perms.user_perm(
kwargs['directory'].whoami
)
_extra_info[_repo]['perm'] = _perm
_logger.debug(' List of repositories:')
for _repo in _repos:
_logger.debug(
' [{1}] {0} - {2}'.format(
colored(_repo, color_code['repo'], attrs=['bold']),
colored('{:>5}'.format(_extra_info[_repo]['id']),
color_code['repo']),
_extra_info[_repo]))
if _type not in cls._repo_list_cache:
cls._repo_list_cache[_type] = dict()
cache = cls._repo_list_cache[_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().__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().__init__(name, *args, **kwargs)
self._url = 'https://{0}@svn.epfl.ch/svn/{1}'.format(
self._username, name)

Event Timeline

c4science · Help