Passed
Push — master ( 62d7d9...8230b1 )
by Alexander
03:29
created

JIRA.report_issue_from_testexecution()   B

Complexity

Conditions 5

Size

Total Lines 38
Code Lines 22

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 22
dl 0
loc 38
rs 8.8853
c 0
b 0
f 0
cc 5
nop 3
1
"""
2
    This module implements Kiwi TCMS interface to external issue tracking systems.
3
    Refer to each implementor class for integration specifics!
4
"""
5
6
import os
7
import tempfile
8
from urllib.parse import urlencode, quote
9
10
import jira
11
import github
12
import bugzilla
13
import gitlab
14
import redminelib
15
16
from django.conf import settings
17
18
from tcms.issuetracker.base import IssueTrackerType
19
from tcms.issuetracker.kiwitcms import KiwiTCMS  # noqa
20
from tcms.issuetracker import bugzilla_integration
21
from tcms.issuetracker import jira_integration
22
from tcms.issuetracker import github_integration
23
from tcms.issuetracker import gitlab_integration
24
from tcms.issuetracker import redmine_integration
25
26
27
def from_name(name):
28
    """
29
        Return the class which matches ``name`` if it exists inside this
30
        module or raise an exception.
31
    """
32
    if name not in globals():
33
        raise NotImplementedError('IT of type %s is not supported' % name)
34
    return globals()[name]
35
36
37
class Bugzilla(IssueTrackerType):
38
    """
39
        Support for Bugzilla. Requires:
40
41
        :api_url: - the XML-RPC URL for your Bugzilla instance
42
        :api_username: - a username registered in Bugzilla
43
        :api_password: - the password for this username
44
45
        You can also provide the ``BUGZILLA_AUTH_CACHE_DIR`` setting (in ``product.py``)
46
        to control where authentication cookies for Bugzilla will be saved. If this
47
        is not provided a temporary directory will be used each time we try to login
48
        into Bugzilla!
49
    """
50
51
    def __init__(self, bug_system):
52
        super().__init__(bug_system)
53
54
        # directory for Bugzilla credentials
55
        self._bugzilla_cache_dir = getattr(
56
            settings,
57
            "BUGZILLA_AUTH_CACHE_DIR",
58
            tempfile.mkdtemp(prefix='.bugzilla-')
59
        )
60
61
    def _rpc_connection(self):
62
        if not os.path.exists(self._bugzilla_cache_dir):
63
            os.makedirs(self._bugzilla_cache_dir, 0o700)
64
65
        return bugzilla.Bugzilla(
66
            self.bug_system.api_url,
67
            user=self.bug_system.api_username,
68
            password=self.bug_system.api_password,
69
            cookiefile=self._bugzilla_cache_dir + 'cookie',
70
            tokenfile=self._bugzilla_cache_dir + 'token',
71
        )
72
73
    def add_testexecution_to_issue(self, executions, issue_url):
74
        bug_id = self.bug_id_from_url(issue_url)
75
        for execution in executions:
76
            bugzilla_integration.BugzillaThread(self.rpc,
77
                                                self.bug_system,
78
                                                execution,
79
                                                bug_id).start()
80
81
    def report_issue_from_testexecution(self, execution, user):
82
        args = {}
83
        args['cf_build_id'] = execution.run.build.name
84
85
        args['comment'] = self._report_comment(execution)
86
        args['component'] = execution.case.component.values_list('name',
87
                                                                 flat=True)
88
        args['product'] = execution.run.plan.product.name
89
        args['short_desc'] = 'Test case failure: %s' % execution.case.summary
90
        args['version'] = execution.run.product_version
91
92
        url = self.bug_system.base_url
93
        if not url.endswith('/'):
94
            url += '/'
95
96
        return url + 'enter_bug.cgi?' + urlencode(args, True)
97
98
99
class JIRA(IssueTrackerType):
100
    """
101
        Support for JIRA. Requires:
102
103
        :api_url: - the API URL for your JIRA instance
104
        :api_username: - a username registered in JIRA
105
        :api_password: - the password for this username
106
107
        Additional control can be applied via the ``JIRA_OPTIONS`` configuration
108
        setting (in ``product.py``). By default this setting is not provided and
109
        the code uses ``jira.JIRA.DEFAULT_OPTIONS`` from the ``jira`` Python module!
110
    """
111
112
    def _rpc_connection(self):
113
        if hasattr(settings, 'JIRA_OPTIONS'):
114
            options = settings.JIRA_OPTIONS
115
        else:
116
            options = None
117
118
        return jira.JIRA(
119
            self.bug_system.api_url,
120
            basic_auth=(self.bug_system.api_username, self.bug_system.api_password),
121
            options=options,
122
        )
123
124
    @classmethod
125
    def bug_id_from_url(cls, url):
126
        """
127
            Jira IDs are the last group of chars at the end of the URL.
128
            For example https://issues.jenkins-ci.org/browse/JENKINS-31044
129
        """
130
        return url.strip().split('/')[-1]
131
132
    def add_testexecution_to_issue(self, executions, issue_url):
133
        bug_id = self.bug_id_from_url(issue_url)
134
        for execution in executions:
135
            jira_integration.JiraThread(self.rpc, self.bug_system, execution, bug_id).start()
136
137
    def report_issue_from_testexecution(self, execution, user):
138
        """
139
            For the HTML API description see:
140
            https://confluence.atlassian.com/display/JIRA050/Creating+Issues+via+direct+HTML+links
141
        """
142
        # note: your jira instance needs to have the same projects
143
        # defined otherwise this will fail!
144
        project = self.rpc.project(execution.run.plan.product.name)
145
146
        try:
147
            issue_type = self.rpc.issue_type_by_name('Bug')
148
        except KeyError:
149
            issue_type = self.rpc.issue_types()[0]
150
151
        args = {
152
            'pid': project.id,
153
            'issuetype': issue_type.id,
154
            'summary': 'Failed test: %s' % execution.case.summary,
155
        }
156
157
        try:
158
            # apparently JIRA can't search users via their e-mail so try to
159
            # search by username and hope that it matches
160
            tested_by = execution.tested_by
161
            if not tested_by:
162
                tested_by = execution.assignee
163
164
            args['reporter'] = self.rpc.user(tested_by.username).key
165
        except jira.JIRAError:
166
            pass
167
168
        args['description'] = self._report_comment(execution)
169
170
        url = self.bug_system.base_url
171
        if not url.endswith('/'):
172
            url += '/'
173
174
        return url + '/secure/CreateIssueDetails!init.jspa?' + urlencode(args, True)
175
176
177
class GitHub(IssueTrackerType):
178
    """
179
        Support for GitHub. Requires:
180
181
        :base_url: - URL to a GitHub repository for which we're going to report issues
182
        :api_password: - GitHub API token.
183
184
        .. note::
185
186
            You can leave the ``api_url`` and ``api_username`` fields blank because
187
            the integration code doesn't use them!
188
    """
189
190
    def _rpc_connection(self):
191
        # NOTE: we use an access token so only the password field is required
192
        return github.Github(self.bug_system.api_password)
193
194
    def add_testexecution_to_issue(self, executions, issue_url):
195
        bug_id = self.bug_id_from_url(issue_url)
196
        for execution in executions:
197
            github_integration.GitHubThread(self.rpc, self.bug_system, execution, bug_id).start()
198
199
    def is_adding_testcase_to_issue_disabled(self):
200
        return not (self.bug_system.base_url and self.bug_system.api_password)
201
202 View Code Duplication
    def report_issue_from_testexecution(self, execution, user):
0 ignored issues
show
Duplication introduced by Mr. Senko
This code seems to be duplicated in your project.
Loading history...
203
        """
204
            GitHub only supports title and body parameters
205
        """
206
        args = {
207
            'title': 'Failed test: %s' % execution.case.summary,
208
            'body': self._report_comment(execution),
209
        }
210
211
        url = self.bug_system.base_url
212
        if not url.endswith('/'):
213
            url += '/'
214
215
        return url + '/issues/new?' + urlencode(args, True)
216
217
218
class Gitlab(IssueTrackerType):
219
    """
220
        Support for Gitlab. Requires:
221
222
        :base_url: URL to a GitLab repository for which we're going to report issues
223
        :api_url: URL to GitLab instance. Usually gitlab.com!
224
        :api_password: GitLab API token.
225
226
        .. note::
227
228
            You can leave ``api_username`` field blank because
229
            the integration code doesn't use it!
230
    """
231
232
    def _rpc_connection(self):
233
        # we use an access token so only the password field is required
234
        return gitlab.Gitlab(self.bug_system.api_url,
235
                             private_token=self.bug_system.api_password)
236
237
    def add_testexecution_to_issue(self, executions, issue_url):
238
        bug_id = self.bug_id_from_url(issue_url)
239
        for execution in executions:
240
            gitlab_integration.GitlabThread(self.rpc, self.bug_system, execution, bug_id).start()
241
242
    def is_adding_testcase_to_issue_disabled(self):
243
        return not (self.bug_system.api_url and self.bug_system.api_password)
244
245 View Code Duplication
    def report_issue_from_testexecution(self, execution, user):
0 ignored issues
show
Duplication introduced by Mr. Senko
This code seems to be duplicated in your project.
Loading history...
246
        args = {
247
            'issue[title]': 'Failed test: %s' % execution.case.summary,
248
            'issue[description]': self._report_comment(execution),
249
        }
250
251
        url = self.bug_system.base_url
252
        if not url.endswith('/'):
253
            url += '/'
254
255
        return url + '/issues/new?' + urlencode(args, True)
256
257
258
class Redmine(IssueTrackerType):
259
    """
260
        Support for Redmine. Requires:
261
262
        :api_url: - the API URL for your Redmine instance
263
        :api_username: - a username registered in Redmine
264
        :api_password: - the password for this username
265
    """
266
267
    def _rpc_connection(self):
268
        return redminelib.Redmine(
269
            self.bug_system.api_url,
270
            username=self.bug_system.api_username,
271
            password=self.bug_system.api_password
272
        )
273
274
    def add_testexecution_to_issue(self, executions, issue_url):
275
        bug_id = self.bug_id_from_url(issue_url)
276
        for execution in executions:
277
            redmine_integration.RedmineThread(self.rpc,
278
                                              self.bug_system,
279
                                              execution,
280
                                              bug_id).start()
281
282
    def find_project_by_name(self, name):
283
        """
284
            Return a Redmine project which matches the given product name.
285
286
            .. note::
287
288
                If there is no match then return the first project in Redmine.
289
        """
290
        try:
291
            return self.rpc.project.get(name)
292
        except redminelib.exceptions.ResourceNotFoundError:
293
            projects = self.rpc.project.all()
294
            return projects[0]
295
296
    @staticmethod
297
    def find_issue_type_by_name(project, name):
298
        """
299
            Return a Redmine tracker matching name ('Bug').
300
301
            .. note::
302
303
                If there is no match then return the first one!
304
        """
305
        for trk in project.trackers:
306
            if str(trk).lower() == name.lower():
307
                return trk
308
309
        return project.trackers[0]
310
311
    def report_issue_from_testexecution(self, execution, user):
312
        project = self.find_project_by_name(execution.run.plan.product.name)
313
314
        issue_type = self.find_issue_type_by_name(project, 'Bug')
315
316
        query = "issue[tracker_id]=" + str(issue_type.id)
317
        query += "&issue[subject]=" + quote('Failed test: %s' % execution.case.summary)
318
319
        comment = self._report_comment(execution)
320
        query += "&issue[description]=%s" % quote(comment)
321
322
        url = self.bug_system.base_url
323
        if not url.endswith('/'):
324
            url += '/'
325
326
        return url + '/projects/%s/issues/new?' % project.id + query
327