tcms.issuetracker.types.Redmine._rpc_connection() - Code Metrics - Inspection of "Implement search bug page. Refs #699" - kiwitcms/Kiwi - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 62d7d9...8230b1 )

by Alexander

created 2019-09-19 18:27 UTC

tcms.issuetracker.types.Redmine._rpc_connection() A

↳ Parent: tcms.issuetracker.types

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	5
rs	10
c	0
b	0
f	0
cc	1
nop	1

"""
    This module implements Kiwi TCMS interface to external issue tracking systems.
    Refer to each implementor class for integration specifics!
"""

import os
import tempfile
from urllib.parse import urlencode, quote

import jira
import github
import bugzilla
import gitlab
import redminelib

from django.conf import settings

from tcms.issuetracker.base import IssueTrackerType
from tcms.issuetracker.kiwitcms import KiwiTCMS  # noqa
from tcms.issuetracker import bugzilla_integration
from tcms.issuetracker import jira_integration
from tcms.issuetracker import github_integration
from tcms.issuetracker import gitlab_integration
from tcms.issuetracker import redmine_integration


def from_name(name):
    """
        Return the class which matches ``name`` if it exists inside this
        module or raise an exception.
    """
    if name not in globals():
        raise NotImplementedError('IT of type %s is not supported' % name)
    return globals()[name]


class Bugzilla(IssueTrackerType):
    """
        Support for Bugzilla. Requires:

        :api_url: - the XML-RPC URL for your Bugzilla instance
        :api_username: - a username registered in Bugzilla
        :api_password: - the password for this username

        You can also provide the ``BUGZILLA_AUTH_CACHE_DIR`` setting (in ``product.py``)
        to control where authentication cookies for Bugzilla will be saved. If this
        is not provided a temporary directory will be used each time we try to login
        into Bugzilla!
    """

    def __init__(self, bug_system):
        super().__init__(bug_system)

        # directory for Bugzilla credentials
        self._bugzilla_cache_dir = getattr(
            settings,
            "BUGZILLA_AUTH_CACHE_DIR",
            tempfile.mkdtemp(prefix='.bugzilla-')
        )

    def _rpc_connection(self):
        if not os.path.exists(self._bugzilla_cache_dir):
            os.makedirs(self._bugzilla_cache_dir, 0o700)

        return bugzilla.Bugzilla(
            self.bug_system.api_url,
            user=self.bug_system.api_username,
            password=self.bug_system.api_password,
            cookiefile=self._bugzilla_cache_dir + 'cookie',
            tokenfile=self._bugzilla_cache_dir + 'token',
        )

    def add_testexecution_to_issue(self, executions, issue_url):
        bug_id = self.bug_id_from_url(issue_url)
        for execution in executions:
            bugzilla_integration.BugzillaThread(self.rpc,
                                                self.bug_system,
                                                execution,
                                                bug_id).start()

    def report_issue_from_testexecution(self, execution, user):
        args = {}
        args['cf_build_id'] = execution.run.build.name

        args['comment'] = self._report_comment(execution)
        args['component'] = execution.case.component.values_list('name',
                                                                 flat=True)
        args['product'] = execution.run.plan.product.name
        args['short_desc'] = 'Test case failure: %s' % execution.case.summary
        args['version'] = execution.run.product_version

        url = self.bug_system.base_url
        if not url.endswith('/'):
            url += '/'

        return url + 'enter_bug.cgi?' + urlencode(args, True)


class JIRA(IssueTrackerType):
    """
        Support for JIRA. Requires:

        :api_url: - the API URL for your JIRA instance
        :api_username: - a username registered in JIRA
        :api_password: - the password for this username

        Additional control can be applied via the ``JIRA_OPTIONS`` configuration
        setting (in ``product.py``). By default this setting is not provided and
        the code uses ``jira.JIRA.DEFAULT_OPTIONS`` from the ``jira`` Python module!
    """

    def _rpc_connection(self):
        if hasattr(settings, 'JIRA_OPTIONS'):
            options = settings.JIRA_OPTIONS
        else:
            options = None

        return jira.JIRA(
            self.bug_system.api_url,
            basic_auth=(self.bug_system.api_username, self.bug_system.api_password),
            options=options,
        )

    @classmethod
    def bug_id_from_url(cls, url):
        """
            Jira IDs are the last group of chars at the end of the URL.
            For example https://issues.jenkins-ci.org/browse/JENKINS-31044
        """
        return url.strip().split('/')[-1]

    def add_testexecution_to_issue(self, executions, issue_url):
        bug_id = self.bug_id_from_url(issue_url)
        for execution in executions:
            jira_integration.JiraThread(self.rpc, self.bug_system, execution, bug_id).start()

    def report_issue_from_testexecution(self, execution, user):
        """
            For the HTML API description see:
            https://confluence.atlassian.com/display/JIRA050/Creating+Issues+via+direct+HTML+links
        """
        # note: your jira instance needs to have the same projects
        # defined otherwise this will fail!
        project = self.rpc.project(execution.run.plan.product.name)

        try:
            issue_type = self.rpc.issue_type_by_name('Bug')
        except KeyError:
            issue_type = self.rpc.issue_types()[0]

        args = {
            'pid': project.id,
            'issuetype': issue_type.id,
            'summary': 'Failed test: %s' % execution.case.summary,
        }

        try:
            # apparently JIRA can't search users via their e-mail so try to
            # search by username and hope that it matches
            tested_by = execution.tested_by
            if not tested_by:
                tested_by = execution.assignee

            args['reporter'] = self.rpc.user(tested_by.username).key
        except jira.JIRAError:
            pass

        args['description'] = self._report_comment(execution)

        url = self.bug_system.base_url
        if not url.endswith('/'):
            url += '/'

        return url + '/secure/CreateIssueDetails!init.jspa?' + urlencode(args, True)


class GitHub(IssueTrackerType):
    """
        Support for GitHub. Requires:

        :base_url: - URL to a GitHub repository for which we're going to report issues
        :api_password: - GitHub API token.

        .. note::

            You can leave the ``api_url`` and ``api_username`` fields blank because
            the integration code doesn't use them!
    """

    def _rpc_connection(self):
        # NOTE: we use an access token so only the password field is required
        return github.Github(self.bug_system.api_password)

    def add_testexecution_to_issue(self, executions, issue_url):
        bug_id = self.bug_id_from_url(issue_url)
        for execution in executions:
            github_integration.GitHubThread(self.rpc, self.bug_system, execution, bug_id).start()

    def is_adding_testcase_to_issue_disabled(self):
        return not (self.bug_system.base_url and self.bug_system.api_password)

    def report_issue_from_testexecution(self, execution, user):

        """
            GitHub only supports title and body parameters
        """
        args = {
            'title': 'Failed test: %s' % execution.case.summary,
            'body': self._report_comment(execution),
        }

        url = self.bug_system.base_url
        if not url.endswith('/'):
            url += '/'

        return url + '/issues/new?' + urlencode(args, True)


class Gitlab(IssueTrackerType):
    """
        Support for Gitlab. Requires:

        :base_url: URL to a GitLab repository for which we're going to report issues
        :api_url: URL to GitLab instance. Usually gitlab.com!
        :api_password: GitLab API token.

        .. note::

            You can leave ``api_username`` field blank because
            the integration code doesn't use it!
    """

    def _rpc_connection(self):
        # we use an access token so only the password field is required
        return gitlab.Gitlab(self.bug_system.api_url,
                             private_token=self.bug_system.api_password)

    def add_testexecution_to_issue(self, executions, issue_url):
        bug_id = self.bug_id_from_url(issue_url)
        for execution in executions:
            gitlab_integration.GitlabThread(self.rpc, self.bug_system, execution, bug_id).start()

    def is_adding_testcase_to_issue_disabled(self):
        return not (self.bug_system.api_url and self.bug_system.api_password)

    def report_issue_from_testexecution(self, execution, user):

        args = {
            'issue[title]': 'Failed test: %s' % execution.case.summary,
            'issue[description]': self._report_comment(execution),
        }

        url = self.bug_system.base_url
        if not url.endswith('/'):
            url += '/'

        return url + '/issues/new?' + urlencode(args, True)


class Redmine(IssueTrackerType):
    """
        Support for Redmine. Requires:

        :api_url: - the API URL for your Redmine instance
        :api_username: - a username registered in Redmine
        :api_password: - the password for this username
    """

    def _rpc_connection(self):
        return redminelib.Redmine(
            self.bug_system.api_url,
            username=self.bug_system.api_username,
            password=self.bug_system.api_password
        )

    def add_testexecution_to_issue(self, executions, issue_url):
        bug_id = self.bug_id_from_url(issue_url)
        for execution in executions:
            redmine_integration.RedmineThread(self.rpc,
                                              self.bug_system,
                                              execution,
                                              bug_id).start()

    def find_project_by_name(self, name):
        """
            Return a Redmine project which matches the given product name.

            .. note::

                If there is no match then return the first project in Redmine.
        """
        try:
            return self.rpc.project.get(name)
        except redminelib.exceptions.ResourceNotFoundError:
            projects = self.rpc.project.all()
            return projects[0]

    @staticmethod
    def find_issue_type_by_name(project, name):
        """
            Return a Redmine tracker matching name ('Bug').

            .. note::

                If there is no match then return the first one!
        """
        for trk in project.trackers:
            if str(trk).lower() == name.lower():
                return trk

        return project.trackers[0]

    def report_issue_from_testexecution(self, execution, user):
        project = self.find_project_by_name(execution.run.plan.product.name)

        issue_type = self.find_issue_type_by_name(project, 'Bug')

        query = "issue[tracker_id]=" + str(issue_type.id)
        query += "&issue[subject]=" + quote('Failed test: %s' % execution.case.summary)

        comment = self._report_comment(execution)
        query += "&issue[description]=%s" % quote(comment)

        url = self.bug_system.base_url
        if not url.endswith('/'):
            url += '/'

        return url + '/projects/%s/issues/new?' % project.id + query


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 2019-09-19 17:34 UTC by Report Bug Copy Issue Report 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 2019-09-19 17:34 UTC by Report Bug Copy Issue Report 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

kiwitcms / Kiwi

Push — master ( 62d7d9...8230b1 )

tcms.issuetracker.types.Redmine._rpc_connection() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like