Redmine implements ICapBugTracker (refs #684)

This commit is contained in:
Romain Bignon 2011-07-11 22:29:34 +02:00
commit 311205543f
4 changed files with 528 additions and 2 deletions

View file

@ -21,7 +21,10 @@
from __future__ import with_statement
from weboob.capabilities.content import ICapContent, Content
from weboob.capabilities.bugtracker import ICapBugTracker, Issue, Project, User, Version, Status, Update, Attachment, Query
from weboob.capabilities.collection import ICapCollection, Collection, CollectionNotFound
from weboob.tools.backend import BaseBackend, BackendConfig
from weboob.tools.browser import BrowserHTTPNotFound
from weboob.tools.value import ValueBackendPassword, Value
from .browser import RedmineBrowser
@ -30,7 +33,7 @@ from .browser import RedmineBrowser
__all__ = ['RedmineBackend']
class RedmineBackend(BaseBackend, ICapContent):
class RedmineBackend(BaseBackend, ICapContent, ICapBugTracker, ICapCollection):
NAME = 'redmine'
MAINTAINER = 'Romain Bignon'
EMAIL = 'romain@weboob.org'
@ -47,6 +50,8 @@ class RedmineBackend(BaseBackend, ICapContent):
self.config['username'].get(),
self.config['password'].get())
############# CapContent ######################################################
def id2path(self, id):
return id.split('/', 2)
@ -85,3 +90,183 @@ class RedmineBackend(BaseBackend, ICapContent):
with self.browser:
return self.browser.get_wiki_preview(project, page, content.content)
############# CapCollection ###################################################
def iter_resources(self, path):
if len(path) == 0:
return [Collection(project.id) for project in self.iter_projects()]
if len(path) == 1:
query = Query()
query.project = unicode(path[0])
return self.iter_issues(query)
raise CollectionNotFound()
############# CapBugTracker ###################################################
def _build_project(self, project_dict):
project = Project(project_dict['name'], project_dict['name'])
project.members = [User(int(u[0]), u[1]) for u in project_dict['members']]
project.versions = [Version(int(v[0]), v[1]) for v in project_dict['versions']]
project.categories = [c[1] for c in project_dict['categories']]
# TODO set the value of status
project.statuses = [Status(int(s[0]), s[1], 0) for s in project_dict['statuses']]
return project
def iter_issues(self, query):
"""
Iter issues with optionnal patterns.
@param query [Query]
@return [iter(Issue)] issues
"""
# TODO link between text and IDs.
kwargs = {'subject': query.title,
'author_id': query.author,
'assigned_to_id': query.assignee,
'fixed_version_id': query.version,
'category_id': query.category,
'status_id': query.status,
}
r = self.browser.query_issues(query.project, **kwargs)
project = self._build_project(r['project'])
for issue in r['iter']:
obj = Issue(issue['id'])
obj.project = project
obj.title = issue['subject']
obj.creation = issue['created_on']
obj.updated = issue['updated_on']
if isinstance(issue['author'], tuple):
obj.author = project.find_user(*issue['author'])
else:
obj.author = User(0, issue['author'])
if isinstance(issue['assigned_to'], tuple):
obj.assignee = project.find_user(*issue['assigned_to'])
else:
obj.assignee = issue['assigned_to']
obj.category = issue['category']
if issue['fixed_version'] is not None:
obj.version = project.find_version(*issue['fixed_version'])
else:
obj.version = None
obj.status = project.find_status(issue['status'])
yield obj
def get_issue(self, issue):
if isinstance(issue, Issue):
id = issue.id
else:
id = issue
issue = Issue(issue)
try:
with self.browser:
params = self.browser.get_issue(id)
except BrowserHTTPNotFound:
return None
issue.project = self._build_project(params['project'])
issue.title = params['subject']
issue.body = params['body']
issue.creation = params['created_on']
issue.updated = params['updated_on']
issue.attachments = []
for a in params['attachments']:
attachment = Attachment(a['id'])
attachment.filename = a['filename']
attachment.url = a['url']
issue.attachments.append(attachment)
issue.history = []
for u in params['updates']:
update = Update(u['id'])
update.author = issue.project.find_user(*u['author'])
update.date = u['date']
update.message = u['message']
issue.history.append(update)
issue.author = issue.project.find_user(*params['author'])
issue.assignee = issue.project.find_user(*params['assignee'])
issue.category = params['category'][1]
issue.version = issue.project.find_version(*params['version'])
issue.status = issue.project.find_status(params['status'][1])
return issue
def create_issue(self, project):
try:
with self.browser:
r = self.browser.query_issues(project)
except BrowserHTTPNotFound:
return None
issue = Issue(0)
issue.project = self._build_project(r['project'])
return issue
def post_issue(self, issue):
project = issue.project.id
kwargs = {'title': issue.title,
'version': issue.version.id if issue.version else None,
'assignee': issue.assignee.id if issue.assignee else None,
'category': issue.category,
'status': issue.status.id if issue.status else None,
'body': issue.body,
}
with self.browser:
if int(issue.id) < 1:
id = self.browser.create_issue(project, **kwargs)
else:
id = self.browser.edit_issue(issue.id, **kwargs)
if id is None:
return None
issue.id = id
return issue
def update_issue(self, issue, update):
if isinstance(issue, Issue):
issue = issue.id
with self.browser:
return self.browser.update_issue(issue, update.message)
def remove_issue(self, issue):
"""
Remove an issue.
"""
if isinstance(issue, Issue):
issue = issue.id
with self.browser:
return self.browser.remove_issue(issue)
def iter_projects(self):
"""
Iter projects.
@return [iter(Project)] projects
"""
with self.browser:
for project in self.browser.iter_projects():
yield Project(project['id'], project['name'])
def get_project(self, id):
try:
with self.browser:
params = self.browser.get_issue(id)
except BrowserHTTPNotFound:
return None
return self._build_project(params['project'])
def fill_issue(self, issue, fields):
# currently there isn't cases where an Issue is uncompleted.
return issue
OBJECTS = {Issue: fill_issue}

View file

@ -24,8 +24,9 @@ import lxml.html
from weboob.tools.browser import BaseBrowser, BrowserIncorrectPassword
from .pages.index import LoginPage, IndexPage, MyPage
from .pages.index import LoginPage, IndexPage, MyPage, ProjectsPage
from .pages.wiki import WikiPage, WikiEditPage
from .pages.issues import IssuesPage, IssuePage, NewIssuePage
__all__ = ['RedmineBrowser']
@ -39,8 +40,12 @@ class RedmineBrowser(BaseBrowser):
# compatibility with redmine 0.9
'https?://[^/]+/login\?back_url.*': MyPage,
'https?://[^/]+/my/page': MyPage,
'https?://[^/]+/projects': ProjectsPage,
'https?://[^/]+/projects/([\w-]+)/wiki/([^\/]+)/edit': WikiEditPage,
'https?://[^/]+/projects/[\w-]+/wiki/[^\/]*': WikiPage,
'https?://[^/]+/projects/[\w-]+/issues/new': NewIssuePage,
'https?://[^/]+/issues(|/?\?.*)': IssuesPage,
'https?://[^/]+/issues/(\d+)': IssuePage,
}
def __init__(self, url, *args, **kwargs):
@ -52,6 +57,7 @@ class RedmineBrowser(BaseBrowser):
if self.BASEPATH.endswith('/'):
self.BASEPATH = self.BASEPATH[:-1]
BaseBrowser.__init__(self, *args, **kwargs)
self.projects = {}
def is_logged(self):
return self.is_on_page(LoginPage) or self.page and len(self.page.document.getroot().cssselect('a.my-account')) == 1
@ -100,3 +106,76 @@ class RedmineBrowser(BaseBrowser):
preview_html.find("legend").drop_tree()
return lxml.html.tostring(preview_html)
def query_issues(self, project_name, **kwargs):
data = (('project_id', project_name),
('query[column_names][]', 'tracker'),
('query[column_names][]', 'status'),
('query[column_names][]', 'priority'),
('query[column_names][]', 'subject'),
('query[column_names][]', 'assigned_to'),
('query[column_names][]', 'updated_on'),
('query[column_names][]', 'category'),
('query[column_names][]', 'fixed_version'),
('query[column_names][]', 'done_ratio'),
('query[column_names][]', 'author'),
('query[column_names][]', 'start_date'),
('query[column_names][]', 'due_date'),
('query[column_names][]', 'estimated_hours'),
('query[column_names][]', 'created_on'),
)
for key, value in kwargs.iteritems():
if value:
data += (('values[%s][]' % key, value),)
data += (('fields[]', key),)
data += (('operators[%s]' % key, '~'),)
self.location('/issues?set_filter=1&per_page=100', urllib.urlencode(data))
assert self.is_on_page(IssuesPage)
return {'project': self.page.get_project(project_name),
'iter': self.page.iter_issues(),
}
def get_issue(self, id):
self.location('/issues/%s' % id)
assert self.is_on_page(IssuePage)
return self.page.get_params()
def update_issue(self, id, message):
data = (('_method', 'put'),
('notes', message.encode('utf-8')),
)
self.openurl('/issues/%s/edit' % id, urllib.urlencode(data))
def create_issue(self, project, **kwargs):
self.location('/projects/%s/issues/new' % project)
assert self.is_on_page(NewIssuePage)
self.page.fill_form(**kwargs)
assert self.is_on_page(IssuePage)
return int(self.page.groups[0])
def edit_issue(self, id, **kwargs):
self.location('/issues/%s' % id)
assert self.is_on_page(IssuePage)
self.page.fill_form(**kwargs)
assert self.is_on_page(IssuePage)
return int(self.page.groups[0])
def remove_issue(self, id):
self.location('/issues/%s' % id)
assert self.is_on_page(IssuePage)
token = self.page.get_authenticity_token()
data = (('authenticity_token', token),)
self.openurl('/issues/%s/destroy' % id, urllib.urlencode(data))
def iter_projects(self):
self.location('/projects')
return self.page.iter_projects()

View file

@ -32,3 +32,13 @@ class IndexPage(BasePage):
class MyPage(BasePage):
pass
class ProjectsPage(BasePage):
def iter_projects(self):
for ul in self.parser.select(self.document.getroot(), 'ul.projects'):
for li in ul.findall('li'):
prj = {}
link = li.find('div').find('a')
prj['id'] = link.attrib['href'].split('/')[-1]
prj['name'] = link.text
yield prj

View file

@ -0,0 +1,252 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 Romain Bignon
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
import re
import datetime
from weboob.capabilities.bugtracker import IssueError
from weboob.tools.browser import BasePage, BrokenPageError
from weboob.tools.misc import to_unicode
from weboob.tools.mech import ClientForm
class BaseIssuePage(BasePage):
def parse_datetime(self, text):
m = re.match('(\d+)/(\d+)/(\d+) (\d+):(\d+) (\w+)', text)
if not m:
self.logger.warning('Unable to parse "%s"' % text)
return text
date = datetime.datetime(int(m.group(3)),
int(m.group(1)),
int(m.group(2)),
int(m.group(4)),
int(m.group(5)))
if m.group(6) == 'pm':
date += datetime.timedelta(0,12*3600)
return date
PROJECT_FIELDS = {'members': 'values_assigned_to_id',
'categories': 'values_category_id',
'versions': 'values_fixed_version_id',
'statuses': 'values_status_id',
}
def iter_choices(self, name):
try:
select = self.parser.select(self.document.getroot(), 'select#%s' % name, 1)
except BrokenPageError:
return
for option in select.findall('option'):
if option.attrib['value'].isdigit():
yield (option.attrib['value'], option.text)
def get_project(self, project_name):
project = {}
project['name'] = project_name
for field, elid in self.PROJECT_FIELDS.iteritems():
project[field] = list(self.iter_choices(elid))
return project
class IssuesPage(BaseIssuePage):
PROJECT_FIELDS = {'members': 'values_assigned_to_id',
'categories': 'values_category_id',
'versions': 'values_fixed_version_id',
'statuses': 'values_status_id',
}
def iter_issues(self):
try:
issues = self.parser.select(self.document.getroot(), 'table.issues', 1)
except BrokenPageError:
# No results.
return
for tr in issues.getiterator('tr'):
if not tr.attrib.get('id', '').startswith('issue-'):
continue
issue = {}
for td in tr.getiterator('td'):
field = td.attrib['class']
if field in ('checkbox','todo'):
continue
a = td.find('a')
if a is not None:
if a.attrib['href'].startswith('/users/') or \
a.attrib['href'].startswith('/versions/'):
text = (int(a.attrib['href'].split('/')[-1]), a.text)
else:
text = a.text
else:
text = td.text
if field.endswith('_on'):
text = self.parse_datetime(text)
elif field.endswith('_date') and text is not None:
m = re.match('(\d+)-(\d+)-(\d+)', text)
if m:
text = datetime.datetime(int(m.group(1)),
int(m.group(2)),
int(m.group(3)))
if isinstance(text, str):
text = to_unicode(text)
issue[field] = text
if len(issue) != 0:
yield issue
class NewIssuePage(BaseIssuePage):
PROJECT_FIELDS = {'members': 'issue_assigned_to_id',
'categories': 'issue_category_id',
'versions': 'issue_fixed_version_id',
'statuses': 'issue_status_id',
}
def set_title(self, title):
self.browser['issue[subject]'] = title.encode('utf-8')
def set_body(self, body):
self.browser['issue[description]'] = body.encode('utf-8')
def set_assignee(self, member):
if member:
self.browser['issue[assigned_to_id]'] = [str(member)]
else:
self.browser['issue[assigned_to_id]'] = ['']
def set_version(self, version):
try:
if version:
self.browser['issue[fixed_version_id]'] = [str(version)]
else:
self.browser['issue[fixed_version_id]'] = ['']
except ClientForm.ItemNotFoundError:
self.logger.warning('Version not found: %s' % version)
def set_category(self, category):
if category:
select = self.parser.select(self.document.getroot(), 'select#issue_category_id', 1)
for option in select.findall('option'):
if option.text and option.text.strip() == category:
self.browser['issue[category_id]'] = [option.attrib['value']]
return
self.logger.warning('Category "%s" not found' % category)
else:
self.browser['issue[category_id]'] = ['']
def set_status(self, status):
assert status is not None
self.browser['issue[status_id]'] = [str(status)]
def fill_form(self, **kwargs):
self.browser.select_form(predicate=lambda form: form.attrs.get('id', '') == 'issue-form')
for key, value in kwargs.iteritems():
if value is not None:
getattr(self, 'set_%s' % key)(value)
self.browser.submit()
class IssuePage(NewIssuePage):
def _parse_selection(self, id):
try:
select = self.parser.select(self.document.getroot(), 'select#%s' % id, 1)
except BrokenPageError:
# not available for this project
return ('', None)
else:
options = select.findall('option')
for option in options:
if 'selected' in option.attrib:
return (int(option.attrib['value']), to_unicode(option.text))
return ('', None)
def get_params(self):
params = {}
content = self.parser.select(self.document.getroot(), 'div#content', 1)
issue = self.parser.select(content, 'div.issue', 1)
params['project'] = self.get_project(to_unicode(self.parser.select(self.document.getroot(), 'h1', 1).text))
params['subject'] = to_unicode(self.parser.select(issue, 'div.subject', 1).find('div').find('h3').text.strip())
params['body'] = to_unicode(self.parser.select(self.document.getroot(), 'textarea#issue_description', 1).text)
author = self.parser.select(issue, 'p.author', 1)
# check issue 666 on symlink.me
i = 0
alist = author.findall('a')
if not 'title' in alist[i].attrib:
params['author'] = (int(alist[i].attrib['href'].split('/')[-1]),
to_unicode(alist[i].text))
i += 1
else:
params['author'] = (0, 'Anonymous')
params['created_on'] = self.parse_datetime(alist[i].attrib['title'])
if len(alist) > i+1:
params['updated_on'] = self.parse_datetime(alist[i+1].attrib['title'])
else:
params['updated_on'] = None
params['status'] = self._parse_selection('issue_status_id')
params['assignee'] = self._parse_selection('issue_assigned_to_id')
params['category'] = self._parse_selection('issue_category_id')
params['version'] = self._parse_selection('issue_fixed_version_id')
params['attachments'] = []
try:
for p in self.parser.select(content, 'div.attachments', 1).findall('p'):
attachment = {}
a = p.find('a')
attachment['id'] = int(a.attrib['href'].split('/')[-2])
attachment['filename'] = p.find('a').text
attachment['url'] = '%s://%s%s' % (self.browser.PROTOCOL, self.browser.DOMAIN, p.find('a').attrib['href'])
params['attachments'].append(attachment)
except BrokenPageError:
pass
params['updates'] = []
for div in self.parser.select(content, 'div.journal'):
update = {}
update['id'] = div.find('h4').find('div').find('a').text[1:]
alist = div.find('h4').findall('a')
update['author'] = (int(alist[-2].attrib['href'].split('/')[-1]),
to_unicode(alist[-2].text))
update['date'] = self.parse_datetime(alist[-1].attrib['title'])
if div.find('div') is not None:
comment = div.find('div')
subdiv = comment.find('div')
if subdiv is not None:
# a subdiv which contains changes is found, move the tail text
# of this div to comment text, and remove it.
comment.text = (comment.text or '') + (subdiv.tail or '')
comment.remove(comment.find('div'))
update['message'] = self.parser.tostring(comment).strip()
else:
update['message'] = None
params['updates'].append(update)
return params
def get_authenticity_token(self):
tokens = self.parser.select(self.document.getroot(), 'input[name=authenticity_token]')
if len(tokens) == 0:
raise IssueError("You doesn't have rights to remove this issue.")
token = tokens[0].attrib['value']
return token