Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F80811848
repo.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Subscribers
None
File Metadata
Details
File Info
Storage
Attached
Created
Mon, Sep 2, 17:25
Size
7 KB
Mime Type
text/x-python
Expires
Wed, Sep 4, 17:25 (2 d)
Engine
blob
Format
Raw Data
Handle
20448215
Attached To
rPHAPI Phabricator API scripts
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
.
directory
.
get_user_name
(
_ug_id
)
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
):
_logger
.
info
(
"Retrieving the list of repositories"
)
_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
,
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
issubclass
(
cls
,
RepoGitEPFL
):
_type
=
'git'
elif
issubclass
(
cls
,
RepoSvnEPFL
):
_type
=
'svn'
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'
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
Log In to Comment